ADUANASOFT/plantillas-proyectos
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature/correccion-de-carga-csv-falta-de-datos-fk

Browse Source
This commit is contained in:
hreyes
2026-03-12 11:41:11 -06:00
parent a57b01b190
commit e1d4ef4a68
13 changed files with 852 additions and 345 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
backend/api/v1/modules/a76/items/exports/validators/calculations.py
View File

@@ -69,9 +69,10 @@ def apply_calculations(
line.depreciation_date = invoice_date
if (not line.description.description_spanish and not line.description.description_english) and (line.part_info.description_spanish and line.part_info.description_english):
line.description.description_spanish = line.part_info.description_spanish
line.description.description_english = line.part_info.description_english
part_info = getattr(line, "part_info", None)
if (not line.description.description_spanish and not line.description.description_english) and part_info and (part_info.description_spanish and part_info.description_english):
line.description.description_spanish = part_info.description_spanish
line.description.description_english = part_info.description_english
else:
if not line.description.description_spanish:
class_desc = (
@@ -184,10 +185,14 @@ def calculate_values(
# ==========================================
# CÁLCULOS DE VALORES EN MONEDA
# foreign=ME, local=MN, manual=MC
# Una sola fuente: currency_type (USD/ME, MXN/MN) cuando está presente, paridad con import y CSV.
# ==========================================
result = (
db.query(InvoiceFinancials.currency, InvoiceFinancials.exchange_rate)
db.query(
InvoiceFinancials.currency,
InvoiceFinancials.currency_type,
InvoiceFinancials.exchange_rate,
)
.filter(
InvoiceFinancials.invoice_id == line.invoice_id,
InvoiceFinancials.tenant_id == tenant_id,
@@ -198,23 +203,27 @@ def calculate_values(
if not result:
return
currency, exchange_rate = result
currency, currency_type, exchange_rate = result
if currency_type in ("USD", "ME"):
currency = "foreign"
elif currency_type in ("MXN", "MN"):
currency = "local"
if currency == "foreign": # ME
line.financial.unit_cost_usd = line.financial.unit_cost_capture
line.financial.value_usd = line.financial.unit_cost_usd * line.quantity.quantity
line.financial.unit_cost_mxn = line.financial.unit_cost_capture * exchange_rate
line.financial.unit_cost_mxn = line.financial.unit_cost_capture * (exchange_rate or 1)
line.financial.value_mxn = line.financial.unit_cost_mxn * line.quantity.quantity
line.financial.value_mc = line.financial.unit_cost_usd * line.quantity.quantity
elif currency == "local": # MN
line.financial.unit_cost_mxn = line.financial.unit_cost_capture
line.financial.value_mxn = line.financial.unit_cost_mxn * line.quantity.quantity
line.financial.unit_cost_usd = line.financial.unit_cost_capture / exchange_rate
line.financial.unit_cost_usd = line.financial.unit_cost_capture / (exchange_rate or 1)
line.financial.value_usd = line.financial.unit_cost_usd * line.quantity.quantity
line.financial.value_mc = line.financial.unit_cost_usd * line.quantity.quantity
elif currency == "manual": # MC
line.financial.unit_cost_usd = line.financial.unit_cost_capture / exchange_rate
line.financial.unit_cost_usd = line.financial.unit_cost_capture / (exchange_rate or 1)
line.financial.value_usd = line.financial.unit_cost_usd * line.quantity.quantity
line.financial.unit_cost_mxn = line.financial.unit_cost_usd * exchange_rate
line.financial.unit_cost_mxn = line.financial.unit_cost_usd * (exchange_rate or 1)
line.financial.value_mxn = line.financial.unit_cost_mxn * line.quantity.quantity
line.financial.value_mc = line.financial.unit_cost_capture * line.quantity.quantity

31
backend/api/v1/modules/a76/items/exports/validators/create.py
View File

@@ -17,6 +17,17 @@ from api.v1.modules.public.reference_data.payment_methods.models import PaymentM
from .common import validate_common
def _normalize_weight_type(logistics) -> str:
"""Paridad con CSV: enum o string a 'KGS'/'LBS' para comparaciones."""
if logistics is None:
return "KGS"
wt = getattr(logistics, "weight_type", None) or "KGS"
if hasattr(wt, "value"):
wt = wt.value
weight_str = str(wt).upper() if wt else "KGS"
return weight_str if weight_str in ("KGS", "LBS") else "KGS"
def validate_create(
db: Session,
line: LineItem,
@@ -212,16 +223,28 @@ def validate_create(
line.financial.value_temp_material_usd = line.financial.value_usd
line.financial.value_temp_material_mxn = line.financial.value_mxn
# value_mc: paridad con calculate_values y cargas CSV
currency = getattr(invoice.financials, "currency", None)
if currency == "foreign":
line.financial.value_mc = line.financial.value_usd
elif currency == "local":
line.financial.value_mc = line.financial.value_usd
elif currency == "manual":
line.financial.value_mc = unit_cost_capture * quantity
else:
line.financial.value_mc = line.financial.value_usd
# ==========================================
# VALIDAR Y CONVERTIR PESOS NETOS
# ==========================================
invoice_weight_type = invoice.logistics.weight_type # 'kgs' o 'lbs'
invoice_weight_type = _normalize_weight_type(invoice.logistics)
quantity = line.quantity.quantity or Decimal("0")
net_weight_input = line.quantity.net_weight or Decimal("0")
# Determinar si la unidad de medida es de peso
unit_is_kgs = line.unit_of_measure and line.unit_of_measure == "24" #KGS
unit_is_lbs = line.unit_of_measure and line.unit_of_measure == "25" #LBS
# UOM peso: aceptar 24/"24" y 25/"25" (paridad con CSV)
uom = line.unit_of_measure
unit_is_kgs = uom is not None and (str(uom) == "24" or uom == 24)
unit_is_lbs = uom is not None and (str(uom) == "25" or uom == 25)
# Calcular peso neto en kilogramos (estándar interno)
if unit_is_kgs:

103
backend/api/v1/modules/a76/items/imports/validators/calculations.py
View File

@@ -1,8 +1,7 @@
from sqlalchemy.orm import Session
from api.v1.modules.a76.invoices.models import InvoiceFinancials, InvoiceHeader, InvoiceLogistics
from api.v1.modules.a76.invoices.models import InvoiceFinancials, InvoiceHeader
from core.exceptions import ErrorCollector
from ...models import LineItem
from ...models import LineItem
from api.v1.modules.a76.classes.models import Class
@@ -13,57 +12,97 @@ def apply_calculations(
#TODO: SSisGen Logic
# if ssisgen.calcularcostounitarioenbaseavalortotalscaf = 1:
# unit_cost_capture = line.financial.total_value / line.financial.total_value <-- habria que revisar por que esta asi, por que para mi no tiene sentido, pero es lo que esta en clarion
caluclate_values(db, line, tenant_id, company_id)
if not line.fa_data.is_subitem:
line.fa_data.subitem_number = None
invoice_date = db.query(InvoiceHeader.invoice_date).filter(InvoiceHeader.id == line.invoice_id, InvoiceHeader.tenant_id == tenant_id, InvoiceHeader.company_id == company_id).scalar()
line.depreciation_date = invoice_date
if (not line.description.description_spanish and not line.description.description_english) and (line.part_info.description_spanish and line.part_info.description_english):
line.description.description_spanish = line.part_info.description_spanish
line.description.description_english = line.part_info.description_english
else:
if not line.description.description_spanish:
class_desc = (
db.query(Class.description_es, Class.description_en)
.filter(Class.id == line.class_id, Class.tenant_id == tenant_id, Class.company_id == company_id)
.first()
)
if class_desc:
line.description.description_spanish, line.description.description_english = class_desc
calculate_values(db, line, tenant_id, company_id)
apply_calculations_after_values(db, line, tenant_id, company_id, line_number)
def caluclate_values(
def apply_calculations_after_values(
db: Session, line: LineItem, tenant_id: int, company_id: int, line_number: int
):
"""
Aplica solo depreciation_date, descripción desde part/class y subitem_number.
Usado por el flujo CSV para no sobrescribir los valores ya calculados por currency_type.
"""
fa_data = getattr(line, "fa_data", None)
if fa_data is not None and not getattr(fa_data, "is_subitem", True):
fa_data.subitem_number = None
invoice_date = db.query(InvoiceHeader.invoice_date).filter(
InvoiceHeader.id == line.invoice_id,
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
).scalar()
if invoice_date is not None:
line.depreciation_date = invoice_date
if not getattr(line.description, "description_spanish", None) and not getattr(
line.description, "description_english", None
):
part_info = getattr(line, "part_info", None)
if part_info and getattr(part_info, "description_spanish", None) and getattr(part_info, "description_english", None):
line.description.description_spanish = part_info.description_spanish
line.description.description_english = part_info.description_english
else:
if not getattr(line.description, "description_spanish", None):
class_desc = (
db.query(Class.description_es, Class.description_en)
.filter(
Class.id == line.class_id,
Class.tenant_id == tenant_id,
Class.company_id == company_id,
)
.first()
)
if class_desc:
line.description.description_spanish, line.description.description_english = class_desc
def calculate_values(
db: Session, line: LineItem, tenant_id: int, company_id: int
):
"""
Una sola fuente de verdad: usa currency_type (USD/ME, MXN/MN) cuando está presente,
para paridad con create.py y cargas CSV. Si no hay currency_type, usa currency (foreign/local/manual).
"""
result = (
db.query(InvoiceFinancials.currency, InvoiceFinancials.exchange_rate)
.filter(InvoiceFinancials.invoice_id == line.invoice_id, InvoiceFinancials.tenant_id == tenant_id, InvoiceFinancials.company_id == company_id)
db.query(
InvoiceFinancials.currency,
InvoiceFinancials.currency_type,
InvoiceFinancials.exchange_rate,
)
.filter(
InvoiceFinancials.invoice_id == line.invoice_id,
InvoiceFinancials.tenant_id == tenant_id,
InvoiceFinancials.company_id == company_id,
)
.first()
)
if not result:
return
currency, exchange_rate = result
currency, currency_type, exchange_rate = result
# Prioridad: currency_type para alinear con create.py y CSV
if currency_type in ("USD", "ME"):
currency = "foreign"
elif currency_type in ("MXN", "MN"):
currency = "local"
# si currency_type es otro o None, se usa currency tal cual
if currency == "foreign":
line.financial.unit_cost_usd = line.financial.unit_cost_capture
line.financial.value_usd = line.financial.unit_cost_usd * line.quantity.quantity
line.financial.unit_cost_mxn = line.financial.unit_cost_capture * exchange_rate
line.financial.unit_cost_mxn = line.financial.unit_cost_capture * (exchange_rate or 1)
line.financial.value_mxn = line.financial.unit_cost_mxn * line.quantity.quantity
line.financial.value_mc = line.financial.unit_cost_usd * line.quantity.quantity
elif currency == "local":
line.financial.unit_cost_mxn = line.financial.unit_cost_capture
line.financial.value_mxn = line.financial.unit_cost_mxn * line.quantity.quantity
line.financial.unit_cost_usd = line.financial.unit_cost_capture / exchange_rate
line.financial.unit_cost_usd = line.financial.unit_cost_capture / (exchange_rate or 1)
line.financial.value_usd = line.financial.unit_cost_usd * line.quantity.quantity
line.financial.value_mc = line.financial.unit_cost_usd * line.quantity.quantity
elif currency == "manual":
line.financial.unit_cost_usd = line.financial.unit_cost_capture/exchange_rate
line.financial.unit_cost_usd = line.financial.unit_cost_capture / (exchange_rate or 1)
line.financial.value_usd = line.financial.unit_cost_usd * line.quantity.quantity
line.financial.unit_cost_mxn = line.financial.unit_cost_usd * exchange_rate
line.financial.unit_cost_mxn = line.financial.unit_cost_usd * (exchange_rate or 1)
line.financial.value_mxn = line.financial.unit_cost_mxn * line.quantity.quantity
line.financial.value_mc = line.financial.unit_cost_capture * line.quantity.quantity

31
backend/api/v1/modules/a76/items/imports/validators/create.py
View File

@@ -16,6 +16,17 @@ from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.models im
from .common import validate_common
def _normalize_weight_type(logistics) -> str:
"""Paridad con CSV: enum o string a 'KGS'/'LBS' para comparaciones."""
if logistics is None:
return "KGS"
wt = getattr(logistics, "weight_type", None) or "KGS"
if hasattr(wt, "value"):
wt = wt.value
weight_str = str(wt).upper() if wt else "KGS"
return weight_str if weight_str in ("KGS", "LBS") else "KGS"
def validate_create(
db: Session,
line: LineItem,
@@ -196,16 +207,28 @@ def validate_create(
line.financial.value_temp_material_usd = line.financial.value_usd
line.financial.value_temp_material_mxn = line.financial.value_mxn
# value_mc: paridad con calculate_values y cargas CSV
currency = getattr(invoice.financials, "currency", None)
if currency == "foreign":
line.financial.value_mc = line.financial.value_usd
elif currency == "local":
line.financial.value_mc = line.financial.value_usd
elif currency == "manual":
line.financial.value_mc = unit_cost_capture * quantity
else:
line.financial.value_mc = line.financial.value_usd
# ==========================================
# VALIDAR Y CONVERTIR PESOS NETOS
# ==========================================
invoice_weight_type = invoice.logistics.weight_type # 'kgs' o 'lbs'
invoice_weight_type = _normalize_weight_type(invoice.logistics)
quantity = line.quantity.quantity or Decimal("0")
net_weight_input = line.quantity.net_weight or Decimal("0")
# Determinar si la unidad de medida es de peso
unit_is_kgs = line.unit_of_measure and line.unit_of_measure == "24" #KGS
unit_is_lbs = line.unit_of_measure and line.unit_of_measure == "25" #LBS
# UOM peso: aceptar 24/"24" y 25/"25" (paridad con CSV)
uom = line.unit_of_measure
unit_is_kgs = uom is not None and (str(uom) == "24" or uom == 24)
unit_is_lbs = uom is not None and (str(uom) == "25" or uom == 25)
# Calcular peso neto en kilogramos (estándar interno)
if unit_is_kgs:

1
backend/api/v1/modules/a76/layouts_csv/classes/tasks.py
View File

@@ -2,6 +2,7 @@
Tareas Celery para importación CSV de Clases de Materiales.
Flujo: scan_file (validación) → insert_valid_rows (commit).
Usa layouts_csv.common (storage, normalize, csv_reader, meta, responses) y common.fk_loader, validators, mappers.
Sin ClassService de creación en API; los mappers CSV (row_to_class_data, row_to_class_data_merge_existing) son la fuente de verdad para reglas de negocio al crear/actualizar.
"""
import json
import logging

6
backend/api/v1/modules/a76/layouts_csv/exportacion/routes.py
View File

@@ -65,6 +65,12 @@ async def upload_import_file(
"tenant_id": tenant_id,
"company_id": company_id,
"user_id": current_user.get("id"),
"capture_user": (
current_user.get("preferred_username")
or current_user.get("email")
or current_user.get("sub")
or "CSV"
),
"footer_config": footer_config,
"operation_type": operation_type or "exp",
"template_id": template_id or default_template,

12
backend/api/v1/modules/a76/layouts_csv/facturas/line_item_enrichment/__init__.py Normal file
View File

@@ -0,0 +1,12 @@
"""
Enriquecimiento de partidas (line items) en la carga CSV de facturas.
Aplica las mismas reglas que el proceso manual (items/imports y items/exports validators)
para que los datos insertados por CSV coincidan con la API.
"""
from .import_enrichment import apply_import_defaults_and_calculations_for_csv
from .export_enrichment import apply_export_defaults_and_calculations_for_csv
__all__ = [
"apply_import_defaults_and_calculations_for_csv",
"apply_export_defaults_and_calculations_for_csv",
]

99
backend/api/v1/modules/a76/layouts_csv/facturas/line_item_enrichment/export_enrichment.py Normal file
View File

@@ -0,0 +1,99 @@
"""
Enriquecimiento de partidas de exportación cargadas por CSV.
Aplica las mismas reglas que items/exports/validators (create + calculations):
hereda de línea de importación y calcula valores por currency; defaults de export.
"""
from sqlalchemy.orm import Session, joinedload
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.classes.models import Class
from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a24.fa.fa_item_lines.dto import FaLineItemCreateDTO
from api.v1.modules.a76.items.schemas import LineItemCreate
from api.v1.modules.a76.items.exports.validators.calculations import (
calculate_values,
apply_calculations,
)
def _ensure_nested(line_data: LineItemCreate) -> None:
"""Asegura que existan los objetos anidados para calculate_values y apply_calculations."""
from api.v1.modules.a76.items.line_financials.schemas import LineFinancialCreate
from api.v1.modules.a76.items.line_quantities.schemas import LineQuantityCreate
from api.v1.modules.a76.items.line_customs.schemas import LineCustomCreate
from api.v1.modules.a76.items.line_descriptions.schemas import LineDescriptionCreate
if line_data.financial is None:
line_data.financial = LineFinancialCreate()
if line_data.quantity is None:
line_data.quantity = LineQuantityCreate()
if line_data.customs is None:
line_data.customs = LineCustomCreate()
if line_data.description is None:
line_data.description = LineDescriptionCreate()
if line_data.fa_data is None:
line_data.fa_data = FaLineItemCreateDTO(is_subitem=False, contains_subitems=False)
def apply_export_defaults_and_calculations_for_csv(
db: Session,
line_data: LineItemCreate,
tenant_id: int,
company_id: int,
line_number: int,
) -> bool:
"""
Aplica defaults y cálculos de partida de exportación según reglas del proceso manual
(copia desde línea de importación, currency, defaults has_fda_code, tax_payment, etc.).
Se invoca tras armar LineItemCreate desde el CSV.
Devuelve False si faltan factura/financials; True en caso contrario.
"""
_ensure_nested(line_data)
invoice: InvoiceHeader = (
db.query(InvoiceHeader)
.options(joinedload(InvoiceHeader.financials))
.filter(
InvoiceHeader.id == line_data.invoice_id,
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
)
.first()
)
if not invoice or not invoice.financials:
return False
part: Part | None = None
if line_data.part_number_id:
part = (
db.query(Part)
.filter(
Part.id == line_data.part_number_id,
Part.tenant_id == tenant_id,
Part.company_id == company_id,
)
.first()
)
setattr(line_data, "part_info", part)
if not line_data.unit_of_measure and line_data.class_id:
class_info = (
db.query(Class)
.filter(
Class.id == line_data.class_id,
Class.tenant_id == tenant_id,
Class.company_id == company_id,
)
.first()
)
if class_info:
line_data.unit_of_measure = class_info.unit_of_measure
# Copia desde línea de importación y cálculos por currency (reglas manual)
calculate_values(db, line_data, tenant_id, company_id)
# Defaults: has_fda_code, tax_payment, payment_method, depreciation_date, descripciones
apply_calculations(db, line_data, tenant_id, company_id, line_number)
return True

246
backend/api/v1/modules/a76/layouts_csv/facturas/line_item_enrichment/import_enrichment.py Normal file
View File

@@ -0,0 +1,246 @@
"""
Enriquecimiento de partidas de importación cargadas por CSV.
Aplica las mismas reglas que items/imports/validators (create + calculations)
en base a currency_type y peso de la factura; no re-sobrescribe con currency
para evitar discrepancias con el proceso manual.
"""
from decimal import Decimal
from sqlalchemy.orm import Session, joinedload
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.classes.models import Class
from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.general_catalogs.packages.models import Package
from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.models import (
USTariffFraction,
)
from api.v1.modules.a24.fa.fa_item_lines.dto import FaLineItemCreateDTO
from api.v1.modules.a76.items.schemas import LineItemCreate
from api.v1.modules.a76.items.imports.validators.calculations import (
apply_calculations_after_values,
)
def _normalize_weight_type(invoice) -> str:
"""Obtiene weight_type de la factura como 'KGS' o 'LBS' (paridad con proceso manual)."""
wt = getattr(invoice.logistics, "weight_type", None) or "KGS"
if hasattr(wt, "value"):
wt = wt.value
weight_str = str(wt).upper() if wt else "KGS"
return weight_str if weight_str in ("KGS", "LBS") else "KGS"
def apply_import_defaults_and_calculations_for_csv(
db: Session,
line_data: LineItemCreate,
tenant_id: int,
company_id: int,
line_number: int,
) -> bool:
"""
Aplica defaults y cálculos de partida de importación según reglas del proceso manual
(currency_type, peso, descripciones). Se invoca tras armar LineItemCreate desde el CSV.
Devuelve False si faltan factura/financials/logistics; True en caso contrario.
"""
invoice: InvoiceHeader = (
db.query(InvoiceHeader)
.options(
joinedload(InvoiceHeader.financials),
joinedload(InvoiceHeader.logistics),
)
.filter(
InvoiceHeader.id == line_data.invoice_id,
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
)
.first()
)
if not invoice or not invoice.financials or not invoice.logistics:
return False
class_info: Class | None = None
if line_data.class_id:
class_info = (
db.query(Class)
.filter(
Class.id == line_data.class_id,
Class.tenant_id == tenant_id,
Class.company_id == company_id,
)
.first()
)
part: Part | None = None
if line_data.part_number_id:
part = (
db.query(Part)
.filter(
Part.id == line_data.part_number_id,
Part.tenant_id == tenant_id,
Part.company_id == company_id,
)
.first()
)
setattr(line_data, "part_info", part)
if not line_data.fa_data:
line_data.fa_data = FaLineItemCreateDTO(
is_subitem=False,
contains_subitems=False,
)
exchange_rate = invoice.financials.exchange_rate or Decimal("1.0")
if not line_data.unit_of_measure and class_info:
line_data.unit_of_measure = class_info.unit_of_measure
# Reglas de moneda igual que proceso manual (create.py): currency_type
currency_type = getattr(invoice.financials, "currency_type", None) or "USD"
unit_cost_capture = line_data.financial.unit_cost_capture or Decimal("0")
if currency_type in ("USD", "ME"):
line_data.financial.unit_cost_capture = unit_cost_capture
line_data.financial.unit_cost_usd = unit_cost_capture
line_data.financial.unit_cost_mxn = unit_cost_capture * exchange_rate
elif currency_type in ("MXN", "MN"):
line_data.financial.unit_cost_capture = unit_cost_capture
line_data.financial.unit_cost_usd = (
unit_cost_capture / exchange_rate if exchange_rate else Decimal("0")
)
line_data.financial.unit_cost_mxn = unit_cost_capture
else:
line_data.financial.unit_cost_capture = unit_cost_capture
line_data.financial.unit_cost_usd = unit_cost_capture
line_data.financial.unit_cost_mxn = unit_cost_capture * exchange_rate
quantity = line_data.quantity.quantity or Decimal("0")
if line_data.financial.unit_cost_usd is not None:
line_data.financial.value_usd = line_data.financial.unit_cost_usd * quantity
if line_data.financial.unit_cost_mxn is not None:
line_data.financial.value_mxn = line_data.financial.unit_cost_mxn * quantity
line_data.financial.customs_value_usd = line_data.financial.value_usd
line_data.financial.customs_value_mxn = line_data.financial.value_mxn
line_data.financial.value_temp_material_usd = line_data.financial.value_usd
line_data.financial.value_temp_material_mxn = line_data.financial.value_mxn
# value_mc según currency de la factura (paridad con calculate_values manual)
currency = getattr(invoice.financials, "currency", None)
if currency == "foreign":
line_data.financial.value_mc = line_data.financial.value_usd
elif currency == "local":
line_data.financial.value_mc = line_data.financial.value_usd
elif currency == "manual":
line_data.financial.value_mc = unit_cost_capture * quantity
else:
line_data.financial.value_mc = line_data.financial.value_usd
# Peso: misma lógica que proceso manual (create.py) con weight_type normalizado
weight_str = _normalize_weight_type(invoice)
quantity = line_data.quantity.quantity or Decimal("0")
net_weight_input = line_data.quantity.net_weight or Decimal("0")
uom = line_data.unit_of_measure
unit_is_kgs = uom is not None and (str(uom) == "24" or uom == 24)
unit_is_lbs = uom is not None and (str(uom) == "25" or uom == 25)
if unit_is_kgs:
if weight_str == "KGS":
line_data.quantity.net_weight = quantity
else:
line_data.quantity.net_weight = quantity * Decimal("2.204624")
elif unit_is_lbs:
if weight_str == "KGS":
line_data.quantity.net_weight = quantity / Decimal("2.204624")
else:
line_data.quantity.net_weight = quantity
else:
if weight_str == "KGS":
line_data.quantity.net_weight = net_weight_input
else:
line_data.quantity.net_weight = net_weight_input / Decimal("2.204624")
gross_weight_input = line_data.quantity.gross_weight
package_quantity = line_data.quantity.package_quantity or 0
package_weight_unit = Decimal("0")
if line_data.quantity.package_id:
package = (
db.query(Package)
.filter(
Package.id == line_data.quantity.package_id,
Package.tenant_id == tenant_id,
Package.company_id == company_id,
)
.first()
)
if package and package.weight_unit:
package_weight_unit = package.weight_unit
if not gross_weight_input or gross_weight_input == 0:
if weight_str == "KGS":
line_data.quantity.gross_weight = line_data.quantity.net_weight + (
package_weight_unit * package_quantity
)
else:
line_data.quantity.gross_weight = line_data.quantity.net_weight + (
(package_weight_unit * Decimal("2.204624")) * package_quantity
)
else:
if weight_str == "KGS":
line_data.quantity.gross_weight = gross_weight_input
else:
line_data.quantity.gross_weight = gross_weight_input / Decimal("2.204624")
if line_data.quantity.gross_weight < line_data.quantity.net_weight:
line_data.quantity.gross_weight = line_data.quantity.net_weight + (
package_weight_unit * package_quantity
)
if package_quantity and package_quantity > 0 and line_data.quantity.package_id:
package = (
db.query(Package)
.filter(
Package.id == line_data.quantity.package_id,
Package.tenant_id == tenant_id,
Package.company_id == company_id,
)
.first()
)
if package:
line_data.description.package_description = package.description_es
else:
line_data.quantity.package_quantity = 0
line_data.quantity.package_id = None
line_data.description.package_description = None
if not line_data.customs.american_fraction and class_info and class_info.us_fraction:
line_data.customs.american_fraction = class_info.us_fraction
if line_data.customs.american_fraction:
us_fraction = (
db.query(USTariffFraction)
.filter(
USTariffFraction.code == line_data.customs.american_fraction,
USTariffFraction.tenant_id == tenant_id,
USTariffFraction.company_id == company_id,
)
.first()
)
if us_fraction:
if getattr(us_fraction, "type_code", None) == "foreign":
line_data.customs.advalorem_american = us_fraction.fixed_cost
else:
line_data.customs.advalorem_american = us_fraction.ad_valorem
if not line_data.description.description_spanish and class_info:
line_data.description.description_spanish = class_info.description_es
if not line_data.description.description_english and class_info:
line_data.description.description_english = class_info.description_en
if line_data.description.brand:
line_data.description.brand = line_data.description.brand.upper().strip()
if line_data.description.model:
line_data.description.model = line_data.description.model.upper().strip()
# Solo depreciation_date, descripción part/class y subitem_number (sin re-calcular moneda)
apply_calculations_after_values(db, line_data, tenant_id, company_id, line_number)
return True

362
backend/api/v1/modules/a76/layouts_csv/facturas/routes.py
View File

@@ -1,178 +1,184 @@
from datetime import datetime
from uuid import uuid4
import base64
import os
import json
import logging
from fastapi import APIRouter, UploadFile, File, Form, HTTPException, Depends, Query
from sqlalchemy.orm import Session
from typing import Optional, Literal, Dict, Any
from core.celery_app import celery_app
from core.config import settings
from core.database import get_core_db
from core.paths import layout_path
from core.security import get_current_user, validate_access_to_resource
from .tasks import (
scan_file,
insert_valid_rows,
IMPORT_FILE_KEY_PREFIX,
IMPORT_META_KEY_PREFIX,
IMPORT_REDIS_TTL,
)
from .schemas import ImportJobResponse, ImportJobStatus, CommitRequest
router = APIRouter()
logger = logging.getLogger(__name__)
def _get_redis():
"""Redis client (same broker as Celery so worker can read)."""
import redis
url = os.getenv("VALKEY_URL", os.getenv("REDIS_URL", "redis://valkey:6379/0"))
return redis.Redis.from_url(url, decode_responses=False)
@router.post("/upload/{model_target}", response_model=ImportJobResponse)
async def upload_import_file(
model_target: Literal["invoice_header", "invoice_details", "invoice_series"],
file: UploadFile = File(...),
footer_config: Optional[str] = Form(None), # JSON string with settings
template_id: Optional[str] = Form(None), # id de la plantilla (ej. imp_temp_header) para respetar columnas
company_id: int = Query(..., description="Company ID"), # Required for context
operation_type: Optional[str] = Query("imp"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""
Step 1: Upload CSV, save to temp, trigger scan task.
Si se envía template_id, solo se leen las columnas de esa plantilla.
"""
# 1. Validate Access & Get Tenant
try:
tenant_id = validate_access_to_resource(db, company_id, current_user)
except Exception as e:
logger.error(f"Access validation failed: {e}")
raise HTTPException(status_code=403, detail="Invalid company access")
if not file.filename.endswith(".csv"):
raise HTTPException(status_code=400, detail="Only .csv files allowed")
job_id = str(uuid4())
contents = await file.read()
meta_data = {
"tenant_id": tenant_id,
"company_id": company_id,
"user_id": current_user.get("id"),
"footer_config": footer_config,
"operation_type": operation_type,
"template_id": template_id,
}
# Store file and meta in Redis so the Celery worker can read them (no shared filesystem needed)
try:
redis_client = _get_redis()
redis_client.set(
f"{IMPORT_FILE_KEY_PREFIX}{job_id}",
base64.b64encode(contents),
ex=IMPORT_REDIS_TTL,
)
redis_client.set(
f"{IMPORT_META_KEY_PREFIX}{job_id}",
json.dumps(meta_data).encode("utf-8"),
ex=IMPORT_REDIS_TTL,
)
except Exception as e:
logger.error(f"Redis store error: {e}")
raise HTTPException(status_code=500, detail="Failed to queue file for processing.")
# Optional: also write to local disk (e.g. for same-machine worker or debugging)
try:
upload_dir = layout_path("imports", "temp")
os.makedirs(upload_dir, exist_ok=True)
file_path = os.path.join(upload_dir, f"{job_id}.csv")
meta_path = os.path.join(upload_dir, f"{job_id}.meta.json")
with open(file_path, "wb") as f:
f.write(contents)
with open(meta_path, "w") as f:
json.dump(meta_data, f)
except Exception as e:
logger.warning(f"Local file save failed (worker will use Redis): {e}")
# Trigger Celery Task (Async). Worker loads file from Redis.
scan_file.apply_async(args=[job_id, model_target, footer_config], task_id=job_id)
return ImportJobResponse(
job_id=job_id,
status="queued",
message="File uploaded. Scanning started."
)
@router.get("/{job_id}/status")
async def get_import_status(job_id: str):
"""
Poll to get progress or final report. Always returns an object with "status".
"""
task_result = celery_app.AsyncResult(job_id)
if task_result.state == "PENDING":
return {"status": "processing", "progress": 0}
if task_result.state == "PROGRESS":
return {
"status": "processing",
"progress": (task_result.info or {}).get("current", 0),
"total": (task_result.info or {}).get("total", 0),
}
if task_result.state == "SUCCESS":
result = task_result.result
if isinstance(result, dict) and "status" in result:
return result
return {"status": "finished", "result": result}
# FAILURE: obtener mensaje real (traceback, result o get(propagate=False))
logger.warning("Import task %s failed: state=%s", job_id, task_result.state)
err_msg = None
tb = getattr(task_result, "traceback", None)
if tb:
logger.debug("Task traceback: %s", tb[:500] if isinstance(tb, str) else tb)
if tb and isinstance(tb, str):
lines = [l.strip() for l in tb.strip().split("\n") if l.strip()]
if lines:
err_msg = lines[-1]
if not err_msg and len(lines) > 1:
err_msg = lines[-2] + " " + (lines[-1] or "")
if not err_msg:
try:
exc = task_result.get(propagate=False)
if exc is not None:
err_msg = str(exc)
except Exception:
pass
if not err_msg:
result = getattr(task_result, "result", None)
info = getattr(task_result, "info", None)
if result is not None and not isinstance(result, dict):
err_msg = str(result)
elif isinstance(result, dict) and (result.get("error") or result.get("message")):
err_msg = result.get("error") or result.get("message")
if not err_msg and isinstance(info, str):
err_msg = info
elif not err_msg and isinstance(info, dict) and "error" in info:
err_msg = str(info["error"])
if not err_msg:
err_msg = "Task failed"
return {"status": "failed", "error": err_msg}
@router.post("/{job_id}/commit")
async def commit_import_job(job_id: str, body: CommitRequest):
"""
Step 2: User confirms import. Trigger bulk insert.
"""
task = insert_valid_rows.delay(job_id, body.model_target)
return {
"status": "committing",
"message": "Bulk insert started.",
"commit_job_id": task.id
}
from datetime import datetime
from uuid import uuid4
import base64
import os
import json
import logging
from fastapi import APIRouter, UploadFile, File, Form, HTTPException, Depends, Query
from sqlalchemy.orm import Session
from typing import Optional, Literal, Dict, Any
from core.celery_app import celery_app
from core.config import settings
from core.database import get_core_db
from core.paths import layout_path
from core.security import get_current_user, validate_access_to_resource
from .tasks import (
scan_file,
insert_valid_rows,
IMPORT_FILE_KEY_PREFIX,
IMPORT_META_KEY_PREFIX,
IMPORT_REDIS_TTL,
)
from .schemas import ImportJobResponse, ImportJobStatus, CommitRequest
router = APIRouter()
logger = logging.getLogger(__name__)
def _get_redis():
"""Redis client (same broker as Celery so worker can read)."""
import redis
url = os.getenv("VALKEY_URL", os.getenv("REDIS_URL", "redis://valkey:6379/0"))
return redis.Redis.from_url(url, decode_responses=False)
@router.post("/upload/{model_target}", response_model=ImportJobResponse)
async def upload_import_file(
model_target: Literal["invoice_header", "invoice_details", "invoice_series"],
file: UploadFile = File(...),
footer_config: Optional[str] = Form(None), # JSON string with settings
template_id: Optional[str] = Form(None), # id de la plantilla (ej. imp_temp_header) para respetar columnas
company_id: int = Query(..., description="Company ID"), # Required for context
operation_type: Optional[str] = Query("imp"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""
Step 1: Upload CSV, save to temp, trigger scan task.
Si se envía template_id, solo se leen las columnas de esa plantilla.
"""
# 1. Validate Access & Get Tenant
try:
tenant_id = validate_access_to_resource(db, company_id, current_user)
except Exception as e:
logger.error(f"Access validation failed: {e}")
raise HTTPException(status_code=403, detail="Invalid company access")
if not file.filename.endswith(".csv"):
raise HTTPException(status_code=400, detail="Only .csv files allowed")
job_id = str(uuid4())
contents = await file.read()
meta_data = {
"tenant_id": tenant_id,
"company_id": company_id,
"user_id": current_user.get("id"),
"capture_user": (
current_user.get("preferred_username")
or current_user.get("email")
or current_user.get("sub")
or "CSV"
),
"footer_config": footer_config,
"operation_type": operation_type,
"template_id": template_id,
}
# Store file and meta in Redis so the Celery worker can read them (no shared filesystem needed)
try:
redis_client = _get_redis()
redis_client.set(
f"{IMPORT_FILE_KEY_PREFIX}{job_id}",
base64.b64encode(contents),
ex=IMPORT_REDIS_TTL,
)
redis_client.set(
f"{IMPORT_META_KEY_PREFIX}{job_id}",
json.dumps(meta_data).encode("utf-8"),
ex=IMPORT_REDIS_TTL,
)
except Exception as e:
logger.error(f"Redis store error: {e}")
raise HTTPException(status_code=500, detail="Failed to queue file for processing.")
# Optional: also write to local disk (e.g. for same-machine worker or debugging)
try:
upload_dir = layout_path("imports", "temp")
os.makedirs(upload_dir, exist_ok=True)
file_path = os.path.join(upload_dir, f"{job_id}.csv")
meta_path = os.path.join(upload_dir, f"{job_id}.meta.json")
with open(file_path, "wb") as f:
f.write(contents)
with open(meta_path, "w") as f:
json.dump(meta_data, f)
except Exception as e:
logger.warning(f"Local file save failed (worker will use Redis): {e}")
# Trigger Celery Task (Async). Worker loads file from Redis.
scan_file.apply_async(args=[job_id, model_target, footer_config], task_id=job_id)
return ImportJobResponse(
job_id=job_id,
status="queued",
message="File uploaded. Scanning started."
)
@router.get("/{job_id}/status")
async def get_import_status(job_id: str):
"""
Poll to get progress or final report. Always returns an object with "status".
"""
task_result = celery_app.AsyncResult(job_id)
if task_result.state == "PENDING":
return {"status": "processing", "progress": 0}
if task_result.state == "PROGRESS":
return {
"status": "processing",
"progress": (task_result.info or {}).get("current", 0),
"total": (task_result.info or {}).get("total", 0),
}
if task_result.state == "SUCCESS":
result = task_result.result
if isinstance(result, dict) and "status" in result:
return result
return {"status": "finished", "result": result}
# FAILURE: obtener mensaje real (traceback, result o get(propagate=False))
logger.warning("Import task %s failed: state=%s", job_id, task_result.state)
err_msg = None
tb = getattr(task_result, "traceback", None)
if tb:
logger.debug("Task traceback: %s", tb[:500] if isinstance(tb, str) else tb)
if tb and isinstance(tb, str):
lines = [l.strip() for l in tb.strip().split("\n") if l.strip()]
if lines:
err_msg = lines[-1]
if not err_msg and len(lines) > 1:
err_msg = lines[-2] + " " + (lines[-1] or "")
if not err_msg:
try:
exc = task_result.get(propagate=False)
if exc is not None:
err_msg = str(exc)
except Exception:
pass
if not err_msg:
result = getattr(task_result, "result", None)
info = getattr(task_result, "info", None)
if result is not None and not isinstance(result, dict):
err_msg = str(result)
elif isinstance(result, dict) and (result.get("error") or result.get("message")):
err_msg = result.get("error") or result.get("message")
if not err_msg and isinstance(info, str):
err_msg = info
elif not err_msg and isinstance(info, dict) and "error" in info:
err_msg = str(info["error"])
if not err_msg:
err_msg = "Task failed"
return {"status": "failed", "error": err_msg}
@router.post("/{job_id}/commit")
async def commit_import_job(job_id: str, body: CommitRequest):
"""
Step 2: User confirms import. Trigger bulk insert.
"""
task = insert_valid_rows.delay(job_id, body.model_target)
return {
"status": "committing",
"message": "Bulk insert started.",
"commit_job_id": task.id
}

274
backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py
View File

@@ -1,3 +1,13 @@
"""
Tareas Celery para importación CSV de facturas (encabezados, partidas, series).
Flujo: scan_file (validación) → insert_valid_rows (commit).
Objetivo en BD (paridad con flujo normal): al terminar el commit, los datos deben quedar
igual que por UI/API: encabezados con capture_user/who_updated; partidas con costos,
pesos y descripciones calculados/heredados según items/imports/validators; series con
campos no presentes en CSV en null. No se modifican plantillas CSV; no se inventan
datos sin fuente (p. ej. LineReference solo si hay fuente explícita).
"""
import os
from datetime import datetime
from decimal import Decimal
@@ -4233,6 +4243,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
return {"status": "failed", "error": str(e)}
# --- Series de Importación Temporal: commit (INSERT/UPDATE item_line_series) ---
# Paridad CSV: campos no presentes en CSV se persisten como null; no se exigen campos que no están en la plantilla.
if use_series_flow:
try:
from api.v1.modules.a76.invoices.models import InvoiceHeader
@@ -4490,6 +4501,17 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
from api.v1.modules.a76.items.line_quantities.models import LineQuantity
from api.v1.modules.a76.items.line_customs.models import LineCustom
from api.v1.modules.a76.items.line_descriptions.models import LineDescription
from api.v1.modules.a76.items.schemas import LineItemCreate
from api.v1.modules.a76.items.line_financials.schemas import LineFinancialCreate
from api.v1.modules.a76.items.line_quantities.schemas import LineQuantityCreate
from api.v1.modules.a76.items.line_customs.schemas import LineCustomCreate
from api.v1.modules.a76.items.line_descriptions.schemas import LineDescriptionCreate
from api.v1.modules.a24.fa.fa_item_lines.dto import FaLineItemCreateDTO
from api.v1.modules.a76.layouts_csv.facturas.line_item_enrichment import (
apply_import_defaults_and_calculations_for_csv,
apply_export_defaults_and_calculations_for_csv,
)
from api.v1.modules.a76.items.service import ItemService
from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.classes.models import Class
from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
@@ -4887,12 +4909,18 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
remesa_val = (max_rem or 0) + 1
if existing_header:
# UPDATE existing header
# UPDATE existing header (paridad con InvoiceService.update)
header = existing_header
header.invoice_date = invoice_date
header.operation_type = op_type_value
header.is_updated = True # Mark as updated
header.updated_date = datetime.utcnow()
capture_user = meta.get("capture_user") or "CSV"
header.who_updated = capture_user
# Backfill capture_user if missing or generic (paridad con service)
if not header.capture_user or header.capture_user == "System":
if capture_user != "CSV":
header.capture_user = capture_user
header.document_type = (
None if inv_type_value == "MEX" else
resolve_public_code(
@@ -4918,7 +4946,8 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
# SQLAlchemy relationship assignment usually handles 1-to-1 updates correctly.
else:
# CREATE new header
# CREATE new header (paridad con InvoiceService.create: capture_user, who_updated)
capture_user = meta.get("capture_user") or "CSV"
header = InvoiceHeader(
invoice_number=invoice_number,
invoice_date=invoice_date,
@@ -4926,6 +4955,8 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
is_updated=False,
system="CSV",
capture_date=datetime.utcnow(),
capture_user=capture_user,
who_updated=capture_user,
invoice_type=inv_type_value,
document_type=(
None if inv_type_value == "MEX" else
@@ -5115,7 +5146,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
skipped_missing_invoice += 1
continue
# --- Partidas Exportación Definitiva: inserción real ---
# --- Partidas Exportación Definitiva: paridad con flujo normal (validators + ItemService) ---
if _template_id_insert == "exp_def_partidas":
part_num = (row_norm.get('NUM. PARTE') or row_norm.get('NUMPARTE') or row_norm.get('NUMERO PARTE') or row_norm.get('NUM PARTE') or '').strip()
part_id = part_cache.get(part_num) if part_num else None
@@ -5123,7 +5154,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
p = session.query(Part.id).filter(Part.part_number == part_num, Part.tenant_id == tenant_id, Part.company_id == company_id).first()
if p:
part_id = p.id
part_cache[part_num] = part_id
part_cache[part_num] = p.id
line_num_val = (row_norm.get('LINEA EXPO') or row_norm.get('LINEA EXPO.') or row_norm.get('RENGLON EXPO'))
line_num = parse_int(line_num_val) or (len(details_to_insert) + 1)
@@ -5153,83 +5184,83 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
session.query(InvoiceSalesDetails).filter(InvoiceSalesDetails.invoice_id == invoice_id).delete(synchronize_session=False)
cleared_invoices.add(invoice_id)
line = LineItem(
invoice_id=invoice_id,
line_number=line_num,
tenant_id=tenant_id,
company_id=company_id,
part_number_id=part_id,
unit_of_measure=uom_id,
order=(row_norm.get('ORDEN DE COMPRA') or row_norm.get('ORDENCOMPRA') or row_norm.get('ORDEN DE VENTA') or None),
tax_payment=(se_pago == 'SI'),
payment_method=forma_pago,
)
session.add(line)
session.flush()
# FaLineItem (a24 extension: subpartidas, descarga, factura impo ref)
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
fa_line = FaLineItem(
id=line.id,
tenant_id=tenant_id,
company_id=company_id,
search_invoice=factura_impo or None,
search_line=parse_int(linea_impo_val),
search_type=tipo_impo or None,
download=(descarga_val == 'SI'),
is_subitem=is_subitem,
contains_subitems=contains_subitems,
subitem_number=parse_int(linea_principal_val) if is_subitem else None,
)
session.add(fa_line)
price = parse_decimal(row_norm.get('COSTO UNITARIO') or row_norm.get('COSTOUNITARIO'))
qty = parse_decimal(row_norm.get('CANTIDAD EXPORTADA/DESCARGAR') or row_norm.get('CANTIDAD EXPORTADA') or row_norm.get('CANTIDAD'))
commercial_total = (price * qty) if price and qty else None
session.add(LineFinancial(
item_line_id=line.id,
unit_cost_capture=decimal_or_zero(price),
total_commercial_value=decimal_or_zero(commercial_total),
))
net_w = parse_decimal(row_norm.get('PESO NETO') or row_norm.get('PESONETO'))
gross_w = parse_decimal(row_norm.get('PESO BRUTO') or row_norm.get('PESOBRUTO'))
session.add(LineQuantity(
item_line_id=line.id,
quantity=decimal_or_zero(qty),
net_weight=decimal_or_zero(net_w),
gross_weight=decimal_or_zero(gross_w),
package_quantity=int_or_zero(row_norm.get('CANTIDAD BULTOS') or row_norm.get('CANTIDADBULTOS')),
package_id=package_id,
))
origin = (row_norm.get('PAIS ORIGEN') or row_norm.get('PAISORIGEN') or row_norm.get('PAIS') or '').strip()
fraction = (row_norm.get('FRACCION ARANCELARIA') or row_norm.get('FRACCIONARANCELARIA') or '').strip()
american_fraction = (row_norm.get('FRACCION AMERICANA') or row_norm.get('FRACCIONAMERICANA') or '').strip()
session.add(LineCustom(
item_line_id=line.id,
origin_country=origin or None,
fraction=fraction or None,
american_fraction=american_fraction or None,
))
extra_desc = (row_norm.get('DESCRIPCION EXTRA') or row_norm.get('DESCRIPCIONEXTRA') or '').strip()
additional_info = (row_norm.get('INFORMACION ADICIONAL') or row_norm.get('INFORMACIONADICIONAL') or '').strip()
lot = (row_norm.get('LOTE') or '').strip()
entry_number = (row_norm.get('NUMERO ENTRADA') or row_norm.get('NUM ENTRADA') or '').strip()
session.add(LineDescription(
item_line_id=line.id,
extra_description=extra_desc or None,
additional_info_spanish=additional_info or None,
lot=lot or None,
entry_number=entry_number or None,
))
order_compra = (row_norm.get('ORDEN DE COMPRA') or row_norm.get('ORDENCOMPRA') or row_norm.get('ORDEN DE VENTA') or None)
line_data = LineItemCreate(
invoice_id=invoice_id,
line_number=line_num,
part_number_id=part_id,
class_id=None,
unit_of_measure=uom_id,
order=order_compra,
tax_payment=(se_pago == 'SI'),
payment_method=forma_pago,
financial=LineFinancialCreate(
unit_cost_capture=decimal_or_zero(price),
total_commercial_value=decimal_or_zero(commercial_total),
),
quantity=LineQuantityCreate(
quantity=decimal_or_zero(qty),
net_weight=decimal_or_zero(net_w),
gross_weight=decimal_or_zero(gross_w),
package_quantity=int_or_zero(row_norm.get('CANTIDAD BULTOS') or row_norm.get('CANTIDADBULTOS')),
package_id=package_id,
),
customs=LineCustomCreate(
origin_country=origin or None,
fraction=fraction or None,
american_fraction=american_fraction or None,
),
description=LineDescriptionCreate(
extra_description=extra_desc or None,
additional_info_spanish=additional_info or None,
lot=lot or None,
entry_number=entry_number or None,
),
fa_data=FaLineItemCreateDTO(
search_invoice=factura_impo or None,
search_line=parse_int(linea_impo_val),
search_type=tipo_impo or None,
download=(descarga_val == 'SI'),
is_subitem=is_subitem,
contains_subitems=contains_subitems,
subitem_number=parse_int(linea_principal_val) if is_subitem else 0,
),
)
if not apply_export_defaults_and_calculations_for_csv(
session, line_data, tenant_id, company_id, line_num
):
skipped_invalid += 1
skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": "Factura sin datos financieros para enriquecer partida de exportación."})
continue
item_dict = line_data.model_dump(
exclude={"financial", "quantity", "customs", "description", "reference", "fa_data", "series"}
)
item_dict["tenant_id"] = tenant_id
item_dict["company_id"] = company_id
item_dict["line_number"] = line_num
line = LineItem(**item_dict)
session.add(line)
session.flush()
ItemService._create_line_nested_data(session, line, line_data, tenant_id, company_id)
session.add(InvoiceSalesDetails(
invoice_id=invoice_id,
line_number=line_num,
sales_order=(row_norm.get('ORDEN DE COMPRA') or row_norm.get('ORDENCOMPRA') or row_norm.get('ORDEN DE VENTA') or None),
sales_order=order_compra,
line_bundles=int_or_zero(row_norm.get('CANTIDAD BULTOS') or row_norm.get('CANTIDADBULTOS')),
tenant_id=tenant_id,
company_id=company_id,
@@ -5258,7 +5289,9 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
session.query(InvoiceSalesDetails).filter(InvoiceSalesDetails.invoice_id == invoice_id).delete(synchronize_session=False)
cleared_invoices.add(invoice_id)
# --- Partidas: LineItem with invoice_id (no Item parent) + full CSV mapping ---
# --- Partidas importación: paridad con flujo normal (validators + ItemService) ---
# LineReference: solo se crea si line_data.reference viene informado; no inventar datos sin fuente (plan paridad CSV).
# Build LineItemCreate from CSV, apply import defaults/calculations, then persist.
part_num = (row_norm.get('NUM. PARTE') or row_norm.get('NUMPARTE') or row_norm.get('NUMERO PARTE') or row_norm.get('NUM PARTE') or '').strip()
part_id = part_cache.get(part_num) if part_num else None
if part_id is None and part_num:
@@ -5279,23 +5312,6 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
bulk_key = (row_norm.get('CLAVE BULTOS') or row_norm.get('CLAVEBULTOS') or '').strip()
package_id = package_id_by_key.get(bulk_key) if bulk_key else None
line = LineItem(
invoice_id=invoice_id,
line_number=line_num,
tenant_id=tenant_id,
company_id=company_id,
part_number_id=part_id,
class_id=class_id,
unit_of_measure=uom_id,
order=(row_norm.get('ORDEN DE COMPRA') or row_norm.get('ORDENCOMPRA') or None),
material_type=(row_norm.get('ID TYPE') or row_norm.get('IDTYPE') or None),
tax_payment=(str(row_norm.get('SE PAGO IMPUESTO') or row_norm.get('SEPAGOIMPUESTO') or '').strip().upper() == 'SI'),
payment_method=(row_norm.get('FORMA DE PAGO') or row_norm.get('FORMADEPAGO') or row_norm.get('FORMA PAGO') or None),
valuation_method=(row_norm.get('METODO DE VALORACION') or row_norm.get('METODODEVALORACION') or row_norm.get('METODO VALORACION') or None),
)
session.add(line)
session.flush()
price = parse_decimal(row_norm.get('COSTO UNITARIO') or row_norm.get('PRECIO UNITARIO') or row_norm.get('PRECIOUNITARIO'))
if price is None:
total_val = parse_decimal(row_norm.get('TOTAL'))
@@ -5303,23 +5319,8 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
price = (total_val / qty) if (total_val and qty and qty != 0) else None
qty = parse_decimal(row_norm.get('CANTIDAD IMPORTADA') or row_norm.get('CANTIDAD'))
commercial_total = (price * qty) if price and qty else parse_decimal(row_norm.get('TOTAL'))
session.add(LineFinancial(
item_line_id=line.id,
unit_cost_capture=decimal_or_zero(price),
total_commercial_value=decimal_or_zero(commercial_total),
))
net_w = parse_decimal(row_norm.get('PESO NETO') or row_norm.get('PESONETO'))
gross_w = parse_decimal(row_norm.get('PESO BRUTO') or row_norm.get('PESOBRUTO'))
session.add(LineQuantity(
item_line_id=line.id,
quantity=decimal_or_zero(qty),
net_weight=decimal_or_zero(net_w),
gross_weight=decimal_or_zero(gross_w),
package_quantity=int_or_zero(row_norm.get('CANTIDAD BULTOS') or row_norm.get('CANTIDADBULTOS')),
package_id=package_id,
))
origin = (row_norm.get('PAIS ORIGEN') or row_norm.get('PAISORIGEN') or row_norm.get('PAIS') or '').strip()
fraction = (row_norm.get('FRACCION ARANCELARIA') or row_norm.get('FRACCION') or row_norm.get('FRACCIONARANCELARIA') or '').strip()
@@ -5328,14 +5329,6 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
fraction_type = (row_norm.get('PREFERENCIA ARANCELARIA') or row_norm.get('PREFERENCIA') or '').strip()
sector = (row_norm.get('SECTOR') or '').strip()
american_fraction = (row_norm.get('FRACCION AMERICANA') or row_norm.get('FRACCIONAMERICANA') or '').strip()
session.add(LineCustom(
item_line_id=line.id,
origin_country=origin or None,
fraction=fraction or None,
fraction_type=fraction_type or None,
sector=sector or None,
american_fraction=american_fraction or None,
))
desc_es = (row_norm.get('DESCRIPCION ESPAÑOL') or row_norm.get('DESCRIPCIONE') or row_norm.get('DESCRIPCION') or '').strip()
if not desc_es and class_code:
@@ -5349,18 +5342,65 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
additional_info = (row_norm.get('INFORMACION ADICIONAL') or row_norm.get('INFORMACIONADICIONAL') or '').strip()
lot = (row_norm.get('LOTE') or '').strip()
entry_number = (row_norm.get('NUMERO ENTRADA') or row_norm.get('NUMEROENTRADA') or row_norm.get('NUM ENTRADA') or '').strip()
session.add(LineDescription(
item_line_id=line.id,
description_spanish=desc_es or None,
description_english=desc_en or None,
brand=brand or None,
model=model or None,
extra_description=extra_desc or None,
additional_info_spanish=additional_info or None,
lot=lot or None,
entry_number=entry_number or None,
))
line_data = LineItemCreate(
invoice_id=invoice_id,
line_number=line_num,
part_number_id=part_id,
class_id=class_id,
unit_of_measure=uom_id,
order=(row_norm.get('ORDEN DE COMPRA') or row_norm.get('ORDENCOMPRA') or None),
material_type=(row_norm.get('ID TYPE') or row_norm.get('IDTYPE') or None),
tax_payment=(str(row_norm.get('SE PAGO IMPUESTO') or row_norm.get('SEPAGOIMPUESTO') or '').strip().upper() == 'SI'),
payment_method=(row_norm.get('FORMA DE PAGO') or row_norm.get('FORMADEPAGO') or row_norm.get('FORMA PAGO') or None),
valuation_method=(row_norm.get('METODO DE VALORACION') or row_norm.get('METODODEVALORACION') or row_norm.get('METODO VALORACION') or None),
financial=LineFinancialCreate(
unit_cost_capture=decimal_or_zero(price),
total_commercial_value=decimal_or_zero(commercial_total),
),
quantity=LineQuantityCreate(
quantity=decimal_or_zero(qty),
net_weight=decimal_or_zero(net_w),
gross_weight=decimal_or_zero(gross_w),
package_quantity=int_or_zero(row_norm.get('CANTIDAD BULTOS') or row_norm.get('CANTIDADBULTOS')),
package_id=package_id,
),
customs=LineCustomCreate(
origin_country=origin or None,
fraction=fraction or None,
fraction_type=fraction_type or None,
sector=sector or None,
american_fraction=american_fraction or None,
),
description=LineDescriptionCreate(
description_spanish=desc_es or None,
description_english=desc_en or None,
brand=brand or None,
model=model or None,
extra_description=extra_desc or None,
additional_info_spanish=additional_info or None,
lot=lot or None,
entry_number=entry_number or None,
),
fa_data=FaLineItemCreateDTO(is_subitem=False, contains_subitems=False),
)
if not apply_import_defaults_and_calculations_for_csv(
session, line_data, tenant_id, company_id, line_num
):
skipped_invalid += 1
skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": "Factura sin datos financieros/logísticos para enriquecer partida."})
continue
item_dict = line_data.model_dump(
exclude={"financial", "quantity", "customs", "description", "reference", "fa_data", "series"}
)
item_dict["tenant_id"] = tenant_id
item_dict["company_id"] = company_id
item_dict["line_number"] = line_num
line = LineItem(**item_dict)
session.add(line)
session.flush()
ItemService._create_line_nested_data(session, line, line_data, tenant_id, company_id)
session.add(InvoiceSalesDetails(
invoice_id=invoice_id,
line_number=line_num,

1
backend/api/v1/modules/a76/layouts_csv/parts/tasks.py
View File

@@ -2,6 +2,7 @@
Tareas Celery para importación CSV de Números de Parte.
Flujo: scan_file (validación) → insert_valid_rows (commit).
Paridad Clarion: actualizar (ACT), validación full/parcial, merge existente, reemplazar_sin_preguntar, RFC desde clase.
Sin PartService de creación en API; los mappers CSV (row_to_part_data, apply_rfc_exception_from_class) son la fuente de verdad para reglas de negocio al crear/actualizar.
"""
import json
import logging

2
backend/api/v1/modules/a76/layouts_csv/pedmientos/tasks.py
View File

@@ -2,6 +2,8 @@
Tareas Celery para importación CSV de Pedimentos.
Flujo: scan_file (validación) → insert_valid_rows (commit).
Usa layouts_csv.common (storage, normalize, meta, responses, csv_reader), fk_loader, validators, mappers.
Create/update delegan en PedimentosService; defaults de fechas (pedimento_dates) y merge vs replace
están alineados con el servicio para paridad con el flujo API.
"""
import json
import logging