From a1c3b97375655d72e2216aa5d8bae3ce5848c85b Mon Sep 17 00:00:00 2001 From: hreyes Date: Tue, 10 Mar 2026 10:31:30 -0600 Subject: [PATCH] feature/clarion-expo-csv-header --- .../a76/layouts_csv/boms/template_config.py | 6 +- .../template_config.py | 6 +- .../layouts_csv/classes/template_config.py | 9 +- .../clients_and_providers/template_config.py | 6 +- .../a76/layouts_csv/common/cell_value.py | 19 + .../customs_brokers/template_config.py | 6 +- .../layouts_csv/drivers/template_config.py | 6 +- .../exchange_rate/template_config.py | 6 +- .../exchange_rate/validators/common.py | 15 +- .../a76/layouts_csv/exportacion/tasks.py | 20 +- .../exportacion/template_config.py | 6 +- .../modules/a76/layouts_csv/facturas/tasks.py | 521 +++++++++++++++--- .../layouts_csv/facturas/template_config.py | 49 +- .../facturas/validators/__init__.py | 2 + .../facturas/validators/encabezados_expo.py | 464 ++++++++++++++++ .../a76/layouts_csv/parts/template_config.py | 6 +- .../layouts_csv/pedmientos/template_config.py | 11 +- .../pedmientos/validators/common.py | 15 +- .../layouts_csv/trailers/template_config.py | 6 +- .../transportistas/template_config.py | 6 +- .../us_tariff_fractions/template_config.py | 8 +- .../us_tariff_fractions/validators/common.py | 4 +- .../layouts_csv/vehicles/template_config.py | 6 +- 23 files changed, 1086 insertions(+), 117 deletions(-) create mode 100644 backend/api/v1/modules/a76/layouts_csv/common/cell_value.py create mode 100644 backend/api/v1/modules/a76/layouts_csv/facturas/validators/encabezados_expo.py diff --git a/backend/api/v1/modules/a76/layouts_csv/boms/template_config.py b/backend/api/v1/modules/a76/layouts_csv/boms/template_config.py index 1d4efa48..eb50d89f 100644 --- a/backend/api/v1/modules/a76/layouts_csv/boms/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/boms/template_config.py @@ -5,6 +5,8 @@ Placeholders hasta tener el XLS definitivo; ajustar canónicos y aliases según from typing import Dict, List, Any +from ..common.cell_value import cell_to_str + TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { "boms": [ {"canonical": "NUMPARTE_PADRE", "aliases": ["PARTE PADRE", "PART NUMBER", "PARENT PART", "NUM PARTE PADRE"]}, @@ -33,10 +35,10 @@ def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]: def row_from_template(row: Dict[str, Any], normalize_header_fn) -> Dict[str, Any]: lookup = build_normalized_lookup(normalize_header_fn) if not lookup: - return {normalize_header_fn(k): v for k, v in row.items()} + return {normalize_header_fn(k): cell_to_str(v) for k, v in row.items()} out: Dict[str, Any] = {} for csv_header, value in row.items(): key_norm = normalize_header_fn(csv_header) if key_norm in lookup: - out[lookup[key_norm]] = value + out[lookup[key_norm]] = cell_to_str(value) return out diff --git a/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/template_config.py b/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/template_config.py index 642e413e..b780736c 100644 --- a/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/template_config.py @@ -5,6 +5,8 @@ Por ahora misma estructura que encabezado/partidas de exportación; luego se aju from typing import Dict, List, Any, Optional +from ..common.cell_value import cell_to_str + # Cambio de régimen: cam_reg_header, cam_reg_details # Regularización: regulariz_header, regulariz_details TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { @@ -129,10 +131,10 @@ def row_from_template(row: Dict[str, Any], template_id: str, normalize_header_fn """Fila CSV -> dict con nombres canónicos de la plantilla.""" lookup = build_normalized_lookup(template_id, normalize_header_fn) if not lookup: - return {normalize_header_fn(k): v for k, v in row.items()} + return {normalize_header_fn(k): cell_to_str(v) for k, v in row.items()} out: Dict[str, Any] = {} for csv_header, value in row.items(): key_norm = normalize_header_fn(csv_header) if key_norm in lookup: - out[lookup[key_norm]] = value + out[lookup[key_norm]] = cell_to_str(value) return out diff --git a/backend/api/v1/modules/a76/layouts_csv/classes/template_config.py b/backend/api/v1/modules/a76/layouts_csv/classes/template_config.py index b4c9623e..f4d9d2e5 100644 --- a/backend/api/v1/modules/a76/layouts_csv/classes/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/classes/template_config.py @@ -6,6 +6,8 @@ import csv import io from typing import Dict, List, Any, Optional, Tuple +from ..common.cell_value import cell_to_str + # Valores que indican que la primera fila es cabecera (primera columna normalizada) FIRST_COLUMN_HEADER_VALUES = ("CLAVE CLASE", "CLASE") @@ -92,16 +94,17 @@ def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]: def row_from_template(row: Dict[str, Any], normalize_header_fn) -> Dict[str, Any]: lookup = build_normalized_lookup(normalize_header_fn) if not lookup: - return {normalize_header_fn(k): v for k, v in row.items()} + return {normalize_header_fn(k): cell_to_str(v) for k, v in row.items()} out: Dict[str, Any] = {} for csv_header, value in row.items(): key_norm = normalize_header_fn(csv_header) if key_norm in lookup: - out[lookup[key_norm]] = value + out[lookup[key_norm]] = cell_to_str(value) elif key_norm.startswith("CLAVE CLASE"): # CSV leído con delimitador incorrecto: primera columna es "CLAVE CLASE,..." -> usar primer valor como CLASE if "CLASE" not in out and value: - first_val = (value.split(",")[0] if "," in str(value) else value).strip() + val_str = cell_to_str(value) + first_val = (val_str.split(",")[0] if "," in val_str else val_str).strip() if first_val: out["CLASE"] = first_val return out diff --git a/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/template_config.py b/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/template_config.py index be387efc..c06c1e2a 100644 --- a/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/template_config.py @@ -12,6 +12,8 @@ AH=COL_EXTRA (desfase). from typing import Dict, List, Any, Optional +from ..common.cell_value import cell_to_str + TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { "client_providers": [ # Col A - Procedencia (E=Extranjero, N=Nacional) @@ -109,10 +111,10 @@ def row_from_template(row: Dict[str, Any], normalize_header_fn) -> Dict[str, Any """Fila CSV con solo columnas de la plantilla, en nombres canónicos.""" lookup = build_normalized_lookup(normalize_header_fn) if not lookup: - return {normalize_header_fn(k): v for k, v in row.items()} + return {normalize_header_fn(k): cell_to_str(v) for k, v in row.items()} out: Dict[str, Any] = {} for csv_header, value in row.items(): key_norm = normalize_header_fn(csv_header) if key_norm in lookup: - out[lookup[key_norm]] = value + out[lookup[key_norm]] = cell_to_str(value) return out diff --git a/backend/api/v1/modules/a76/layouts_csv/common/cell_value.py b/backend/api/v1/modules/a76/layouts_csv/common/cell_value.py new file mode 100644 index 00000000..a5989742 --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/common/cell_value.py @@ -0,0 +1,19 @@ +""" +Convierte valor de celda CSV a str. Evita 'list' object has no attribute 'strip' +cuando columnas duplicadas o el lector devuelve listas. +""" +from typing import Any + + +def cell_to_str(value: Any) -> str: + """ + Convierte valor de celda a str. + Si es lista (p. ej. CSV con columnas duplicadas), usa el primer elemento. + """ + if value is None: + return "" + if isinstance(value, list): + if not value: + return "" + return str(value[0]) if value[0] is not None else "" + return str(value) diff --git a/backend/api/v1/modules/a76/layouts_csv/customs_brokers/template_config.py b/backend/api/v1/modules/a76/layouts_csv/customs_brokers/template_config.py index b656b59a..b1a3afa6 100644 --- a/backend/api/v1/modules/a76/layouts_csv/customs_brokers/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/customs_brokers/template_config.py @@ -8,6 +8,8 @@ Solo se leen columnas definidas aquí; el resto se ignora. from typing import Dict, List, Any, Optional +from ..common.cell_value import cell_to_str + TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { "customs_brokers": [ # Col A - TIPO (MEX/Mexicano, AME/Americano) @@ -91,10 +93,10 @@ def row_from_template(row: Dict[str, Any], normalize_header_fn) -> Dict[str, Any """Fila CSV con solo columnas de la plantilla, en nombres canónicos.""" lookup = build_normalized_lookup(normalize_header_fn) if not lookup: - return {normalize_header_fn(k): v for k, v in row.items()} + return {normalize_header_fn(k): cell_to_str(v) for k, v in row.items()} out: Dict[str, Any] = {} for csv_header, value in row.items(): key_norm = normalize_header_fn(csv_header) if key_norm in lookup: - out[lookup[key_norm]] = value + out[lookup[key_norm]] = cell_to_str(value) return out diff --git a/backend/api/v1/modules/a76/layouts_csv/drivers/template_config.py b/backend/api/v1/modules/a76/layouts_csv/drivers/template_config.py index 891dccb1..6c4be901 100644 --- a/backend/api/v1/modules/a76/layouts_csv/drivers/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/drivers/template_config.py @@ -4,6 +4,8 @@ Configuracion de plantilla CSV para Conductores (EstructuraCatConductor.xls). from typing import Dict, List, Any +from ..common.cell_value import cell_to_str + TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { "drivers": [ {"canonical": "TRANSPORTISTA", "aliases": ["TRANSPORTISTA CLAVE", "CLAVE TRANSPORTISTA", "TRANSPORTER"]}, @@ -48,10 +50,10 @@ def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]: def row_from_template(row: Dict[str, Any], normalize_header_fn) -> Dict[str, Any]: lookup = build_normalized_lookup(normalize_header_fn) if not lookup: - return {normalize_header_fn(k): v for k, v in row.items()} + return {normalize_header_fn(k): cell_to_str(v) for k, v in row.items()} out: Dict[str, Any] = {} for csv_header, value in row.items(): key_norm = normalize_header_fn(csv_header) if key_norm in lookup: - out[lookup[key_norm]] = value + out[lookup[key_norm]] = cell_to_str(value) return out diff --git a/backend/api/v1/modules/a76/layouts_csv/exchange_rate/template_config.py b/backend/api/v1/modules/a76/layouts_csv/exchange_rate/template_config.py index c907f9a0..9f32c526 100644 --- a/backend/api/v1/modules/a76/layouts_csv/exchange_rate/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/exchange_rate/template_config.py @@ -5,6 +5,8 @@ Solo se leen columnas definidas aquí; el resto se ignora. from typing import Dict, List, Any, Optional +from ..common.cell_value import cell_to_str + TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { "exchange_rates": [ {"canonical": "FECHA", "aliases": ["FECHA APLICABLE", "DATE", "FECHA TIPO CAMBIO"]}, @@ -33,10 +35,10 @@ def row_from_template(row: Dict[str, Any], normalize_header_fn, template_id: str """Fila CSV con solo columnas de la plantilla, en nombres canónicos.""" lookup = build_normalized_lookup(normalize_header_fn, template_id) if not lookup: - return {normalize_header_fn(k): v for k, v in row.items()} + return {normalize_header_fn(k): cell_to_str(v) for k, v in row.items()} out: Dict[str, Any] = {} for csv_header, value in row.items(): key_norm = normalize_header_fn(csv_header) if key_norm in lookup: - out[lookup[key_norm]] = value + out[lookup[key_norm]] = cell_to_str(value) return out diff --git a/backend/api/v1/modules/a76/layouts_csv/exchange_rate/validators/common.py b/backend/api/v1/modules/a76/layouts_csv/exchange_rate/validators/common.py index 6ba36f2e..314af8c9 100644 --- a/backend/api/v1/modules/a76/layouts_csv/exchange_rate/validators/common.py +++ b/backend/api/v1/modules/a76/layouts_csv/exchange_rate/validators/common.py @@ -5,6 +5,7 @@ validación fecha (longitud, día acorde al mes, mes ≤ 12; sin límite de año """ from typing import Dict, Any, Optional +from ...common.cell_value import cell_to_str from ..common.common_validators import ( check_required_value_positive, check_optional_max_length, @@ -25,12 +26,14 @@ def validate_row_desfase(raw_row: Dict[str, Any], line_num: int) -> Optional[Dic Si la fila tiene 3 o más columnas y la 3ª tiene valor, error de desfase (Clarion ColumnaC <> ''). """ values_ordered = list(raw_row.values()) if raw_row else [] - if len(values_ordered) >= 3 and (values_ordered[2] or "").strip(): - return { - "line": line_num, - "col": "", - "msg": f"{MSG_DESFASE} {MSG_DESFASE_SOLUCION}", - } + if len(values_ordered) >= 3: + cell = cell_to_str(values_ordered[2]) + if cell.strip(): + return { + "line": line_num, + "col": "", + "msg": f"{MSG_DESFASE} {MSG_DESFASE_SOLUCION}", + } return None diff --git a/backend/api/v1/modules/a76/layouts_csv/exportacion/tasks.py b/backend/api/v1/modules/a76/layouts_csv/exportacion/tasks.py index 0d43f5b1..96c41870 100644 --- a/backend/api/v1/modules/a76/layouts_csv/exportacion/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/exportacion/tasks.py @@ -1,7 +1,7 @@ """ Tareas Celery para importación CSV de Exportación (encabezado y partidas). -Flujo: scan_file (sin validaciones) → insert_valid_rows (sin inserción en BD). -Usa layouts_csv.common (storage, normalize, meta, responses, csv_reader). +Para invoice_header: scan_file delega en facturas._do_scan_file (validaciones FK y reporte de errores); +insert_valid_rows delega en facturas._do_insert_valid_rows (inserción en BD). Storage "exp". """ import logging import os @@ -37,10 +37,16 @@ def _norm_row(row: Dict[str, Any], template_id: str) -> Dict[str, Any]: @celery_app.task(bind=True, name="api.v1.modules.a76.layouts_csv.exportacion.tasks.scan_file") def scan_file(self, job_id: str, model_target: str, config: str = None): """ - Scan CSV sin validaciones: leer, normalizar con plantilla, devolver total_rows y 0 errores. + Para invoice_header: delega en facturas._do_scan_file con storage "exp" (validaciones FK, encabezados_expo). + Para invoice_details: scan sin validaciones (stub). """ logger.info("Exportación import: starting scan for job %s target %s", job_id, model_target) + if model_target == "invoice_header": + from api.v1.modules.a76.layouts_csv.facturas.tasks import _do_scan_file + return _do_scan_file(job_id, "invoice_header", config, job_type_override="exp") + + # invoice_details: stub sin validaciones file_path = _ensure_file(job_id) if not file_path: return {"status": "failed", "error": "Archivo no encontrado (expirado o no subido). Sube de nuevo."} @@ -85,10 +91,16 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): @celery_app.task(bind=True, name="api.v1.modules.a76.layouts_csv.exportacion.tasks.insert_valid_rows") def insert_valid_rows(self, job_id: str, model_target: str): """ - Commit sin inserción en BD: leer CSV, omitir líneas de error (vacío por ahora), cleanup, devolver finished con inserted=0. + Commit: para invoice_header delega en facturas (inserción real en BD con storage "exp"). + Para invoice_details mantiene stub (sin inserción). """ logger.info("Exportación import: starting commit for job %s target %s", job_id, model_target) + if model_target == "invoice_header": + from api.v1.modules.a76.layouts_csv.facturas.tasks import _do_insert_valid_rows + return _do_insert_valid_rows(job_id, "invoice_header", job_type_override="exp") + + # invoice_details: stub (sin inserción en BD) file_path = _ensure_file(job_id) if not file_path: alt_path = common_storage.file_path_for_job(JOB_TYPE, job_id) diff --git a/backend/api/v1/modules/a76/layouts_csv/exportacion/template_config.py b/backend/api/v1/modules/a76/layouts_csv/exportacion/template_config.py index d6d598fc..60ce5cb5 100644 --- a/backend/api/v1/modules/a76/layouts_csv/exportacion/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/exportacion/template_config.py @@ -5,6 +5,8 @@ Misma estructura que facturas exp_def_header / exp_def_details; módulo autocont from typing import Dict, List, Any, Optional +from ..common.cell_value import cell_to_str + # Columnas para encabezado y partidas de exportación (EstructuraEncFacExpoCamReg / EstructuraParExpoCamReg) TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { "exp_def_header": [ @@ -80,10 +82,10 @@ def row_from_template(row: Dict[str, Any], template_id: str, normalize_header_fn """Fila CSV -> dict con nombres canónicos de la plantilla.""" lookup = build_normalized_lookup(template_id, normalize_header_fn) if not lookup: - return {normalize_header_fn(k): v for k, v in row.items()} + return {normalize_header_fn(k): cell_to_str(v) for k, v in row.items()} out: Dict[str, Any] = {} for csv_header, value in row.items(): key_norm = normalize_header_fn(csv_header) if key_norm in lookup: - out[lookup[key_norm]] = value + out[lookup[key_norm]] = cell_to_str(value) return out 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 b877c1bc..429a310b 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py @@ -173,20 +173,25 @@ def _validate_customs_broker_ref( return None @celery_app.task(bind=True) -def scan_file(self, job_id: str, model_target: str, config: str = None): - """ - Pass 1: Read CSV, Validate types, Write Errors to JSONL. - File content is loaded from Redis (written by API on upload) so worker does not need shared filesystem. - """ +def scan_file(self, job_id: str, model_target: str, config: str = None, job_type_override: Optional[str] = None): + """Pass 1: Read CSV, Validate types, Write Errors to JSONL. Delegates to _do_scan_file.""" + return _do_scan_file(job_id, model_target, config, job_type_override) + + +def _do_scan_file(job_id: str, model_target: str, config: Optional[str] = None, job_type_override: Optional[str] = None) -> Dict[str, Any]: + """Pass 1 body: load file/meta from storage, run validations, store error lines. Uses effective_job_type for storage.""" + effective_job_type = job_type_override if job_type_override is not None else JOB_TYPE + log_prefix = "Exportación import" if effective_job_type else "Invoices import" + logger.info(f"Starting scan for job {job_id} target {model_target}") # 1. Get file from Redis and write to worker local disk - file_path = _ensure_worker_has_file_from_redis(job_id) + file_path = common_storage.ensure_file_from_redis(effective_job_type, job_id, log_prefix) if not file_path: return {"status": "failed", "error": "File not found (missing or expired in queue). Please upload again."} - _ensure_worker_has_meta_from_redis(job_id, file_path) + common_storage.ensure_meta_from_redis(effective_job_type, job_id, file_path, log_prefix) - error_path = common_storage.error_path_for_job(JOB_TYPE, job_id) + error_path = common_storage.error_path_for_job(effective_job_type, job_id) total_rows = 0 error_count = 0 @@ -217,6 +222,9 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): "imp_temp_header" if model_target == "invoice_header" else "imp_temp_details" if model_target == "invoice_details" else "imp_temp_series" ) + # Cuando el scan viene de Exportación (job_type_override "exp"), forzar exp_def_header para que corran las validaciones FK en el escaneo + if job_type_override == "exp" and model_target == "invoice_header": + template_id = "exp_def_header" inv_type_value = normalize_public_code(footer_config.get("invoice_type") or meta.get("invoice_type") or "TEM") if not inv_type_value: inv_type_value = "TEM" @@ -384,8 +392,7 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): 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) + common_storage.store_error_lines(effective_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 ) @@ -545,8 +552,7 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): 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) + common_storage.store_error_lines(effective_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 ) @@ -674,8 +680,7 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): 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) + common_storage.store_error_lines(effective_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 ) @@ -932,8 +937,7 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): errors_detail.append({"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")}) processed_rows += 1 - if error_lines_list: - common_storage.store_error_lines(JOB_TYPE, job_id, error_lines_list) + common_storage.store_error_lines(effective_job_type, job_id, error_lines_list) return common_responses.scan_result(job_id, processed_rows, error_count, errors_detail) except Exception as e: logger.exception("Partidas import scan failed: %s", e) @@ -1192,8 +1196,7 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): errors_detail.append({"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")}) processed_rows += 1 - if error_lines_list: - common_storage.store_error_lines(JOB_TYPE, job_id, error_lines_list) + common_storage.store_error_lines(effective_job_type, job_id, error_lines_list) return common_responses.scan_result(job_id, processed_rows, error_count, errors_detail) except Exception as e: logger.exception("Partidas importación definitiva scan failed: %s", e) @@ -1451,8 +1454,7 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): errors_detail.append({"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")}) processed_rows += 1 - if error_lines_list: - common_storage.store_error_lines(JOB_TYPE, job_id, error_lines_list) + common_storage.store_error_lines(effective_job_type, job_id, error_lines_list) return common_responses.scan_result(job_id, processed_rows, error_count, errors_detail) except Exception as e: logger.exception("Partidas Compras Mexicanas scan failed: %s", e) @@ -1779,8 +1781,7 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): 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) + common_storage.store_error_lines(effective_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 ) @@ -2114,8 +2115,7 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): 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) + common_storage.store_error_lines(effective_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 ) @@ -2123,6 +2123,365 @@ 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 Exportación (Expo Def) y Cambio de Régimen: flujo específico (Clarion VALIDA_TODA_FAC_EXPO / VALIDA_PARCIAL) --- + if model_target == "invoice_header" and template_id == "exp_def_header": + try: + from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceComplianceMx, InvoiceFinancials + from api.v1.modules.a76.items.models import LineItem + from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos + from api.v1.modules.a76.pedmientos.models.pedimento_dates import PedimentoDates + from api.v1.modules.a76.clients_and_providers.models import ClientProvider + from api.v1.modules.a76.customs_brokers.models import CustomsBroker + from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection + 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 api.v1.modules.a76.manifests.manifest.models import Manifest + from .validators.encabezados_expo import validate_row_encabezados_expo + from .validators.encabezados_impo_temp import _pedimento_key_from_parsed + from .validators.encabezados_impo_def import parse_pedimento_col_a_impo_def + + _fc = parse_footer_config(meta.get("footer_config")) + actualizar = meta.get("actualizar", False) + autonumerar_remesas = meta.get("autonumerar_remesas", False) + recalcular_fecha_pedimentos = meta.get("recalcular_fecha_pedimentos", False) + if _fc: + if "actualizar" in _fc: + actualizar = bool(_fc["actualizar"]) + if "autonumerar_remesas" in _fc: + autonumerar_remesas = bool(_fc["autonumerar_remesas"]) + if "recalcular_fecha_pedimentos" in _fc: + recalcular_fecha_pedimentos = bool(_fc["recalcular_fecha_pedimentos"]) + + cambio_regimen_raw = (meta.get("cambio_regimen") or _fc.get("cambio_regimen") or "NO").strip().upper() + cambio_regimen = cambio_regimen_raw == "SI" + tipo_factura = (meta.get("tipo_factura") or _fc.get("tipo_factura") or "AFIJO").strip().upper() + + TIPOS_FACTURA_EXPO_VALIDOS = frozenset({"NODES", "AFIJO", "DONAC", "SCRAP"}) + if tipo_factura not in TIPOS_FACTURA_EXPO_VALIDOS: + return { + "status": "failed", + "error": f"Tipo de factura '{tipo_factura}' no válido para Exportación. Debe ser uno de: NODES, AFIJO, DONAC, SCRAP.", + } + + if cambio_regimen and tipo_factura != "AFIJO": + return { + "status": "failed", + "error": "Este tipo de factura no es compatible para Cambio de Régimen, seleccionar AFIJO.", + } + + def _ped_key_from_row_expo(ped_str: str): + parsed = parse_pedimento_col_a_impo_def(ped_str) + if not parsed: + return None + return _pedimento_key_from_parsed(parsed[0], parsed[1], parsed[2]) + + with CoreSessionLocal() as session: + q_inv = ( + session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.is_updated, InvoiceHeader.is_updated_rep) + .filter( + InvoiceHeader.tenant_id == tenant_id, + InvoiceHeader.company_id == company_id, + InvoiceHeader.operation_type == "exp", + ) + ) + invoice_exists_by_number = {} + invoice_updated_by_number = {} + invoice_in_report_by_number = {} + for num, iid, is_upd, is_rep in q_inv.all(): + if num: + n = str(num).strip() + invoice_exists_by_number[n] = True + invoice_updated_by_number[n] = bool(is_upd) + invoice_in_report_by_number[n] = bool(is_rep) if is_rep is not None else False + + if cambio_regimen: + ped_filter_op = "imp" + ped_filter_regimes = ["IMD"] + else: + ped_filter_op = "exp" + ped_filter_regimes = ["EXD", "ETE", "ETR"] + + pedimento_data_by_key = {} + for p in ( + session.query( + Pedimentos.id, + Pedimentos.customs_office, + Pedimentos.license, + Pedimentos.pedimento_number, + Pedimentos.operation_type, + Pedimentos.regime, + Pedimentos.pedimento_type, + Pedimentos.pedimento_code, + ) + .filter( + Pedimentos.tenant_id == tenant_id, + Pedimentos.company_id == company_id, + Pedimentos.operation_type == ped_filter_op, + Pedimentos.regime.in_(ped_filter_regimes), + ) + .all() + ): + co = (p.customs_office or "").strip() + lic = (p.license or "").strip() + num = (p.pedimento_number or "").strip() + if not co or not lic or not num: + continue + key = _pedimento_key_from_parsed(co, lic, num) + entry_date = None + end_date = None + pd = ( + session.query(PedimentoDates.entry_date, PedimentoDates.end_date) + .filter(PedimentoDates.pedimento_id == p.id).first() + ) + if pd: + entry_date = pd[0] + end_date = pd[1] + info = { + "id": p.id, + "regime": (p.regime or "").strip(), + "operation_type": (p.operation_type or "").strip().upper()[:3], + "pedimento_type": (p.pedimento_type or "").strip(), + "pedimento_code": (p.pedimento_code or "").strip().upper(), + "entry_date": entry_date, + "end_date": end_date, + } + if key not in pedimento_data_by_key: + pedimento_data_by_key[key] = [] + pedimento_data_by_key[key].append(info) + + remesa_por_pedimento_bd = {} + q_rem = ( + session.query( + InvoiceComplianceMx.remesa, + Pedimentos.customs_office, + Pedimentos.license, + Pedimentos.pedimento_number, + ) + .join(InvoiceHeader, InvoiceHeader.id == InvoiceComplianceMx.invoice_id) + .join(Pedimentos, Pedimentos.id == InvoiceComplianceMx.pedimento_id) + .filter( + InvoiceComplianceMx.tenant_id == tenant_id, + InvoiceComplianceMx.company_id == company_id, + InvoiceComplianceMx.pedimento_id.isnot(None), + InvoiceComplianceMx.remesa.isnot(None), + InvoiceHeader.operation_type == "exp", + ) + ) + for rem, co, lic, num in q_rem.all(): + if co and lic and num and rem is not None: + key = _pedimento_key_from_parsed( + (co or "").strip()[:2], + (lic or "").strip(), + (num or "").strip(), + ) + if key not in remesa_por_pedimento_bd: + remesa_por_pedimento_bd[key] = set() + remesa_por_pedimento_bd[key].add(int(rem)) + + 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_broker_ids = set() + valid_broker_claves = set() + for cb in session.query(CustomsBroker.id, CustomsBroker.broker_key).filter( + CustomsBroker.tenant_id == tenant_id, + CustomsBroker.company_id == company_id, + ).all(): + valid_broker_ids.add(cb[0]) + if cb[1] and str(cb[1]).strip(): + valid_broker_claves.add(str(cb[1]).strip()) + + 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_aduana_codes = set() + for cs in session.query(CustomsSection.customs_code).all(): + if cs[0]: + valid_aduana_codes.add((cs[0] or "").strip()) + + valid_currency_codes = set() + for ct in session.query(CurrencyType.code).all(): + if ct[0]: + valid_currency_codes.add((ct[0] or "").strip().upper()) + + valid_manifiesto_codes = set() + for m in session.query(Manifest.manifest_number).filter( + Manifest.tenant_id == tenant_id, + Manifest.company_id == company_id, + ).all(): + if m[0] and str(m[0]).strip(): + valid_manifiesto_codes.add(str(m[0]).strip()) + + 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 == "exp", + ) + .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 == "exp", + ) + ) + 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] + + date_format = _fc.get("dateFormat") or meta.get("date_format") + + 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) + + remesa_por_pedimento_csv = {} + for row in rows_list: + row_norm = row_from_template(row, "exp_def_header", normalize_header) + ped = (row_norm.get("PEDIMENTO") or "").strip() + rem = row_norm.get("REMESA") + factura = (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or row_norm.get("FACTURA") or "").strip() + if not ped or not factura: + continue + key = _ped_key_from_row_expo(ped) + if not key: + continue + try: + rem_int = int(rem) if rem is not None and str(rem).strip() else None + except (TypeError, ValueError): + rem_int = None + if rem_int is not None: + if key not in remesa_por_pedimento_csv: + remesa_por_pedimento_csv[key] = {} + if rem_int not in remesa_por_pedimento_csv[key]: + remesa_por_pedimento_csv[key][rem_int] = factura + + 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, "exp_def_header", normalize_header) + warnings_row = [] + err = validate_row_encabezados_expo( + row_norm, + i, + actualizar=actualizar, + cambio_regimen=cambio_regimen, + tipo_factura=tipo_factura, + invoice_exists_by_number=invoice_exists_by_number, + invoice_updated_by_number=invoice_updated_by_number, + invoice_in_report_by_number=invoice_in_report_by_number, + pedimento_data_by_key=pedimento_data_by_key, + remesa_por_pedimento_bd=remesa_por_pedimento_bd, + remesa_por_pedimento_csv=remesa_por_pedimento_csv, + valid_provider_ids=valid_provider_ids, + valid_sold_to_ids=valid_sold_to_ids, + valid_shipped_to_ids=valid_shipped_to_ids, + valid_broker_ids=valid_broker_ids, + valid_broker_claves=valid_broker_claves, + valid_transporter_keys=valid_transporter_keys, + valid_incoterms=valid_incoterms, + valid_aduana_codes=valid_aduana_codes, + valid_currency_codes=valid_currency_codes, + 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, + valid_manifiesto_codes=valid_manifiesto_codes, + valid_enviado_por_ids=valid_provider_ids, + valid_enviado_por_short_names=valid_provider_short_names, + 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, + autonumerar_remesas=autonumerar_remesas, + recalcular_fecha_pedimentos=recalcular_fecha_pedimentos, + 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 + + common_storage.store_error_lines(effective_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 exportación 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: @@ -2294,8 +2653,7 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): 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) + common_storage.store_error_lines(effective_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 ) @@ -2402,8 +2760,7 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): ) except Exception: pass - if error_lines_list: - common_storage.store_error_lines(JOB_TYPE, job_id, error_lines_list) + common_storage.store_error_lines(effective_job_type, job_id, error_lines_list) except Exception as e: logger.warning(f"Failed to store error lines in Redis: {e}") @@ -2915,22 +3272,31 @@ def resolve_public_code( return cache[normalized] @celery_app.task(bind=True) -def insert_valid_rows(self, job_id: str, model_target: str): +def insert_valid_rows(self, job_id: str, model_target: str, job_type_override: Optional[str] = None): + """Pass 2: Re-read CSV, Skip Errors, Bulk Insert. Delegates to _do_insert_valid_rows.""" + return _do_insert_valid_rows(job_id, model_target, job_type_override) + + +def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Optional[str] = None) -> Dict[str, Any]: """ Pass 2: Re-read CSV, Skip Errors, Bulk Insert. File and meta are loaded from Redis if present (same as scan_file), so worker does not need shared filesystem. + When job_type_override is set (e.g. "exp" for Exportación), storage keys use that prefix. """ + effective_job_type = job_type_override if job_type_override is not None else JOB_TYPE + log_prefix = "Exportación import" if effective_job_type else "Invoices import" + logger.info(f"Starting Commit for {job_id} target {model_target}") # Ensure we have the file on this worker: prefer Redis (so any worker can run commit) - file_path = _ensure_worker_has_file_from_redis(job_id) + file_path = common_storage.ensure_file_from_redis(effective_job_type, job_id, log_prefix) if not file_path: - alt_path = common_storage.file_path_for_job(JOB_TYPE, job_id) + alt_path = common_storage.file_path_for_job(effective_job_type, job_id) if not os.path.exists(alt_path): return {"status": "failed", "error": "File not found (missing or expired). Please upload and confirm again."} file_path = alt_path else: - _ensure_worker_has_meta_from_redis(job_id, file_path) + common_storage.ensure_meta_from_redis(effective_job_type, job_id, file_path, log_prefix) try: tenant_id, company_id = common_meta.require_tenant_context(file_path) @@ -2939,8 +3305,8 @@ def insert_valid_rows(self, job_id: str, model_target: str): meta = common_meta.load_meta(file_path) meta_path = common_meta.get_meta_path(file_path) - error_path = common_storage.error_path_for_job(JOB_TYPE, job_id) - error_lines = common_storage.get_error_lines(JOB_TYPE, job_id, error_path) + error_path = common_storage.error_path_for_job(effective_job_type, job_id) + error_lines = common_storage.get_error_lines(effective_job_type, job_id, error_path) # Si el upload fue de series (template_id imp_temp_series o imp_def_series), usar flujo series aunque model_target venga mal use_series_flow = ( @@ -3212,7 +3578,7 @@ def insert_valid_rows(self, job_id: str, model_target: str): session.commit() - common_storage.cleanup_import_job(JOB_TYPE, job_id, file_path=file_path, error_path=error_path, meta_path=meta_path) + common_storage.cleanup_import_job(effective_job_type, job_id, file_path=file_path, error_path=error_path, meta_path=meta_path) status = "finished" if (inserted_count + updated_count) > 0 else ("warning" if skipped_invalid else "failed") out = { "status": status, @@ -3437,7 +3803,7 @@ def insert_valid_rows(self, job_id: str, model_target: str): session.commit() - common_storage.cleanup_import_job(JOB_TYPE, job_id, file_path=file_path, error_path=error_path, meta_path=meta_path) + common_storage.cleanup_import_job(effective_job_type, job_id, file_path=file_path, error_path=error_path, meta_path=meta_path) status = "finished" if (inserted_count + updated_count) > 0 else ("warning" if skipped_invalid else "failed") out = { "status": status, @@ -3483,6 +3849,7 @@ def insert_valid_rows(self, job_id: str, model_target: str): _pedimento_key_from_parsed, row_to_transport_type_clarion, ) + from .validators.encabezados_impo_def import parse_pedimento_col_a_impo_def from api.v1.modules.a76.items.models import LineItem from api.v1.modules.a76.items.line_financials.models import LineFinancial @@ -3514,6 +3881,17 @@ def insert_valid_rows(self, job_id: str, model_target: str): inv_type_value = "DEF" if model_target == "invoice_header" and _template_id_insert == "cmex_header": inv_type_value = "MEX" + if model_target == "invoice_header" and _template_id_insert == "exp_def_header": + op_type_value = OperationType("exp") + inv_type_value = normalize_public_code( + meta.get("tipo_factura") or footer_config.get("tipo_factura") or "AFIJO" + ) or "AFIJO" + _es_cambio_regimen = None + if model_target == "invoice_header" and _template_id_insert == "exp_def_header": + cambio_regimen_raw = ( + str(meta.get("cambio_regimen") or footer_config.get("cambio_regimen") or "NO").strip().upper() + ) + _es_cambio_regimen = "S" if cambio_regimen_raw == "SI" else "N" if model_target == "invoice_details" and _template_id_insert == "imp_def_details": inv_type_value = "DEF" if model_target == "invoice_details" and _template_id_insert == "cmex_details": @@ -3544,6 +3922,7 @@ def insert_valid_rows(self, job_id: str, model_target: str): customs_section_cache: Dict[str, Optional[str]] = {} part_cache: Dict[str, Optional[int]] = {} pedimento_id_cache: Dict[str, Optional[int]] = {} + shipped_by_cache: Dict[Any, Optional[int]] = {} _fc_insert = parse_footer_config(meta.get("footer_config")) autonumerar_remesas_insert = _fc_insert.get("autonumerar_remesas", False) class_id_by_code: Dict[str, int] = {} @@ -3578,10 +3957,16 @@ def insert_valid_rows(self, job_id: str, model_target: str): ) for i, row in enumerate(reader, start=1): - if i in error_lines: - continue - row_norm = row_from_template(row, template_id, normalize_header) + if i in error_lines: + skipped_invalid += 1 + inv_for_detail = (row_norm.get('NUMERO FACTURA') or row_norm.get('NUM FACTURA') or row_norm.get('FACTURA') or '').strip() if model_target == 'invoice_header' else "" + skipped_fk_details.append({ + "line": i, + "invoice": inv_for_detail or "(vacío)", + "reason": "Línea marcada con error en el escaneo previo (revisar reporte de validación).", + }) + continue # Mapping Logic (solo campos que acepta el modelo de facturas) if model_target == 'invoice_header': @@ -3590,6 +3975,8 @@ def insert_valid_rows(self, job_id: str, model_target: str): if not invoice_number or not invoice_date: skipped_invalid += 1 + reason = "Número de factura o fecha faltante/inválida" + skipped_fk_details.append({"line": i, "invoice": invoice_number or "(vacío)", "reason": reason}) logger.debug(f"Row {i}: Skipped - missing invoice_number or invalid invoice_date. " f"Invoice: {invoice_number}, Date: {row_norm.get('FECHA FACTURA') or row_norm.get('FECHA')}") continue @@ -3648,21 +4035,6 @@ def insert_valid_rows(self, job_id: str, model_target: str): logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") continue - err = _validate_client_provider_ref( - validator, - ClientProvider, - row_norm.get('CLAVE ENVIADO A'), - i, - "CLAVE ENVIADO A", - required=True, - ) - if err: - 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 - if inv_type_value != "MEX": err = _validate_customs_broker_ref( validator, @@ -3782,7 +4154,8 @@ def insert_valid_rows(self, job_id: str, model_target: str): InvoiceHeader.tenant_id == tenant_id, InvoiceHeader.company_id == company_id, InvoiceHeader.invoice_number == invoice_number, - InvoiceHeader.invoice_type == inv_type_value + InvoiceHeader.invoice_type == inv_type_value, + InvoiceHeader.operation_type == op_type_value, ) .first() ) @@ -3792,13 +4165,16 @@ def insert_valid_rows(self, job_id: str, model_target: str): 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 template_id == "exp_def_header": + parsed = parse_pedimento_col_a_impo_def(ped_str) + else: + parsed = parse_pedimento_col_a(ped_str) if parsed: - customs_office_p, license_p, num_p = parsed + customs_office_p, license_p, num_p = (x.strip() if x else "" for x in parsed) key_p = _pedimento_key_from_parsed(customs_office_p, license_p, num_p) if key_p not in pedimento_id_cache: - co_prefix = (customs_office_p or "").strip()[:2] - ped_row = ( + co_prefix = (customs_office_p or "").strip()[:2].zfill(2) + ped_query = ( session.query(Pedimentos.id) .filter( Pedimentos.tenant_id == tenant_id, @@ -3807,8 +4183,19 @@ def insert_valid_rows(self, job_id: str, model_target: str): Pedimentos.license == license_p, Pedimentos.pedimento_number == num_p, ) - .first() ) + if template_id == "exp_def_header": + if _es_cambio_regimen == "S": + ped_query = ped_query.filter( + func.lower(Pedimentos.operation_type) == "imp", + func.upper(Pedimentos.regime) == "IMD", + ) + else: + ped_query = ped_query.filter( + func.lower(Pedimentos.operation_type) == "exp", + func.upper(Pedimentos.regime).in_(["EXD", "ETE", "ETR"]), + ) + ped_row = ped_query.first() pedimento_id_cache[key_p] = ped_row[0] if ped_row else None pedimento_id = pedimento_id_cache[key_p] if pedimento_id is not None and remesa_val is None and autonumerar_remesas_insert: @@ -3843,7 +4230,6 @@ def insert_valid_rows(self, job_id: str, model_target: str): header.emission_date = parse_date(row_norm.get('FECHA EMISION'), date_format) header.observation_es = (row_norm.get('OBSERVACIONES E') or None) header.observation_en = (row_norm.get('OBSERVACIONES I') or None) - logger.info(f"Row {i}: Updating existing invoice {invoice_number}") # Clean up related data that will be re-inserted/updated @@ -3932,6 +4318,15 @@ def insert_valid_rows(self, job_id: str, model_target: str): ), edocument=(row_norm.get('E DOCUMENT') or None), vucem_operation_num=(row_norm.get('NUM OPERACION') or None), + manifest_number=(row_norm.get('MANIFIESTO') or None), + shipped_by_id=resolve_client_provider_id( + session, + ClientProvider, + row_norm.get('ENVIADO POR'), + tenant_id, + company_id, + shipped_by_cache, + ) if row_norm.get('ENVIADO POR') else None, tenant_id=tenant_id, company_id=company_id, ) @@ -4187,7 +4582,7 @@ def insert_valid_rows(self, job_id: str, model_target: str): "skipped_missing_invoice": skipped_missing_invoice, "skipped_missing_fk": skipped_missing_fk, "skipped_details": skipped_fk_details, - "message": f"No se insertaron registros. {total_skipped} fueron rechazados." + "message": f"No se insertaron registros. {total_skipped} fueron rechazados. Revisa el detalle por línea a continuación.", } else: logger.error(f"No valid records found in CSV for job {job_id}") @@ -4220,7 +4615,7 @@ def insert_valid_rows(self, job_id: str, model_target: str): # 5. Cleanup: remove temp files and Redis keys so data is not kept indefinitely try: common_storage.cleanup_import_job( - JOB_TYPE, job_id, + effective_job_type, job_id, file_path=file_path, error_path=error_path, meta_path=meta_path, 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 8f7981c7..e4c01912 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 @@ -6,6 +6,8 @@ Solo se escribe en BD lo que los modelos de facturas aceptan (respetando models) from typing import Dict, List, Any, Optional +from ..common.cell_value import cell_to_str + # Cada plantilla define sus columnas canónicas y alias (otros nombres que aceptamos en el CSV). # canonical = nombre estándar con el que trabajamos internamente; debe coincidir con lo que # espera la lógica de validación e insert (tasks.py). @@ -82,8 +84,44 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { {"canonical": "OBSERVACIONES E"}, {"canonical": "OBSERVACIONES I"}, ], - # --- Encabezado factura: Expo (EstructuraEncFacExpoCamReg.xls) - misma estructura --- - "exp_def_header": None, + # --- Encabezado factura: Expo (EstructuraEncFacExpoCamReg.xls) - Clarion A-AE --- + # PEDIMENTO, REMESA, NUMERO FACTURA, FECHA FACTURA, TIPO DE CAMBIO, REGIMEN, CLAVE PROVEEDOR, + # CLAVE VENDIDO A:, CLAVE ENVIADO A, AGENTE ADUANAL, ... MANIFIESTO, E-DOCUMENT, NUM. OPERACION, + # ENVIADO POR, ADUANA DE CRUCE, OBSERVACIONES E, OBSERVACIONES I, FACTURA ALTERNA + "exp_def_header": [ + {"canonical": "PEDIMENTO", "aliases": ["NUMERO PEDIMENTO", "PEDIMENTO NUMERO", "NUMERO DE PEDIMENTO", "PED"]}, + {"canonical": "REMESA"}, + {"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA", "ID"]}, + {"canonical": "FECHA FACTURA", "aliases": ["FECHA"]}, + {"canonical": "TIPO DE CAMBIO"}, + {"canonical": "REGIMEN", "aliases": ["CLAVEDOCUMENTO"]}, + {"canonical": "CLAVE PROVEEDOR"}, + {"canonical": "CLAVE VENDIDO A", "aliases": ["CLAVE VENDIDO A:"]}, + {"canonical": "CLAVE ENVIADO A"}, + {"canonical": "AGENTE ADUANAL"}, + {"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": "TIPO PESO"}, + {"canonical": "MANIFIESTO"}, + {"canonical": "E DOCUMENT", "aliases": ["E-DOCUMENT", "EDOCUMENT", "E DOCUMENT"]}, + {"canonical": "NUM OPERACION", "aliases": ["NUM. OPERACION", "NUMOPERACION", "NUM OPERACION"]}, + {"canonical": "ENVIADO POR"}, + {"canonical": "ADUANA DE CRUCE"}, + {"canonical": "OBSERVACIONES E"}, + {"canonical": "OBSERVACIONES I"}, + {"canonical": "FACTURA ALTERNA"}, + ], # --- 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 @@ -182,7 +220,7 @@ def _resolve_template_columns(template_id: str) -> Optional[List[Dict[str, Any]] cols = TEMPLATE_COLUMNS.get(template_id) if cols is not None: return cols - if template_id in ("imp_def_header", "exp_def_header"): + if template_id == "imp_def_header": return TEMPLATE_COLUMNS.get("imp_temp_header") if template_id in ("imp_def_details", "exp_def_details"): return TEMPLATE_COLUMNS.get("imp_temp_details") @@ -218,11 +256,10 @@ def row_from_template(row: Dict[str, Any], template_id: str, normalize_header_fn """ lookup = build_normalized_lookup(template_id, normalize_header_fn) if not lookup: - # Sin template definido: comportamiento legacy (normalizar todo) - return {normalize_header_fn(k): v for k, v in row.items()} + return {normalize_header_fn(k): cell_to_str(v) for k, v in row.items()} out: Dict[str, Any] = {} for csv_header, value in row.items(): key_norm = normalize_header_fn(csv_header) if key_norm in lookup: - out[lookup[key_norm]] = value + out[lookup[key_norm]] = cell_to_str(value) return out 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 5ef9ed99..d244c132 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 @@ -10,6 +10,7 @@ from .encabezados_impo_def import ( parse_pedimento_col_a_impo_def, ) from .encabezados_cmex import validate_row_encabezados_cmex +from .encabezados_expo import validate_row_encabezados_expo from .partidas_impo_def import validate_row_partidas_impo_def from .series_impo_def import ( validate_row_series_impo_def, @@ -20,6 +21,7 @@ __all__ = [ "validate_row_encabezados_impo_temp", "validate_row_encabezados_impo_def", "validate_row_encabezados_cmex", + "validate_row_encabezados_expo", "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_expo.py b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/encabezados_expo.py new file mode 100644 index 00000000..f9c698be --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/encabezados_expo.py @@ -0,0 +1,464 @@ +""" +Validaciones CSV para Encabezados de Facturas de Exportación (Expo Def) y Cambio de Régimen. +Paridad Clarion: VALIDA_TODA_FAC_EXPO, VALIDA_PARCIAL_FAC_EXPO, VALIDACIONES_FAC_EXPO. +Estructura CSV: PEDIMENTO (A), REMESA (B), NUMERO FACTURA (C), ... MANIFIESTO (Y), E-DOCUMENT (Z), +NUM. OPERACION (AA), ENVIADO POR (AB), ADUANA DE CRUCE (AC), OBSERVACIONES E/I, FACTURA ALTERNA. +""" +from datetime import datetime +from typing import Any, Dict, List, Optional, Set + +from .encabezados_impo_temp import ( + _clip, + _err, + _get, + _parse_int, + _pedimento_key_from_parsed, + _validaciones_catalogos, + _validaciones_factura_longitud, + _validaciones_moneda, + _validaciones_tipo_cambio, + _validaciones_tipo_peso, + _validaciones_transporte, +) +from .encabezados_impo_def import ( + parse_pedimento_col_a_impo_def, +) + +# Expo: pedimento mismo formato ##-####-####### (15 chars) +MAX_LEN_PEDIMENTO_EXPO = 15 +MAX_LEN_FACTURA = 15 + +# Regímenes: Exportación (sin cambio de régimen) vs Cambio de Régimen (IMD) +REGIMENES_EXPO = frozenset({"EXD", "ETE", "ETR"}) +REGIMEN_IMD = "IMD" +CLAVES_PEDIMENTO_CAMBIO_REGIMEN = frozenset({"F5", "A3"}) + + +def _validaciones_obligatorios_toda_expo( + row: Dict[str, Any], + line_num: int, + tiene_pedimento: bool, +) -> Optional[Dict[str, Any]]: + """Obligatorios VALIDA_TODA para Expo: C, D, F, G, H, I, J; AC (Aduana de Cruce) si hay pedimento.""" + obligatorios: List[str] = [] + if not _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA"): + obligatorios.append("(Col.C) Número de Factura") + if not _get(row, "FECHA FACTURA", "FECHA"): + obligatorios.append("(Col.D) Fecha de la Factura") + if not _get(row, "REGIMEN", "CLAVEDOCUMENTO"): + obligatorios.append("(Col.F) Régimen") + 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 not _get(row, "AGENTE ADUANAL"): + obligatorios.append("(Col.J) Clave del Agente Aduanal") + if tiene_pedimento and not _get(row, "ADUANA DE CRUCE"): + obligatorios.append("(Col.AC) Aduana de Cruce") + if obligatorios: + return _err( + line_num, + "ARCHIVO CSV", + f"Existen campos vacíos que son obligatorios: {', '.join(obligatorios)}. Revisar para Exportación.", + ) + return None + + +def _validaciones_regimen_expo( + row: Dict[str, Any], + line_num: int, + cambio_regimen: bool, +) -> Optional[Dict[str, Any]]: + """Col F: si cambio_regimen → IMD; si no → EXD, ETR, ETE.""" + f = _get(row, "REGIMEN", "CLAVEDOCUMENTO") + if not f: + return None + f_upper = f.upper() + if cambio_regimen: + if f_upper != REGIMEN_IMD: + return _err( + line_num, + "REGIMEN", + f"Error: (Celda F{line_num}) El Régimen Aduanero: {f} no es válido para este tipo de movimiento. " + "Los válidos para Cambio de Régimen son: IMD.", + ) + else: + if f_upper not in REGIMENES_EXPO: + return _err( + line_num, + "REGIMEN", + f"Error: (Celda F{line_num}) El Régimen Aduanero: {f} no es válido para este tipo de movimiento. " + "Los válidos para Exportación son: EXD, ETE y ETR.", + ) + return None + + +def _validaciones_pedimento_remesa_expo( + row: Dict[str, Any], + line_num: int, + cambio_regimen: bool, + pedimento_data_by_key: Dict[str, List[Dict[str, Any]]], + remesa_por_pedimento_bd: Dict[str, Set[int]], + remesa_por_pedimento_csv: Dict[str, Dict[int, str]], + autonumerar_remesas: bool, + invoice_date_parsed: Optional[datetime], + recalcular_fecha_pedimentos: bool = False, +) -> Optional[Dict[str, Any]]: + """Pedimento Col A: formato ##-####-#######. Si cambio_regimen: tipo I, régimen IMD, ClavePed F5/A3. Si no: tipo E, régimen EXD/ETE/ETR. Remesa igual que imp_def.""" + col_a = _get(row, "PEDIMENTO") + col_b_raw = row.get("REMESA") + col_b = _clip(col_b_raw) + col_f = _get(row, "REGIMEN", "CLAVEDOCUMENTO").upper() + + if not col_a: + if col_b: + return _err( + line_num, + "REMESA", + f"Error: (Celda B{line_num}) Está asignado el número de Remesa y no se tiene un pedimento en (Celda A{line_num}).", + ) + return None + + if len(col_a) > MAX_LEN_PEDIMENTO_EXPO: + return _err( + line_num, + "PEDIMENTO", + f"Error: (Celda A{line_num}) El Pedimento: {col_a} supera la longitud de caracteres. Use formato ##-####-#######.", + ) + parsed = parse_pedimento_col_a_impo_def(col_a) + if not parsed: + return _err( + line_num, + "PEDIMENTO", + f"Error: (Celda A{line_num}) El Formato del Pedimento: {col_a} es incorrecto. Use ##-####-#######.", + ) + + customs_office, license_val, pedimento_number = parsed + key = _pedimento_key_from_parsed(customs_office, license_val, pedimento_number) + ped_info_list = pedimento_data_by_key.get(key) + if not ped_info_list: + return _err( + line_num, + "PEDIMENTO", + f"Error: (Celda A{line_num}) El Número de Pedimento: {col_a} no existe en el Catálogo de Pedimentos. " + + ( + "Darlo de alta como pedimento de Importación Definitiva (Cambio de Régimen)." + if cambio_regimen + else "Darlo de alta como pedimento de Exportación." + ), + ) + + ped_info = ped_info_list[0] + op_type = (ped_info.get("operation_type") or "").strip().upper() + regimen_ped = (ped_info.get("regime") or "").strip().upper() + pedimento_code = (ped_info.get("pedimento_code") or "").strip().upper() + + if cambio_regimen: + if op_type != "IMP" and op_type != "I": + return _err( + line_num, + "PEDIMENTO", + f"Error: (Celda A{line_num}) Este Número de Pedimento: {col_a} no está marcado como Importación Definitiva.", + ) + if regimen_ped != REGIMEN_IMD: + return _err( + line_num, + "PEDIMENTO", + f"Error: (Celda A{line_num}) El Número de Pedimento: {col_a} tiene el Régimen {regimen_ped}, no válido para Cambio de Régimen. Válidos: IMD.", + ) + if col_f and col_f != REGIMEN_IMD: + return _err( + line_num, + "REGIMEN", + f"Error: (Celda F{line_num}) El Régimen Aduanero: {col_f} debe ser IMD para Cambio de Régimen.", + ) + if pedimento_code and pedimento_code not in CLAVES_PEDIMENTO_CAMBIO_REGIMEN: + return _err( + line_num, + "PEDIMENTO", + f"Error: (Celda A{line_num}) El Número de Pedimento: {col_a} tiene la clave {pedimento_code}, no definida para Cambio de Régimen/Regularización. Use F5 o A3.", + ) + else: + if op_type != "EXP" and op_type != "E": + return _err( + line_num, + "PEDIMENTO", + f"Error: (Celda A{line_num}) Este Número de Pedimento: {col_a} no está marcado como Exportación.", + ) + if regimen_ped not in REGIMENES_EXPO: + return _err( + line_num, + "PEDIMENTO", + f"Error: (Celda A{line_num}) El Número de Pedimento: {col_a} tiene el Régimen {regimen_ped}, no válido para Exportación. Válidos: EXD, ETE, ETR.", + ) + if col_f and col_f != regimen_ped: + return _err( + line_num, + "REGIMEN", + f"Error: (Celda F{line_num}) El Régimen Aduanero: {col_f} capturado es diferente al del Pedimento: {regimen_ped}.", + ) + + # Rango de fechas si pedimento consolidado (omitir si recalcular_fecha_pedimentos = True, paridad Clarion) + if not recalcular_fecha_pedimentos: + pedimento_type = (ped_info.get("pedimento_type") or "").strip().lower() + if pedimento_type == "consolidated" and invoice_date_parsed and ped_info.get("entry_date") and ped_info.get("end_date"): + entry = ped_info["entry_date"] + end = ped_info["end_date"] + if hasattr(entry, "date"): + entry = entry.date() + if hasattr(end, "date"): + end = end.date() + inv_d = invoice_date_parsed.date() if hasattr(invoice_date_parsed, "date") else invoice_date_parsed + if inv_d < entry or inv_d > end: + return _err( + line_num, + "FECHA FACTURA", + f"Error: (Celda D{line_num} y A{line_num}) La Fecha de la Factura no corresponde al rango de fechas del Pedimento {col_a}.", + ) + + if not autonumerar_remesas and not col_b: + return _err( + line_num, + "REMESA", + f"Error: (Celda B{line_num}) El Número de Remesa está vacío y se tiene un Pedimento en la Celda A{line_num}.", + ) + remesa_int = _parse_int(col_b_raw) + if col_b and remesa_int is not None and remesa_int == 0: + return _err( + line_num, + "REMESA", + f"Error: (Celda B{line_num}) El Número de Remesa no puede ser 0.", + ) + + factura_actual = _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA") + if remesa_int is not None and key in remesa_por_pedimento_csv: + other = remesa_por_pedimento_csv[key].get(remesa_int) + if other and other != factura_actual: + return _err( + line_num, + "REMESA", + f"Error: (Celda B{line_num}) El Número de Remesa ya está asignado a la factura {other} en este archivo CSV.", + ) + return None + + +def _validaciones_manifiesto( + row: Dict[str, Any], + line_num: int, + valid_manifiesto_codes: Set[str], +) -> Optional[Dict[str, Any]]: + """Col Y MANIFIESTO: si viene informado, debe existir en catálogo.""" + y = _get(row, "MANIFIESTO") + if not y: + return None + if valid_manifiesto_codes and y.strip() not in valid_manifiesto_codes: + return _err( + line_num, + "MANIFIESTO", + f"Error: (Celda Y{line_num}) El Número de Manifiesto: {y} no está dado de alta en el Catálogo de Manifiestos.", + ) + return None + + +def _validaciones_enviado_por( + row: Dict[str, Any], + line_num: int, + valid_enviado_por_ids: Set[int], + valid_enviado_por_short_names: Set[str], +) -> Optional[Dict[str, Any]]: + """Col AB ENVIADO POR: Cliente/Proveedor o equivalente.""" + ab = row.get("ENVIADO POR") + if ab is None or str(ab).strip() == "": + return None + v = _parse_int(ab) + if v is not None: + if valid_enviado_por_ids and v not in valid_enviado_por_ids: + return _err( + line_num, + "ENVIADO POR", + f"Error: (Celda AB{line_num}) La Clave del Enviado Por: {ab} no existe en el Catálogo de Clientes/Proveedores o Equivalentes.", + ) + return None + sn_norm = str(ab).strip().upper() + if valid_enviado_por_short_names and sn_norm not in valid_enviado_por_short_names: + return _err( + line_num, + "ENVIADO POR", + f"Error: (Celda AB{line_num}) La Clave del Enviado Por: {ab} no existe en el Catálogo de Clientes/Proveedores o Equivalentes.", + ) + return None + + +def _normalize_tipo_transporte_ferro(row: Dict[str, Any]) -> Dict[str, Any]: + """Clarion acepta 'FERRO BARCAZA'. Normaliza a FERROBARCAZA para reutilizar validación TEM.""" + out = dict(row) + m = out.get("TIPO TRANSPORTE") + if m is not None and str(m).strip().upper().replace(" ", "") == "FERROBARCAZA": + out["TIPO TRANSPORTE"] = "FERROBARCAZA" + return out + + +def validate_row_encabezados_expo( + row: Dict[str, Any], + line_num: int, + actualizar: bool, + cambio_regimen: bool, + tipo_factura: str, + invoice_exists_by_number: Dict[str, bool], + invoice_updated_by_number: Dict[str, bool], + remesa_por_pedimento_bd: Dict[str, Set[int]], + remesa_por_pedimento_csv: Dict[str, Dict[int, str]], + valid_provider_ids: Set[int], + valid_sold_to_ids: Set[int], + valid_shipped_to_ids: Set[int], + valid_broker_ids: Set[int], + valid_transporter_keys: Set[str], + valid_incoterms: Set[str], + valid_aduana_codes: Set[str], + valid_currency_codes: Set[str], + invoice_in_report_by_number: Optional[Dict[str, bool]] = None, + pedimento_data_by_key: Optional[Dict[str, List[Dict[str, Any]]]] = 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, + valid_broker_claves: Optional[Set[str]] = None, + valid_manifiesto_codes: Optional[Set[str]] = None, + valid_enviado_por_ids: Optional[Set[int]] = None, + valid_enviado_por_short_names: Optional[Set[str]] = None, + 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, + autonumerar_remesas: bool = False, + recalcular_fecha_pedimentos: bool = False, + 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 Exportación (Expo Def) o Cambio de Régimen. + Clarion: VALIDA_TODA_FAC_EXPO vs VALIDA_PARCIAL_FAC_EXPO. + """ + 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", + f"Error: (Celda C{line_num}) El Número de Factura: {factura} ya existe y está Actualizada, no se puede hacer cambios.", + ) + + if actualizar and factura.strip() not in invoice_exists_by_number: + return _err( + line_num, + "NUMERO FACTURA", + "Error: (Col.C) Factura de Exportación no existe (modo Actualizar).", + ) + + if actualizar and (invoice_in_report_by_number or {}).get(factura.strip(), False): + return _err( + line_num, + "NUMERO FACTURA", + "Error: (Col.C) Factura de Exportación Rep. La factura está en reporte y no se puede actualizar.", + ) + + use_partial = actualizar and invoice_exists_by_number.get(factura.strip(), False) + tiene_pedimento = bool(_get(row, "PEDIMENTO")) + + if not use_partial: + err = _validaciones_obligatorios_toda_expo(row, line_num, tiene_pedimento) + 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_pedimento_remesa_expo( + row, + line_num, + cambio_regimen, + pedimento_data_by_key or {}, + remesa_por_pedimento_bd, + remesa_por_pedimento_csv, + autonumerar_remesas, + invoice_date_parsed, + recalcular_fecha_pedimentos, + ) + if err: + return err + + err = _validaciones_factura_longitud(row, line_num) + if err: + return err + err = _validaciones_regimen_expo(row, line_num, cambio_regimen) + if err: + return err + + row_transport = _normalize_tipo_transporte_ferro(row) + err = _validaciones_transporte(row_transport, 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( + 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_broker_ids or set(), + valid_broker_claves or set(), + valid_transporter_keys or set(), + valid_incoterms or set(), + valid_aduana_codes or set(), + ) + if err: + return err + + err = _validaciones_manifiesto(row, line_num, valid_manifiesto_codes or set()) + if err: + return err + + err = _validaciones_enviado_por( + row, + line_num, + valid_enviado_por_ids or set(), + valid_enviado_por_short_names or set(), + ) + if err: + return err + + 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/backend/api/v1/modules/a76/layouts_csv/parts/template_config.py b/backend/api/v1/modules/a76/layouts_csv/parts/template_config.py index 5a5e9da7..2993f180 100644 --- a/backend/api/v1/modules/a76/layouts_csv/parts/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/parts/template_config.py @@ -6,6 +6,8 @@ import csv import io from typing import Dict, List, Any, Optional, Tuple +from ..common.cell_value import cell_to_str + # Valores que indican que la primera fila es cabecera (primera columna normalizada) FIRST_COLUMN_HEADER_VALUES = ("NUMERO DE PARTE", "NUMPARTE") @@ -133,10 +135,10 @@ def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]: def row_from_template(row: Dict[str, Any], normalize_header_fn) -> Dict[str, Any]: lookup = build_normalized_lookup(normalize_header_fn) if not lookup: - return {normalize_header_fn(k): v for k, v in row.items()} + return {normalize_header_fn(k): cell_to_str(v) for k, v in row.items()} out: Dict[str, Any] = {} for csv_header, value in row.items(): key_norm = normalize_header_fn(csv_header) if key_norm in lookup: - out[lookup[key_norm]] = value + out[lookup[key_norm]] = cell_to_str(value) return out diff --git a/backend/api/v1/modules/a76/layouts_csv/pedmientos/template_config.py b/backend/api/v1/modules/a76/layouts_csv/pedmientos/template_config.py index ffea72b9..b69bd5c2 100644 --- a/backend/api/v1/modules/a76/layouts_csv/pedmientos/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/pedmientos/template_config.py @@ -7,6 +7,11 @@ import csv import io from typing import Dict, List, Any, Optional, Tuple +# Convierte valor de celda a str; si es lista (p. ej. CSV con columnas duplicadas), toma el primer elemento. +# Re-exportado desde common para uso en validators; ver layouts_csv.common.cell_value. +from ..common.cell_value import cell_to_str as _cell_to_str + + # Longitudes para validación (sin afectar modelos) AÑO_LEN = 2 PATENTE_LEN = 4 @@ -171,15 +176,15 @@ def build_normalized_lookup(normalize_header_fn, template_id: str = "pedimentos" def row_from_template(row: Dict[str, Any], normalize_header_fn, template_id: str = "pedimentos") -> Dict[str, Any]: - """Fila CSV con solo columnas de la plantilla, en nombres canónicos.""" + """Fila CSV con solo columnas de la plantilla, en nombres canónicos. Valores siempre str (listas convertidas).""" lookup = build_normalized_lookup(normalize_header_fn, template_id) if not lookup: - return {normalize_header_fn(k): v for k, v in row.items()} + return {normalize_header_fn(k): _cell_to_str(v) for k, v in row.items()} out: Dict[str, Any] = {} for csv_header, value in row.items(): key_norm = normalize_header_fn(csv_header) if key_norm in lookup: - out[lookup[key_norm]] = value + out[lookup[key_norm]] = _cell_to_str(value) return out diff --git a/backend/api/v1/modules/a76/layouts_csv/pedmientos/validators/common.py b/backend/api/v1/modules/a76/layouts_csv/pedmientos/validators/common.py index e1311509..a8a06874 100644 --- a/backend/api/v1/modules/a76/layouts_csv/pedmientos/validators/common.py +++ b/backend/api/v1/modules/a76/layouts_csv/pedmientos/validators/common.py @@ -12,6 +12,7 @@ from ..template_config import ( PEDIMENTO_LEN, PEDIMENTO_DASH_POSITIONS, parse_pedimento_col_a, + _cell_to_str, ) from ..common.common_validators import ( check_required_max, @@ -79,12 +80,14 @@ def validate_row_desfase_pedimento( """Si la fila tiene al menos 16 columnas y la 16ª (Col N, IEPS/desfase) tiene valor, error de desfase. Orden: AÑO,PATENTE,NUMERO,TIPO,...,IEPS en índice 15.""" values_ordered = list(raw_row.values()) if raw_row else [] desfase_idx = 15 # IEPS en PEDIMENTOS_TEMPLATE_ORDER (tras AÑO,PATENTE,NUMERO + 12 columnas más) - if len(values_ordered) >= (desfase_idx + 1) and (values_ordered[desfase_idx] or "").strip(): - return { - "line": line_num, - "col": "", - "msg": f"{MSG_DESFASE} {MSG_DESFASE_SOLUCION}", - } + if len(values_ordered) >= (desfase_idx + 1): + cell = _cell_to_str(values_ordered[desfase_idx]) + if cell.strip(): + return { + "line": line_num, + "col": "", + "msg": f"{MSG_DESFASE} {MSG_DESFASE_SOLUCION}", + } return None diff --git a/backend/api/v1/modules/a76/layouts_csv/trailers/template_config.py b/backend/api/v1/modules/a76/layouts_csv/trailers/template_config.py index 31c35f12..ee2ea29e 100644 --- a/backend/api/v1/modules/a76/layouts_csv/trailers/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/trailers/template_config.py @@ -5,6 +5,8 @@ Mapeo: NUMERO TRAILER → trailer_number, CLAVE ACE → ace_trailer_number, etc. from typing import Dict, List, Any +from ..common.cell_value import cell_to_str + TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { "trailers": [ {"canonical": "NUMERO TRAILER", "aliases": ["CLAVE TRAILER", "TRAILER NUMBER", "TRAILER", "NUMERO"]}, @@ -37,10 +39,10 @@ def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]: def row_from_template(row: Dict[str, Any], normalize_header_fn) -> Dict[str, Any]: lookup = build_normalized_lookup(normalize_header_fn) if not lookup: - return {normalize_header_fn(k): v for k, v in row.items()} + return {normalize_header_fn(k): cell_to_str(v) for k, v in row.items()} out: Dict[str, Any] = {} for csv_header, value in row.items(): key_norm = normalize_header_fn(csv_header) if key_norm in lookup: - out[lookup[key_norm]] = value + out[lookup[key_norm]] = cell_to_str(value) return out diff --git a/backend/api/v1/modules/a76/layouts_csv/transportistas/template_config.py b/backend/api/v1/modules/a76/layouts_csv/transportistas/template_config.py index d2bf4418..41d80821 100644 --- a/backend/api/v1/modules/a76/layouts_csv/transportistas/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/transportistas/template_config.py @@ -5,6 +5,8 @@ Mapeo Clarion: Col A = CLAVE TRANSPORTISTA, B = NOMBRE, ... R = DIRECTORIO FTP, from typing import Dict, List, Any +from ..common.cell_value import cell_to_str + TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { "transporters": [ {"canonical": "CLAVE TRANSPORTISTA", "aliases": ["TRANSPORTISTA", "CLAVE TRANS", "CARRIER KEY"]}, @@ -46,10 +48,10 @@ def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]: def row_from_template(row: Dict[str, Any], normalize_header_fn) -> Dict[str, Any]: lookup = build_normalized_lookup(normalize_header_fn) if not lookup: - return {normalize_header_fn(k): v for k, v in row.items()} + return {normalize_header_fn(k): cell_to_str(v) for k, v in row.items()} out: Dict[str, Any] = {} for csv_header, value in row.items(): key_norm = normalize_header_fn(csv_header) if key_norm in lookup: - out[lookup[key_norm]] = value + out[lookup[key_norm]] = cell_to_str(value) return out diff --git a/backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/template_config.py b/backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/template_config.py index 8a34a31e..76106678 100644 --- a/backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/template_config.py @@ -5,6 +5,8 @@ Solo se leen columnas definidas aquí; el resto se ignora. from typing import Dict, List, Any, Optional +from ..common.cell_value import cell_to_str as _cell_to_str + TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { "us_tariff_fractions": [ {"canonical": "FRACCION_ARANCELARIA", "aliases": ["FRACCION ARANCELARIA", "CODE", "FRACCION"]}, @@ -40,13 +42,13 @@ def build_normalized_lookup(normalize_header_fn, template_id: str = "us_tariff_f def row_from_template( row: Dict[str, Any], normalize_header_fn, template_id: str = "us_tariff_fractions" ) -> Dict[str, Any]: - """Fila CSV con solo columnas de la plantilla, en nombres canónicos.""" + """Fila CSV con solo columnas de la plantilla, en nombres canónicos. Valores siempre str (listas convertidas).""" lookup = build_normalized_lookup(normalize_header_fn, template_id) if not lookup: - return {normalize_header_fn(k): v for k, v in row.items()} + return {normalize_header_fn(k): _cell_to_str(v) for k, v in row.items()} out: Dict[str, Any] = {} for csv_header, value in row.items(): key_norm = normalize_header_fn(csv_header) if key_norm in lookup: - out[lookup[key_norm]] = value + out[lookup[key_norm]] = _cell_to_str(value) return out diff --git a/backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/validators/common.py b/backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/validators/common.py index e0c20c6e..4734240b 100644 --- a/backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/validators/common.py +++ b/backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/validators/common.py @@ -5,6 +5,7 @@ Paridad Clarion: desfase Col H, VALIDA_TODA_FRACCIONAME / VALIDA_PARCIAL_FRACCIO from typing import Dict, Any, Optional, Set from ..template_config import DESFASE_COLUMN_INDEX +from ...common.cell_value import cell_to_str from ..common.common_validators import ( normalize_code, check_optional_max_length, @@ -47,7 +48,8 @@ def validate_row_desfase_fa( values_ordered = list(raw_row.values()) if len(values_ordered) <= DESFASE_COLUMN_INDEX: return None - if not (values_ordered[DESFASE_COLUMN_INDEX] or "").strip(): + cell = cell_to_str(values_ordered[DESFASE_COLUMN_INDEX]) + if not cell.strip(): return None return { "line": line_num, diff --git a/backend/api/v1/modules/a76/layouts_csv/vehicles/template_config.py b/backend/api/v1/modules/a76/layouts_csv/vehicles/template_config.py index 0f67ef69..248d35e8 100644 --- a/backend/api/v1/modules/a76/layouts_csv/vehicles/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/vehicles/template_config.py @@ -5,6 +5,8 @@ Mapeo: CLAVE → vehicle_key, CLAVE ACE → ace_vehicle_key, etc. from typing import Dict, List, Any +from ..common.cell_value import cell_to_str + TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { "vehicles": [ {"canonical": "CLAVE", "aliases": ["CLAVE VEHICULO", "VEHICLE KEY", "KEY"]}, @@ -45,10 +47,10 @@ def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]: def row_from_template(row: Dict[str, Any], normalize_header_fn) -> Dict[str, Any]: lookup = build_normalized_lookup(normalize_header_fn) if not lookup: - return {normalize_header_fn(k): v for k, v in row.items()} + return {normalize_header_fn(k): cell_to_str(v) for k, v in row.items()} out: Dict[str, Any] = {} for csv_header, value in row.items(): key_norm = normalize_header_fn(csv_header) if key_norm in lookup: - out[lookup[key_norm]] = value + out[lookup[key_norm]] = cell_to_str(value) return out