From 09e20fb1097eb4acfcee672237231c46a2c5fa97 Mon Sep 17 00:00:00 2001 From: hreyes Date: Mon, 9 Mar 2026 14:34:54 -0600 Subject: [PATCH] feature/clarion-validaciones-invoices-csv-headers-comp-mex --- .../v1/modules/a76/csv_templates/registry.py | 13 +- .../api/v1/modules/a76/invoices/schemas.py | 2 +- .../modules/a76/layouts_csv/facturas/tasks.py | 375 ++++++++++++++---- .../layouts_csv/facturas/template_config.py | 27 ++ .../facturas/validators/__init__.py | 2 + .../facturas/validators/encabezados_cmex.py | 280 +++++++++++++ frontend/src/lib/config/csv-upload.ts | 3 +- 7 files changed, 623 insertions(+), 79 deletions(-) create mode 100644 backend/api/v1/modules/a76/layouts_csv/facturas/validators/encabezados_cmex.py diff --git a/backend/api/v1/modules/a76/csv_templates/registry.py b/backend/api/v1/modules/a76/csv_templates/registry.py index 076363c5..af15b48d 100644 --- a/backend/api/v1/modules/a76/csv_templates/registry.py +++ b/backend/api/v1/modules/a76/csv_templates/registry.py @@ -59,8 +59,16 @@ def _canonicals_from_columns(cols: Optional[List[Dict]]) -> List[str]: def _build_registry() -> Dict[str, List[str]]: registry: Dict[str, List[str]] = {} - # a76/imports (facturas): imp_temp_header, imp_temp_details, imp_def_*, exp_def_* - for tid in ("imp_temp_header", "imp_temp_details", "imp_def_header", "imp_def_details", "exp_def_header", "exp_def_details"): + # a76/imports (facturas): imp_temp_header, imp_temp_details, imp_def_*, exp_def_*, cmex_header + for tid in ( + "imp_temp_header", + "imp_temp_details", + "imp_def_header", + "imp_def_details", + "exp_def_header", + "exp_def_details", + "cmex_header", + ): cols = resolve_imports_template(tid) registry[tid] = _canonicals_from_columns(cols) @@ -139,6 +147,7 @@ TEMPLATE_FILENAMES: Dict[str, str] = { "imp_temp_details": "EstructuraParFacImpoTempAF.csv", "imp_def_header": "EstructuraEncFacImpoDef.csv", "imp_def_details": "EstructuraParFacImpoDefAF.csv", + "cmex_header": "EstructuraEncFacComprasMex.csv", "exp_def_header": "EstructuraEncFacExpoCamReg.csv", "exp_def_details": "EstructuraParExpoCamReg.csv", } diff --git a/backend/api/v1/modules/a76/invoices/schemas.py b/backend/api/v1/modules/a76/invoices/schemas.py index 55124a29..089429bb 100644 --- a/backend/api/v1/modules/a76/invoices/schemas.py +++ b/backend/api/v1/modules/a76/invoices/schemas.py @@ -131,7 +131,7 @@ class InvoiceComplianceMxBase(BaseModel): None, max_length=20, description="Shipped by header" ) shipped_by_id: Optional[int] = Field(None, description="Shipped by ID") - customs_broker_id: int = Field(None, description="Customs broker ID") + customs_broker_id: Optional[int] = Field(None, description="Customs broker ID (null for MEX)") customs_broker_us_id: Optional[int] = Field( None, description="US customs broker ID" ) diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py b/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py index 2646d960..7971749c 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py @@ -1703,6 +1703,186 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): logger.exception("Encabezados importación definitiva scan failed: %s", e) return {"status": "failed", "error": str(e)} + # --- Encabezados Compras Mexicanas: flujo específico (Clarion VALIDA_TODA_FAC_COM_MEX / VALIDA_PARCIAL) --- + if model_target == "invoice_header" and template_id == "cmex_header": + try: + from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceFinancials + from api.v1.modules.a76.items.models import LineItem + from api.v1.modules.a76.clients_and_providers.models import ClientProvider + from api.v1.modules.public.reference_data.incoterms.models import Incoterm + from api.v1.modules.public.reference_data.currency_types.models import CurrencyType + from api.v1.modules.a76.general_catalogs.exchange_rate.models import ExchangeRate + from api.v1.modules.a76.transportation.transporters.models import Transporter + from .validators.encabezados_cmex import validate_row_encabezados_cmex + + _fc = parse_footer_config(meta.get("footer_config")) + actualizar = meta.get("actualizar", False) + if _fc and "actualizar" in _fc: + actualizar = bool(_fc["actualizar"]) + + with CoreSessionLocal() as session: + q_inv = ( + session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.is_updated) + .filter( + InvoiceHeader.tenant_id == tenant_id, + InvoiceHeader.company_id == company_id, + InvoiceHeader.operation_type == "imp", + InvoiceHeader.invoice_type == "MEX", + ) + ) + invoice_exists_by_number = {} + invoice_updated_by_number = {} + for num, iid, is_upd in q_inv.all(): + if num: + n = str(num).strip() + invoice_exists_by_number[n] = True + invoice_updated_by_number[n] = bool(is_upd) + + valid_provider_ids = set() + valid_sold_to_ids = set() + valid_shipped_to_ids = set() + valid_provider_short_names = set() + valid_sold_to_short_names = set() + valid_shipped_to_short_names = set() + for cp in session.query(ClientProvider.id, ClientProvider.short_name).filter( + ClientProvider.tenant_id == tenant_id, + ClientProvider.company_id == company_id, + ).all(): + valid_provider_ids.add(cp[0]) + valid_sold_to_ids.add(cp[0]) + valid_shipped_to_ids.add(cp[0]) + if cp[1] and str(cp[1]).strip(): + sn_upper = str(cp[1]).strip().upper() + valid_provider_short_names.add(sn_upper) + valid_sold_to_short_names.add(sn_upper) + valid_shipped_to_short_names.add(sn_upper) + + valid_transporter_keys = set() + for t in session.query(Transporter.transporter_key).filter( + Transporter.tenant_id == tenant_id, + Transporter.company_id == company_id, + ).all(): + if t[0]: + valid_transporter_keys.add((t[0] or "").strip().upper()) + + valid_incoterms = set() + for inc in session.query(Incoterm.code).all(): + if inc[0]: + valid_incoterms.add((inc[0] or "").strip().upper()) + + valid_currency_codes = set() + for ct in session.query(CurrencyType.code).all(): + if ct[0]: + valid_currency_codes.add((ct[0] or "").strip().upper()) + + exchange_rate_by_date = {} + for er in session.query(ExchangeRate.date, ExchangeRate.value).filter( + ExchangeRate.tenant_id == tenant_id, + ExchangeRate.company_id == company_id, + ).all(): + if er[0] and er[1] is not None: + dk = er[0].strftime("%Y-%m-%d") if hasattr(er[0], "strftime") else str(er[0])[:10] + exchange_rate_by_date[dk] = er[1] + + invoice_has_partidas_by_number = {} + existing_tipo_moneda_by_number = {} + q_li_count = ( + session.query(InvoiceHeader.invoice_number, func.count(LineItem.id)) + .join(LineItem, LineItem.invoice_id == InvoiceHeader.id) + .filter( + InvoiceHeader.tenant_id == tenant_id, + InvoiceHeader.company_id == company_id, + InvoiceHeader.operation_type == "imp", + InvoiceHeader.invoice_type == "MEX", + ) + .group_by(InvoiceHeader.invoice_number) + ) + for num, cnt in q_li_count.all(): + if num: + invoice_has_partidas_by_number[str(num).strip()] = cnt > 0 + q_fin = ( + session.query(InvoiceHeader.invoice_number, InvoiceFinancials.currency) + .join(InvoiceFinancials, InvoiceFinancials.invoice_id == InvoiceHeader.id) + .filter( + InvoiceHeader.tenant_id == tenant_id, + InvoiceHeader.company_id == company_id, + InvoiceHeader.operation_type == "imp", + InvoiceHeader.invoice_type == "MEX", + ) + ) + for num, cur in q_fin.all(): + if num and cur: + cur_str = (cur or "").strip().lower() + if cur_str == "foreign": + existing_tipo_moneda_by_number[str(num).strip()] = "ME" + elif cur_str == "local": + existing_tipo_moneda_by_number[str(num).strip()] = "MN" + else: + existing_tipo_moneda_by_number[str(num).strip()] = (cur_str or "").upper()[:2] + + with open(file_path, "r", encoding="utf-8-sig") as f_in: + sample = f_in.read(2048) + f_in.seek(0) + try: + dialect = csv.Sniffer().sniff(sample, delimiters=",;\t") + except Exception: + dialect = "excel" + reader = csv.DictReader(f_in, dialect=dialect) + rows_list = list(reader) + + error_count = 0 + processed_rows = 0 + error_lines_list = [] + errors_detail = [] + + with open(error_path, "w", encoding="utf-8") as f_err: + for i, row in enumerate(rows_list, start=1): + if i % 1000 == 0: + self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows, "errors": error_count}) + row_norm = row_from_template(row, "cmex_header", normalize_header) + warnings_row = [] + err = validate_row_encabezados_cmex( + row_norm, + i, + actualizar=actualizar, + invoice_exists_by_number=invoice_exists_by_number, + invoice_updated_by_number=invoice_updated_by_number, + valid_provider_ids=valid_provider_ids, + valid_sold_to_ids=valid_sold_to_ids, + valid_shipped_to_ids=valid_shipped_to_ids, + valid_transporter_keys=valid_transporter_keys, + valid_incoterms=valid_incoterms, + valid_currency_codes=valid_currency_codes, + exchange_rate_by_date=exchange_rate_by_date, + invoice_has_partidas_by_number=invoice_has_partidas_by_number, + existing_tipo_moneda_by_number=existing_tipo_moneda_by_number, + valid_provider_short_names=valid_provider_short_names, + valid_sold_to_short_names=valid_sold_to_short_names, + valid_shipped_to_short_names=valid_shipped_to_short_names, + date_format=date_format, + parse_date_fn=parse_date, + warnings=warnings_row, + ) + if err: + error_count += 1 + error_lines_list.append(err["line"]) + f_err.write(json.dumps({"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")}) + "\n") + if len(errors_detail) < 500: + errors_detail.append({"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")}) + for w in warnings_row: + if len(errors_detail) < 500: + errors_detail.append({"line": w["line"], "col": w.get("col", ""), "msg": w.get("msg", ""), "warning": True}) + processed_rows += 1 + + if error_lines_list: + common_storage.store_error_lines(JOB_TYPE, job_id, error_lines_list) + return common_responses.scan_result( + job_id, processed_rows, error_count, errors_detail, total_rows_in_file=total_rows + ) + except Exception as e: + logger.exception("Encabezados Compras Mexicanas scan failed: %s", e) + return {"status": "failed", "error": str(e)} + try: from api.v1.modules.a76.invoices.models import InvoiceHeader from api.v1.modules.a76.clients_and_providers.models import ClientProvider @@ -2905,6 +3085,8 @@ def insert_valid_rows(self, job_id: str, model_target: str): ) if model_target == "invoice_header" and _template_id_insert == "imp_def_header": inv_type_value = "DEF" + if model_target == "invoice_header" and _template_id_insert == "cmex_header": + inv_type_value = "MEX" if model_target == "invoice_details" and _template_id_insert == "imp_def_details": inv_type_value = "DEF" @@ -3035,63 +3217,94 @@ def insert_valid_rows(self, job_id: str, model_target: str): logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") continue - err = _validate_customs_broker_ref( + err = _validate_client_provider_ref( validator, - CustomsBroker, - row_norm.get('AGENTE ADUANAL'), + ClientProvider, + row_norm.get('CLAVE ENVIADO A'), i, - "AGENTE ADUANAL", - required=False, + "CLAVE ENVIADO A", + required=True, ) if err: - skipped_missing_fk += 1 + skipped_invalid += 1 reason = f"{err['col']}: {err['msg']}" skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") continue - err = validate_public_code( - validator, - RegimenPedimento, - row_norm.get('REGIMEN') or row_norm.get('CLAVEDOCUMENTO'), - i, - "CLAVEDOCUMENTO", - ) - if err: - skipped_missing_fk += 1 - reason = f"{err['col']}: {err['msg']}" - skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) - logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") - continue + if inv_type_value != "MEX": + err = _validate_customs_broker_ref( + validator, + CustomsBroker, + row_norm.get('AGENTE ADUANAL'), + i, + "AGENTE ADUANAL", + required=False, + ) + if err: + skipped_missing_fk += 1 + reason = f"{err['col']}: {err['msg']}" + skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) + logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") + continue - err = validate_public_code( - validator, - CustomsSection, - row_norm.get('ADUANA DE CRUCE'), - i, - "ADUANA DE CRUCE", - field_name="customs_code", - ) - if err: - skipped_missing_fk += 1 - reason = f"{err['col']}: {err['msg']}" - skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) - logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") - continue + err = validate_public_code( + validator, + RegimenPedimento, + row_norm.get('REGIMEN') or row_norm.get('CLAVEDOCUMENTO'), + i, + "CLAVEDOCUMENTO", + ) + if err: + skipped_missing_fk += 1 + reason = f"{err['col']}: {err['msg']}" + skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) + logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") + continue - err = validate_public_code( - validator, - CurrencyType, - row_norm.get('CLAVE MONEDA'), - i, - "CLAVE MONEDA", - ) - if err: - skipped_missing_fk += 1 - reason = f"{err['col']}: {err['msg']}" - skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) - logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") - continue + err = validate_public_code( + validator, + CustomsSection, + row_norm.get('ADUANA DE CRUCE'), + i, + "ADUANA DE CRUCE", + field_name="customs_code", + ) + if err: + skipped_missing_fk += 1 + reason = f"{err['col']}: {err['msg']}" + skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) + logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") + continue + + if inv_type_value == "MEX": + err = validate_public_code( + validator, + CurrencyType, + row_norm.get('CLAVE MONEDA'), + i, + "CLAVE MONEDA", + ) + if err and (row_norm.get('TIPO MONEDA') or '').strip().upper() == 'MC': + skipped_missing_fk += 1 + reason = f"{err['col']}: {err['msg']}" + skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) + logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") + continue + else: + err = validate_public_code( + validator, + CurrencyType, + row_norm.get('CLAVE MONEDA'), + i, + "CLAVE MONEDA", + ) + if err: + skipped_missing_fk += 1 + reason = f"{err['col']}: {err['msg']}" + skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) + logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") + continue err = validate_public_code( validator, @@ -3145,8 +3358,8 @@ def insert_valid_rows(self, job_id: str, model_target: str): # --- Resolve PEDIMENTO (Col A) to pedimento_id and REMESA (Col B) --- pedimento_id = None - remesa_val = parse_int(row_norm.get('REMESA')) - ped_str = (row_norm.get('PEDIMENTO') or '').strip() + remesa_val = parse_int(row_norm.get('REMESA')) if inv_type_value != "MEX" else None + ped_str = (row_norm.get('PEDIMENTO') or '').strip() if inv_type_value != "MEX" else '' if ped_str: parsed = parse_pedimento_col_a(ped_str) if parsed: @@ -3182,12 +3395,15 @@ def insert_valid_rows(self, job_id: str, model_target: str): header.operation_type = op_type_value header.is_updated = True # Mark as updated header.updated_date = datetime.utcnow() - header.document_type = resolve_public_code( - session, - RegimenPedimento, - RegimenPedimento.code, - (row_norm.get('REGIMEN') or row_norm.get('CLAVEDOCUMENTO')), - regimen_cache, + header.document_type = ( + None if inv_type_value == "MEX" else + resolve_public_code( + session, + RegimenPedimento, + RegimenPedimento.code, + (row_norm.get('REGIMEN') or row_norm.get('CLAVEDOCUMENTO')), + regimen_cache, + ) ) header.project_number = (row_norm.get('NUM PROYECTO') or row_norm.get('NUMPROYECTO') or None) header.purchase_order = (row_norm.get('ORDEN COMPRA') or row_norm.get('ORDENCOMPRA') or None) @@ -3214,12 +3430,15 @@ def insert_valid_rows(self, job_id: str, model_target: str): system="CSV", capture_date=datetime.utcnow(), invoice_type=inv_type_value, - document_type=resolve_public_code( - session, - RegimenPedimento, - RegimenPedimento.code, - (row_norm.get('REGIMEN') or row_norm.get('CLAVEDOCUMENTO')), - regimen_cache, + document_type=( + None if inv_type_value == "MEX" else + resolve_public_code( + session, + RegimenPedimento, + RegimenPedimento.code, + (row_norm.get('REGIMEN') or row_norm.get('CLAVEDOCUMENTO')), + regimen_cache, + ) ), project_number=(row_norm.get('NUM PROYECTO') or row_norm.get('NUMPROYECTO') or None), purchase_order=(row_norm.get('ORDEN COMPRA') or row_norm.get('ORDENCOMPRA') or None), @@ -3233,14 +3452,17 @@ def insert_valid_rows(self, job_id: str, model_target: str): ) compliance = InvoiceComplianceMx( - pedimento_id=pedimento_id, - remesa=remesa_val, - aduana=resolve_public_code( - session, - CustomsSection, - CustomsSection.customs_code, - row_norm.get('ADUANA DE CRUCE'), - customs_section_cache, + pedimento_id=pedimento_id if inv_type_value != "MEX" else None, + remesa=remesa_val if inv_type_value != "MEX" else None, + aduana=( + None if inv_type_value == "MEX" else + resolve_public_code( + session, + CustomsSection, + CustomsSection.customs_code, + row_norm.get('ADUANA DE CRUCE'), + customs_section_cache, + ) ), provider_id=resolve_client_provider_id( session, @@ -3266,13 +3488,16 @@ def insert_valid_rows(self, job_id: str, model_target: str): company_id, shipped_to_cache, ), - customs_broker_id=resolve_customs_broker_id( - session, - CustomsBroker, - row_norm.get('AGENTE ADUANAL'), - tenant_id, - company_id, - broker_cache, + customs_broker_id=( + None if inv_type_value == "MEX" else + resolve_customs_broker_id( + session, + CustomsBroker, + row_norm.get('AGENTE ADUANAL'), + tenant_id, + company_id, + broker_cache, + ) ), edocument=(row_norm.get('E DOCUMENT') or None), vucem_operation_num=(row_norm.get('NUM OPERACION') or None), diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/template_config.py b/backend/api/v1/modules/a76/layouts_csv/facturas/template_config.py index a5e371ff..ec822a38 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/template_config.py @@ -84,6 +84,33 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { ], # --- Encabezado factura: Expo (EstructuraEncFacExpoCamReg.xls) - misma estructura --- "exp_def_header": None, + # --- Encabezado factura: Compras Mexicanas (Clarion VALIDA_TODA_FAC_COM_MEX / VALIDA_PARCIAL) --- + # Estructura CSV: A,B=CAPTURAR CMEX; C=NUMERO FACTURA; D=FECHA FACTURA; E=TIPO DE CAMBIO; F=CAPTURAR CMEX; + # G=CLAVE PROVEEDOR; H=CLAVE VENDIDO A; I=CLAVE ENVIADO A; J=CAPTURAR CMEX; K=CLAVE TRANSPORTISTA; ...; Z=OBSERVACIONES E + "cmex_header": [ + {"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA"]}, + {"canonical": "FECHA FACTURA", "aliases": ["FECHA"]}, + {"canonical": "TIPO DE CAMBIO"}, + {"canonical": "CLAVE PROVEEDOR"}, + {"canonical": "CLAVE VENDIDO A", "aliases": ["CLAVE VENDIDO A:"]}, + {"canonical": "CLAVE ENVIADO A"}, + {"canonical": "CLAVE TRANSPORTISTA"}, + {"canonical": "NOMBRE CONDUCTOR"}, + {"canonical": "TIPO TRANSPORTE"}, + {"canonical": "NUMERO TRANSPORTE"}, + {"canonical": "TIPO MONEDA"}, + {"canonical": "CLAVE MONEDA"}, + {"canonical": "FLETES"}, + {"canonical": "VALOR SEGUROS"}, + {"canonical": "SEGUROS"}, + {"canonical": "EMBALAJES"}, + {"canonical": "OTROS INCREMENTABLES"}, + {"canonical": "CLAVE INCOTERM"}, + {"canonical": "PRECINTO"}, + {"canonical": "FECHA EMISION"}, + {"canonical": "TIPO PESO"}, + {"canonical": "OBSERVACIONES E"}, + ], # --- Partidas factura: Impo Temp (EstructuraParFacImpoTemp - paridad Clarion A-AG) --- "imp_temp_details": [ {"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA"]}, diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/__init__.py b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/__init__.py index 650f053a..5ef9ed99 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/__init__.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/__init__.py @@ -9,6 +9,7 @@ from .encabezados_impo_def import ( validate_row_encabezados_impo_def, parse_pedimento_col_a_impo_def, ) +from .encabezados_cmex import validate_row_encabezados_cmex from .partidas_impo_def import validate_row_partidas_impo_def from .series_impo_def import ( validate_row_series_impo_def, @@ -18,6 +19,7 @@ from .series_impo_def import ( __all__ = [ "validate_row_encabezados_impo_temp", "validate_row_encabezados_impo_def", + "validate_row_encabezados_cmex", "validate_row_partidas_impo_def", "validate_row_series_impo_def", "row_to_series_normalized_def", diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/encabezados_cmex.py b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/encabezados_cmex.py new file mode 100644 index 00000000..bdff4a3f --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/encabezados_cmex.py @@ -0,0 +1,280 @@ +""" +Validaciones CSV para Encabezados de Facturas de Compras Mexicanas. +Paridad Clarion: VALIDA_TODA_FAC_COM_MEX, VALIDA_PARCIAL_FAC_COM_MEX, VALIDACIONES_FAC_COM_MEX. +Sin pedimento, remesa, agente aduanal ni aduana de cruce. +Estructura CSV: NUMERO FACTURA (C), FECHA FACTURA (D), TIPO DE CAMBIO (E), CLAVE PROVEEDOR (G), ... +""" +from datetime import datetime +from decimal import Decimal +from typing import Any, Dict, List, Optional, Set + +from .encabezados_impo_temp import ( + MAX_LEN_FACTURA, + TIPO_PESO_VALIDOS, + TIPOS_MONEDA_VALIDOS, + _clip, + _err, + _get, + _parse_decimal, + _parse_int, + _validaciones_factura_longitud, + _validaciones_moneda, + _validaciones_tipo_cambio, + _validaciones_tipo_peso, +) + +# Clarion Col M: Compras Mexicanas incluye "FERRO BARCAZA" (con espacio) y "PLATAFORMA" +TIPO_TRANSPORTE_VALIDOS_CMEX = frozenset({ + "NINGUNO", "TRANSPORTE", "CAJA", "PLACAS", "CAMION", "BUQUE", + "FERROBARCAZA", "FERRO BARCAZA", "CONTENEDOR", "PLATAFORMA", "AVION", +}) + + +def _validaciones_transporte_cmex(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + """Tipo transporte (Col M) y número (Col N). Acepta FERRO BARCAZA y FERROBARCAZA.""" + m_raw = _get(row, "TIPO TRANSPORTE") + m = m_raw.upper().replace(" ", "") if m_raw else "" + m_with_space = m_raw.upper() if m_raw else "" + n = _get(row, "NUMERO TRANSPORTE") + if m_raw and m_with_space not in TIPO_TRANSPORTE_VALIDOS_CMEX and m not in TIPO_TRANSPORTE_VALIDOS_CMEX: + return _err( + line_num, + "TIPO TRANSPORTE", + "Error: (Celda M{}) El Tipo de Transporte: {} no es válido. " + "Válidos: NINGUNO, TRANSPORTE, CAJA, PLACAS, CAMION, BUQUE, FERRO BARCAZA, CONTENEDOR, PLATAFORMA, AVION.".format( + line_num, m_raw + ), + ) + if not m_raw and n: + return _err( + line_num, + "NUMERO TRANSPORTE", + "Error: (Celda N{}) El Tipo de Transporte está vacío y está capturado un número de transporte.".format(line_num), + ) + if m_raw and (m == "NINGUNO" or m_with_space == "NINGUNO") and n: + return _err( + line_num, + "NUMERO TRANSPORTE", + "Error: (Celda N{}) El Tipo de Transporte es NINGUNO y está capturado un número de transporte.".format(line_num), + ) + if m_raw and m != "NINGUNO" and m_with_space != "NINGUNO" and not n: + return _err( + line_num, + "NUMERO TRANSPORTE", + "Error: (Celda N{}) El Tipo de Transporte es {} y no está capturado el número de transporte.".format(line_num, m_raw), + ) + return None + + +def _validaciones_obligatorios_toda_cmex( + row: Dict[str, Any], + line_num: int, + actualizar: bool, +) -> Optional[Dict[str, Any]]: + """Obligatorios VALIDA_TODA_FAC_COM_MEX: C siempre; si no es actualizar, también D, G, H, I.""" + obligatorios: List[str] = [] + if not _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA"): + obligatorios.append("(Col.C) Factura") + if not actualizar: + if not _get(row, "FECHA FACTURA", "FECHA"): + obligatorios.append("(Col.D) Fecha de la Factura") + if not _get(row, "CLAVE PROVEEDOR"): + obligatorios.append("(Col.G) Clave del Proveedor") + if not _get(row, "CLAVE VENDIDO A"): + obligatorios.append("(Col.H) Clave del Vendido A") + if not _get(row, "CLAVE ENVIADO A"): + obligatorios.append("(Col.I) Clave del Enviado A") + if obligatorios: + return _err( + line_num, + "ARCHIVO CSV", + "Existen campos vacíos que son obligatorios: {}. Revisar la línea del archivo y capturar los campos con la información correcta.".format( + ", ".join(obligatorios) + ), + ) + return None + + +def _validaciones_catalogos_cmex( + row: Dict[str, Any], + line_num: int, + valid_provider_ids: Set[int], + valid_sold_to_ids: Set[int], + valid_shipped_to_ids: Set[int], + valid_provider_short_names: Set[str], + valid_sold_to_short_names: Set[str], + valid_shipped_to_short_names: Set[str], + valid_transporter_keys: Set[str], + valid_incoterms: Set[str], +) -> Optional[Dict[str, Any]]: + """Catálogos para Compras Mexicanas: Proveedor, Vendido A, Enviado A, Transportista, Incoterm (sin Agente Aduanal ni Aduana).""" + + def check_id_or_rfc( + val: Any, + col: str, + catalog_name: str, + valid_ids: Set[int], + valid_short_names: Set[str], + ) -> Optional[Dict[str, Any]]: + if val is None or str(val).strip() == "": + return None + v = _parse_int(val) + if v is not None: + if valid_ids and v not in valid_ids: + return _err(line_num, col, "Error: La clave en {} no existe en el Catálogo de {}.".format(col, catalog_name)) + return None + sn_norm = str(val).strip().upper() + if valid_short_names and sn_norm not in valid_short_names: + return _err(line_num, col, "Error: La clave/corta en {} no existe en el Catálogo de {}.".format(col, catalog_name)) + if not valid_short_names: + return _err(line_num, col, "Error: (Celda) {} debe ser un número entero o clave corta (short name) válida.".format(col)) + return None + + err = check_id_or_rfc( + row.get("CLAVE PROVEEDOR"), "CLAVE PROVEEDOR", "Clientes/Proveedores", + valid_provider_ids, valid_provider_short_names, + ) + if err: + return err + err = check_id_or_rfc( + row.get("CLAVE VENDIDO A"), "CLAVE VENDIDO A", "Clientes/Proveedores", + valid_sold_to_ids, valid_sold_to_short_names, + ) + if err: + return err + err = check_id_or_rfc( + row.get("CLAVE ENVIADO A"), "CLAVE ENVIADO A", "Clientes/Proveedores", + valid_shipped_to_ids, valid_shipped_to_short_names, + ) + if err: + return err + + k = _get(row, "CLAVE TRANSPORTISTA") + if k and valid_transporter_keys and k.upper() not in valid_transporter_keys: + return _err( + line_num, + "CLAVE TRANSPORTISTA", + "Error: (Celda K{}) La Clave del Transportista: {} no existe en el Catálogo de Transportistas.".format(line_num, k), + ) + + v = _get(row, "CLAVE INCOTERM") + if v and valid_incoterms and v.upper() not in valid_incoterms: + return _err( + line_num, + "CLAVE INCOTERM", + "Error: (Celda V{}) La Clave de INCOTERM: {} no existe en el Catálogo de INCOTERMS.".format(line_num, v), + ) + + return None + + +def validate_row_encabezados_cmex( + row: Dict[str, Any], + line_num: int, + actualizar: bool, + invoice_exists_by_number: Dict[str, bool], + invoice_updated_by_number: Dict[str, bool], + valid_provider_ids: Set[int], + valid_sold_to_ids: Set[int], + valid_shipped_to_ids: Set[int], + valid_transporter_keys: Set[str], + valid_incoterms: Set[str], + valid_currency_codes: Set[str], + exchange_rate_by_date: Optional[Dict[str, Any]] = None, + invoice_has_partidas_by_number: Optional[Dict[str, bool]] = None, + existing_tipo_moneda_by_number: Optional[Dict[str, str]] = None, + valid_provider_short_names: Optional[Set[str]] = None, + valid_sold_to_short_names: Optional[Set[str]] = None, + valid_shipped_to_short_names: Optional[Set[str]] = None, + date_format: Optional[str] = None, + parse_date_fn=None, + warnings: Optional[List[Dict[str, Any]]] = None, +) -> Optional[Dict[str, Any]]: + """ + Valida una fila de CSV de Encabezados de Compras Mexicanas. + Clarion: VALIDA_TODA_FAC_COM_MEX (factura nueva o no actualizar) vs VALIDA_PARCIAL_FAC_COM_MEX (actualizar existente). + Siempre ejecuta VALIDACIONES_FAC_COM_MEX (longitud C, tipo cambio, catálogos G/H/I/K, transporte M/N, moneda O/P, incoterm V, tipo peso Y). + """ + factura = _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA") + if not factura: + return _err( + line_num, + "NUMERO FACTURA", + "Error: (Col.C) La columna de Número de Factura está vacía y no se pueden hacer las validaciones.", + ) + + if invoice_updated_by_number.get(factura.strip(), False): + return _err( + line_num, + "NUMERO FACTURA", + "Error: (Celda C{}) El Número de Factura: {} ya existe y está Actualizada, no se puede hacer cambios.".format( + line_num, factura + ), + ) + + if actualizar and factura.strip() not in invoice_exists_by_number: + return _err( + line_num, + "NUMERO FACTURA", + "Error: (Col.C) Factura de importación {} no existe (modo Actualizar).".format(factura), + ) + + use_partial = actualizar and invoice_exists_by_number.get(factura.strip(), False) + + if not use_partial: + err = _validaciones_obligatorios_toda_cmex(row, line_num, actualizar) + if err: + return err + + err = _validaciones_factura_longitud(row, line_num) + if err: + return err + + err = _validaciones_transporte_cmex(row, line_num) + if err: + return err + + has_partidas = invoice_has_partidas_by_number.get(factura.strip(), False) if invoice_has_partidas_by_number else False + existing_moneda = existing_tipo_moneda_by_number.get(factura.strip()) if existing_tipo_moneda_by_number else None + err = _validaciones_moneda( + row, + line_num, + valid_currency_codes or set(), + has_partidas if use_partial else None, + existing_moneda if use_partial else None, + ) + if err: + return err + + err = _validaciones_tipo_peso(row, line_num) + if err: + return err + + err = _validaciones_catalogos_cmex( + row, + line_num, + valid_provider_ids or set(), + valid_sold_to_ids or set(), + valid_shipped_to_ids or set(), + valid_provider_short_names or set(), + valid_sold_to_short_names or set(), + valid_shipped_to_short_names or set(), + valid_transporter_keys or set(), + valid_incoterms or set(), + ) + if err: + return err + + invoice_date_parsed = None + if parse_date_fn: + date_str = _get(row, "FECHA FACTURA", "FECHA") + if date_str: + invoice_date_parsed = parse_date_fn(date_str, date_format) + + err = _validaciones_tipo_cambio( + row, line_num, invoice_date_parsed, exchange_rate_by_date or {}, warnings + ) + if err: + return err + + return None diff --git a/frontend/src/lib/config/csv-upload.ts b/frontend/src/lib/config/csv-upload.ts index 3f67eb5b..b19f5af7 100644 --- a/frontend/src/lib/config/csv-upload.ts +++ b/frontend/src/lib/config/csv-upload.ts @@ -394,7 +394,8 @@ export const importacionConfig: CsvUploadItem[] = [ icon: FileText, group: 'Compras Mex.', modelTarget: 'invoice_header', - disabled: true, + templateId: 'cmex_header', + layoutModule: 'layouts_csv/facturas' }, { id: 'comp_mex_details',