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

feature/clarion-expo-csv-header

Browse Source
This commit is contained in:
hreyes
2026-03-10 10:31:30 -06:00
parent 423d0ac3fd
commit a1c3b97375
23 changed files with 1086 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
backend/api/v1/modules/a76/layouts_csv/boms/template_config.py
View File

@@ -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

6
backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/template_config.py
View File

@@ -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

9
backend/api/v1/modules/a76/layouts_csv/classes/template_config.py
View File

@@ -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

6
backend/api/v1/modules/a76/layouts_csv/clients_and_providers/template_config.py
View File

@@ -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

19
backend/api/v1/modules/a76/layouts_csv/common/cell_value.py Normal file
View File

@@ -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)

6
backend/api/v1/modules/a76/layouts_csv/customs_brokers/template_config.py
View File

@@ -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

6
backend/api/v1/modules/a76/layouts_csv/drivers/template_config.py
View File

@@ -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

6
backend/api/v1/modules/a76/layouts_csv/exchange_rate/template_config.py
View File

@@ -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

15
backend/api/v1/modules/a76/layouts_csv/exchange_rate/validators/common.py
View File

@@ -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

20
backend/api/v1/modules/a76/layouts_csv/exportacion/tasks.py
View File

@@ -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)

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

@@ -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

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

@@ -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,

49
backend/api/v1/modules/a76/layouts_csv/facturas/template_config.py
View File

@@ -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

2
backend/api/v1/modules/a76/layouts_csv/facturas/validators/__init__.py
View File

@@ -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",

464
backend/api/v1/modules/a76/layouts_csv/facturas/validators/encabezados_expo.py Normal file
View File

@@ -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

6
backend/api/v1/modules/a76/layouts_csv/parts/template_config.py
View File

@@ -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

11
backend/api/v1/modules/a76/layouts_csv/pedmientos/template_config.py
View File

@@ -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

15
backend/api/v1/modules/a76/layouts_csv/pedmientos/validators/common.py
View File

@@ -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

6
backend/api/v1/modules/a76/layouts_csv/trailers/template_config.py
View File

@@ -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

6
backend/api/v1/modules/a76/layouts_csv/transportistas/template_config.py
View File

@@ -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

8
backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/template_config.py
View File

@@ -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

4
backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/validators/common.py
View File

@@ -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,

6
backend/api/v1/modules/a76/layouts_csv/vehicles/template_config.py
View File

@@ -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