From 77bbcff223b7b7c19e8c2e77a61be97e1813ca6a Mon Sep 17 00:00:00 2001 From: hreyes Date: Thu, 5 Mar 2026 15:54:58 -0700 Subject: [PATCH] feature/validaciones-clarion-agentes-aduanales --- .../test_client_and_provider.py | 40 ---- .../clients_and_providers/test_validators.py | 143 -------------- .../a76/layouts_csv/common/csv_reader.py | 43 ++++- .../customs_brokers/common/__init__.py | 2 +- .../common/common_validators.py | 178 +++++++++++++++++- .../customs_brokers/common/fk_loader.py | 26 +++ .../customs_brokers/common/mappers.py | 9 +- .../a76/layouts_csv/customs_brokers/tasks.py | 65 ++++++- .../customs_brokers/template_config.py | 73 +++++-- .../customs_brokers/validators/common.py | 113 ++++++++++- 10 files changed, 470 insertions(+), 222 deletions(-) delete mode 100644 backend/api/v1/modules/a76/clients_and_providers/test_client_and_provider.py delete mode 100644 backend/api/v1/modules/a76/layouts_csv/clients_and_providers/test_validators.py create mode 100644 backend/api/v1/modules/a76/layouts_csv/customs_brokers/common/fk_loader.py diff --git a/backend/api/v1/modules/a76/clients_and_providers/test_client_and_provider.py b/backend/api/v1/modules/a76/clients_and_providers/test_client_and_provider.py deleted file mode 100644 index ba8afa4a..00000000 --- a/backend/api/v1/modules/a76/clients_and_providers/test_client_and_provider.py +++ /dev/null @@ -1,40 +0,0 @@ -import pytest -from fastapi import FastAPI -from fastapi.testclient import TestClient - -from .routes import router - -app = FastAPI() -app.include_router(router) -client = TestClient(app) - - -@pytest.mark.usefixtures("client", "access_token") -def test_list_clients_and_providers(client, access_token): - headers = {"Authorization": f"Bearer {access_token}"} - response = client.get("/client_and_provider/", headers=headers) - assert response.status_code == 200 - assert "items" in response.json() - assert "page" in response.json() - assert "page_size" in response.json() - - -@pytest.mark.usefixtures("client", "access_token") -def test_get_client_or_provider_not_found(client, access_token): - headers = {"Authorization": f"Bearer {access_token}"} - response = client.get("/client_and_provider/invalid_id", headers=headers) - assert response.status_code == 404 - - -def test_create_client_or_provider_forbidden(): - response = client.post( - "/client_and_provider/", json={"name": "Test Client/Provider"} - ) - assert response.status_code in (403, 405, 404) - - -def test_update_client_or_provider_forbidden(): - response = client.put( - "/client_and_provider/1", json={"name": "Updated Client/Provider"} - ) - assert response.status_code in (403, 405, 404) diff --git a/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/test_validators.py b/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/test_validators.py deleted file mode 100644 index 5ee1e46e..00000000 --- a/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/test_validators.py +++ /dev/null @@ -1,143 +0,0 @@ -""" -Tests para validaciones CSV de clientes y proveedores (paridad Clarion). -""" -import pytest - -from api.v1.modules.a76.layouts_csv.clients_and_providers.validators.common import ( - validate_row_client_provider, - validate_row_desfase, - validate_row_clave_vacia, - validaciones_cliente_o_prov, - valida_toda_cliente_o_prov, - valida_parcial_cliente_o_prov, -) -from api.v1.modules.a76.layouts_csv.clients_and_providers.common.common_validators import ( - check_procedencia, - check_tipo_programa_secon, - check_es_prosec_num_aut, - check_vinculacion, - check_es_empresa_certificada_registro, - check_transformador_submaq, - check_desfase, - check_short_name_max_clarion, -) - - -def test_clave_vacia(): - err = validate_row_clave_vacia({"SHORT_NAME": ""}, 1) - assert err is not None - assert err["col"] == "SHORT_NAME" - assert "vacía" in err["msg"] - assert validate_row_clave_vacia({"SHORT_NAME": "ABC"}, 1) is None - - -def test_desfase(): - assert validate_row_desfase({"COL_EXTRA": ""}, 1) is None - err = validate_row_desfase({"COL_EXTRA": "x"}, 1) - assert err is not None - assert "desfase" in err["msg"] - - -def test_procedencia_e_n(): - assert check_procedencia({"PROCEDENCIA": "E"}, 1) is None - assert check_procedencia({"PROCEDENCIA": "N"}, 1) is None - err = check_procedencia({"PROCEDENCIA": "X"}, 1) - assert err is not None - assert err["col"] == "PROCEDENCIA" - err = check_procedencia({"PROCEDENCIA": "N", "PAIS": "USA"}, 1) - assert err is not None - assert "MEX" in err["solution"] - err = check_procedencia({"PROCEDENCIA": "E", "PAIS": "MEX"}, 1) - assert err is not None - - -def test_tipo_programa_secon(): - assert check_tipo_programa_secon({"TIPO_PROGRAMA_SECON": "IMMEX", "NUM_PROGRAMA_SECON": "123", "FECHA_AUT_SECON": "01/01/2020"}, 1) is None - err = check_tipo_programa_secon({"TIPO_PROGRAMA_SECON": "IMMEX", "NUM_PROGRAMA_SECON": ""}, 1) - assert err is not None - assert "R" in err["col"] or "NUM_PROGRAMA" in err["col"] - err = check_tipo_programa_secon({"TIPO_PROGRAMA_SECON": "Ninguno", "NUM_PROGRAMA_SECON": "123"}, 1) - assert err is not None - - -def test_es_prosec_num_aut(): - assert check_es_prosec_num_aut({"ES_PROSEC": "SI", "NUM_AUT_PROSEC": "AUTH123"}, 1) is None - err = check_es_prosec_num_aut({"ES_PROSEC": "SI", "NUM_AUT_PROSEC": ""}, 1) - assert err is not None - err = check_es_prosec_num_aut({"ES_PROSEC": "NO", "NUM_AUT_PROSEC": "X"}, 1) - assert err is not None - - -def test_vinculacion(): - assert check_vinculacion({"VINCULACION": "0"}, 1) is None - assert check_vinculacion({"VINCULACION": "1"}, 1) is None - err = check_vinculacion({"VINCULACION": "5"}, 1) - assert err is not None - - -def test_empresa_certificada(): - assert check_es_empresa_certificada_registro({"ES_EMPRESA_CERTIFICADA": "SI", "REGISTRO_EMPRESA_CERT": "REG123"}, 1) is None - err = check_es_empresa_certificada_registro({"ES_EMPRESA_CERTIFICADA": "SI", "REGISTRO_EMPRESA_CERT": ""}, 1) - assert err is not None - - -def test_transformador_submaq(): - assert check_transformador_submaq({"TRANSFORMA_SUBMAQ": "T"}, 1) is None - assert check_transformador_submaq({"TRANSFORMA_SUBMAQ": "Ninguno"}, 1) is None - err = check_transformador_submaq({"TRANSFORMA_SUBMAQ": "X"}, 1) - assert err is not None - - -def test_short_name_max_8(): - assert check_short_name_max_clarion({"SHORT_NAME": "12345678"}, 1) is None - err = check_short_name_max_clarion({"SHORT_NAME": "123456789"}, 1) - assert err is not None - - -def test_valida_toda_requiere_procedencia_y_nombre(): - row = {"SHORT_NAME": "C1", "RFC": "RFC1", "NOMBRE": "Nom", "PROCEDENCIA": ""} - err = valida_toda_cliente_o_prov(row, 1, actualizar=False) - assert err is not None - assert "Procedencia" in err["msg"] or "Col.A" in err["msg"] or "obligatorios" in err["msg"] - row["PROCEDENCIA"] = "N" - row["NOMBRE"] = "" - err = valida_toda_cliente_o_prov(row, 1, actualizar=False) - assert err is not None - row["NOMBRE"] = "Nom" - row["PAIS"] = "MEX" - err = valida_toda_cliente_o_prov(row, 1, actualizar=False) - assert err is None - - -def test_valida_parcial_no_requiere_nombre(): - row = {"SHORT_NAME": "C1", "RFC": "RFC1", "TIPO": "C"} - err = valida_parcial_cliente_o_prov(row, 1) - # Sin errores de dominio (procedencia/nombre no requeridos) - assert err is None or err.get("col") in ("SHORT_NAME", "TIPO", "RFC") - - -def test_validate_row_client_provider_full(): - row = { - "PROCEDENCIA": "N", - "PAIS": "MEX", - "TIPO": "C", - "SHORT_NAME": "CLI1", - "NOMBRE": "Cliente 1", - "RFC": "RFC123456789", - } - err = validate_row_client_provider(row, 1, actualizar=False, existing_short_names=None) - assert err is None - - -def test_validate_row_client_provider_actualizar_clave_no_existe(): - row = {"SHORT_NAME": "NOEXISTE", "RFC": "RFC1", "NOMBRE": "X"} - err = validate_row_client_provider(row, 1, actualizar=True, existing_short_names=set()) - assert err is not None - assert "Clave" in err["msg"] or "no existe" in err["msg"].lower() - - -def test_validate_row_client_provider_actualizar_clave_existe_parcial(): - row = {"SHORT_NAME": "EXIST1", "RFC": "RFC1", "TIPO": "A"} - err = validate_row_client_provider(row, 1, actualizar=True, existing_short_names={"EXIST1"}) - # VALIDA_PARCIAL: no requiere PROCEDENCIA ni NOMBRE; requiere RFC al final - assert err is None or err.get("col") == "RFC" diff --git a/backend/api/v1/modules/a76/layouts_csv/common/csv_reader.py b/backend/api/v1/modules/a76/layouts_csv/common/csv_reader.py index d5fb0c2e..24359188 100644 --- a/backend/api/v1/modules/a76/layouts_csv/common/csv_reader.py +++ b/backend/api/v1/modules/a76/layouts_csv/common/csv_reader.py @@ -1,11 +1,11 @@ """ Lectura de CSV con detección de delimitador (compartida por layouts_csv). Si se pasa fieldnames, no se usa la primera fila como cabecera y se toma como dato (CSV sin cabeceras). -Si fieldnames es None, cabeceras vacías se normalizan a _COL_0_, _COL_1_, ... para no colapsar columnas. +Si se pasa headerless_first_cell_values, se detecta si la primera fila es cabecera o dato por el valor de la primera celda. """ import csv import io -from typing import Iterator, Tuple, Dict, Any, Optional, List +from typing import Iterator, Tuple, Dict, Any, Optional, List, Set def _normalize_empty_headers(headers: List[str]) -> List[str]: @@ -24,12 +24,15 @@ def _normalize_empty_headers(headers: List[str]) -> List[str]: def iter_csv_rows( file_path: str, fieldnames: Optional[List[str]] = None, + headerless_first_cell_values: Optional[Set[str]] = None, + headerless_second_cell_key_pattern: Optional[str] = None, ) -> Iterator[Tuple[int, Dict[str, Any]]]: """ Abre el CSV, detecta dialecto y devuelve (line_num, row_dict) por cada fila. line_num empieza en 1 (primera fila de datos). - Si fieldnames es None: la primera fila del archivo se usa como cabecera (comportamiento por defecto). - Si fieldnames es una lista: no se usa cabecera; la primera fila se considera dato y se usan fieldnames como columnas. + Si fieldnames y headerless_first_cell_values se pasan: si la primera celda de la primera fila + (quitando BOM, strip, upper) está en headerless_first_cell_values, se trata como dato y se usan fieldnames. + headerless_second_cell_key_pattern se ignora si no se usa (reservado para otros layouts). """ with open(file_path, "r", encoding="utf-8-sig") as f: sample = f.read(2048) @@ -38,7 +41,37 @@ def iter_csv_rows( dialect = csv.Sniffer().sniff(sample, delimiters=",;\t") except Exception: dialect = "excel" - if fieldnames: + if fieldnames and headerless_first_cell_values is not None: + first_line = f.readline() + if not first_line: + return + row_reader = csv.reader(io.StringIO(first_line), dialect=dialect) + first_cells = next(row_reader, None) + if not first_cells: + return + first_cell_clean = (first_cells[0] or "").lstrip("\ufeff").strip().upper() + use_headerless = first_cell_clean in headerless_first_cell_values + if use_headerless: + pad = len(fieldnames) - len(first_cells) + cells = first_cells[: len(fieldnames)] + ([""] * pad if pad > 0 else []) + yield 1, dict(zip(fieldnames, cells)) + reader = csv.DictReader(f, fieldnames=fieldnames, dialect=dialect, restval="") + for i, row in enumerate(reader, start=2): + yield i, dict(row) + return + f.seek(0) + first_line = f.readline() + if not first_line: + return + row_reader = csv.reader(io.StringIO(first_line), dialect=dialect) + raw_headers = next(row_reader, None) + if not raw_headers: + return + normalized = _normalize_empty_headers(raw_headers) + reader = csv.DictReader(f, fieldnames=normalized, dialect=dialect, restval="") + for i, row in enumerate(reader, start=1): + yield i, row + elif fieldnames: reader = csv.DictReader(f, fieldnames=fieldnames, dialect=dialect) for i, row in enumerate(reader, start=1): yield i, dict(row) diff --git a/backend/api/v1/modules/a76/layouts_csv/customs_brokers/common/__init__.py b/backend/api/v1/modules/a76/layouts_csv/customs_brokers/common/__init__.py index 9b248794..47ef2168 100644 --- a/backend/api/v1/modules/a76/layouts_csv/customs_brokers/common/__init__.py +++ b/backend/api/v1/modules/a76/layouts_csv/customs_brokers/common/__init__.py @@ -1 +1 @@ -# common_validators, mappers (no fk_loader for customs_brokers) +# common_validators, mappers, fk_loader diff --git a/backend/api/v1/modules/a76/layouts_csv/customs_brokers/common/common_validators.py b/backend/api/v1/modules/a76/layouts_csv/customs_brokers/common/common_validators.py index 1902dad2..65292796 100644 --- a/backend/api/v1/modules/a76/layouts_csv/customs_brokers/common/common_validators.py +++ b/backend/api/v1/modules/a76/layouts_csv/customs_brokers/common/common_validators.py @@ -1,28 +1,194 @@ """ -Validadores reutilizables para import CSV de agentes aduanales (clave, licencia). +Validadores reutilizables para import CSV de agentes aduanales (customs brokers). +Paridad Clarion: VALIDACIONES_AGENTE_ADUANAL, tipo MEX/AME, patente obligatoria si MEX, +país en catálogo, RFC/CURP máx, desfase. """ import re -from typing import Dict, Any, Optional +from typing import Dict, Any, Optional, Set BROKER_KEY_MAX = 5 LICENSE_MAX = 4 +RFC_MAX = 30 +CURP_MAX = 19 def check_required_broker_key(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + """Col B: Clave de Agente Aduanal obligatoria, máx 5 caracteres.""" clave = (row.get("CLAVE") or "").strip() if not clave: - return {"line": line_num, "col": "CLAVE", "msg": "Requerido"} + return { + "line": line_num, + "col": "CLAVE", + "msg": f"Error: (Celda B{line_num}) La Clave de Agente Aduanal está vacía y no se pueden hacer las validaciones.", + "solution": f"Capturar en la Celda B{line_num} una Clave de Agente Aduanal nueva o existente a la cual desee agregar, remplazar o actualizar campos", + } if len(clave) > BROKER_KEY_MAX: - return {"line": line_num, "col": "CLAVE", "msg": f"Máximo {BROKER_KEY_MAX} caracteres"} + return { + "line": line_num, + "col": "CLAVE", + "msg": f"Error: (Col. B) La Clave de Agente Aduanal: {clave} supera la longitud de caracteres.", + "solution": f"Capturar en la columna B una Clave de Agente Aduanal de {BROKER_KEY_MAX} caracteres como máximo.", + } if not re.match(r"^[a-zA-Z0-9]+$", clave): - return {"line": line_num, "col": "CLAVE", "msg": "Solo letras y números"} + return { + "line": line_num, + "col": "CLAVE", + "msg": "Error: (Col. B) La Clave de Agente Aduanal solo puede contener letras y números.", + "solution": "Capturar en la columna B una Clave alfanumérica.", + } return None def check_optional_license(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + """Col C: Patente/Licencia opcional; si tiene valor, máx 4 dígitos.""" licencia = (row.get("LICENCIA") or "").strip() if not licencia: return None if len(licencia) > LICENSE_MAX or not licencia.isdigit(): - return {"line": line_num, "col": "LICENCIA", "msg": "Máximo 4 dígitos numéricos"} + return { + "line": line_num, + "col": "LICENCIA", + "msg": f"Error: (Col. C) La Patente debe ser de hasta {LICENSE_MAX} dígitos numéricos.", + "solution": f"Capturar en la columna C una Patente de hasta {LICENSE_MAX} dígitos.", + } return None + + +def check_desfase(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + """Si COL_EXTRA (Col O) tiene valor → advertencia de desfase.""" + val = (row.get("COL_EXTRA") or "").strip() + if not val: + return None + return { + "line": line_num, + "col": "COL_EXTRA", + "msg": "Advertencia: Podría existir un desfase en esta línea.", + "solution": "Revisar esta línea del archivo CSV y verificar cada campo esté en la posición correcta.", + } + + +def parse_tipo_agente_aduanal(val: Optional[str]) -> Optional[str]: + """ + Parsea TIPO (Col A): letra (M/A) o palabra (MEX/Mexicano, AME/Americano). + Devuelve 'M' o 'A'; si no reconoce, None. + """ + if not val or not str(val).strip(): + return None + s = str(val).strip() + v = s.upper() + if len(v) == 1: + if v == "M": + return "M" + if v == "A": + return "A" + v_lower = s.lower() + if v_lower in ("mex", "mexicano"): + return "M" + if v_lower in ("ame", "americano"): + return "A" + return None + + +def check_tipo_mex_ame(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + """ + Col A: TIPO debe ser MEX o AME (flexible: M/MEX/Mexicano, A/AME/Americano). + Si TIPO=M entonces PAIS (Col J) debe ser MEX; si TIPO=A entonces PAIS no debe ser MEX. + """ + tipo_raw = (row.get("TIPO") or "").strip() + if not tipo_raw: + return None + tipo = parse_tipo_agente_aduanal(tipo_raw) + if tipo is None: + return { + "line": line_num, + "col": "TIPO", + "msg": f"Error: (Col. A) El tipo de agente aduanal: {tipo_raw} es incorrecto.", + "solution": "Capturar en columna A el Tipo de agente aduanal correcto, MEX para Mexicano y AME para Americano.", + } + pais = (row.get("PAIS") or "").strip().upper() + if not pais: + return None + if tipo == "M" and pais != "MEX": + return { + "line": line_num, + "col": "PAIS", + "msg": f"Error: (Col. J) El tipo de cliente: {tipo_raw} tiene el país {pais}, es incorrecto.", + "solution": "Capturar en la columna J el país con clave MEX, ya que es un Agente Aduanal Mexicano", + } + if tipo == "A" and pais == "MEX": + return { + "line": line_num, + "col": "PAIS", + "msg": f"Error: (Col. J) El tipo de cliente: {tipo_raw} tiene el país {pais}, es incorrecto.", + "solution": "Capturar en la columna J un país con clave diferente de MEX, ya que es un Agente Aduanal Americano.", + } + return None + + +def check_patente_obligatoria_si_mex(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + """Col C: Patente obligatoria si TIPO es MEX.""" + tipo = parse_tipo_agente_aduanal((row.get("TIPO") or "").strip()) + if tipo != "M": + return None + licencia = (row.get("LICENCIA") or "").strip() + if not licencia: + clave = (row.get("CLAVE") or "").strip() + return { + "line": line_num, + "col": "LICENCIA", + "msg": f"Error: (Col. C) La Patente para el Agente Aduanal con Clave: {clave} no está capturada.", + "solution": "Capturar en la columna C una Patente de Agente Aduanal.", + } + return None + + +def check_rfc_max(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + """Col E: RFC opcional; si tiene valor, máx 30 caracteres.""" + val = (row.get("RFC") or "").strip() + if not val: + return None + if len(val) > RFC_MAX: + clave = (row.get("CLAVE") or "").strip() + return { + "line": line_num, + "col": "RFC", + "msg": f"Error: (Col. E) El RFC de Agente Aduanal: {clave} supera la longitud de caracteres.", + "solution": f"Capturar en la columna E un RFC de {RFC_MAX} caracteres como máximo.", + } + return None + + +def check_curp_max(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + """Col N: CURP/PERSONAL_ID opcional; si tiene valor, máx 19 caracteres.""" + val = (row.get("PERSONAL_ID") or "").strip() + if not val: + return None + if len(val) > CURP_MAX: + clave = (row.get("CLAVE") or "").strip() + return { + "line": line_num, + "col": "PERSONAL_ID", + "msg": f"Error: (Col. N) El CURP de Agente Aduanal: {clave} supera la longitud de caracteres.", + "solution": f"Capturar en la columna N un CURP de Agente Aduanal de {CURP_MAX} caracteres como máximo.", + } + return None + + +def check_pais_catalogo( + row: Dict[str, Any], + line_num: int, + valid_country_m3: Optional[Set[str]] = None, +) -> Optional[Dict[str, Any]]: + """Col J: Si PAIS tiene valor, debe existir en catálogo (clave M3).""" + val = (row.get("PAIS") or "").strip() + if not val or valid_country_m3 is None: + return None + if val.upper() in valid_country_m3: + return None + clave = (row.get("CLAVE") or "").strip() + return { + "line": line_num, + "col": "PAIS", + "msg": f"Error: (Col. J) El Pais del Agente Aduanal: {clave} no está en el Catálogo de Países.", + "solution": "Capturar en la columna J un País válido.", + } diff --git a/backend/api/v1/modules/a76/layouts_csv/customs_brokers/common/fk_loader.py b/backend/api/v1/modules/a76/layouts_csv/customs_brokers/common/fk_loader.py new file mode 100644 index 00000000..255f2603 --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/customs_brokers/common/fk_loader.py @@ -0,0 +1,26 @@ +""" +Carga de conjuntos FK para validación de import CSV de agentes aduanales. +Clarion: catálogo de países (GPaises / m3_key). +""" +from typing import Set +import logging + +from core.database import CoreSessionLocal + +logger = logging.getLogger(__name__) + + +def load_customs_brokers_fk_sets() -> Set[str]: + """ + Carga valid_country_m3 (códigos país m3_key) para validar Col J PAIS. + """ + valid_country_m3: Set[str] = set() + try: + with CoreSessionLocal() as session: + from api.v1.modules.public.reference_data.countries.models import Country + for row in session.query(Country.m3_key).all(): + if row[0]: + valid_country_m3.add((row[0] or "").strip().upper()) + except Exception as e: + logger.warning("Customs brokers import: could not load country m3 keys: %s", e) + return valid_country_m3 diff --git a/backend/api/v1/modules/a76/layouts_csv/customs_brokers/common/mappers.py b/backend/api/v1/modules/a76/layouts_csv/customs_brokers/common/mappers.py index 5dee87dc..4fbd6eae 100644 --- a/backend/api/v1/modules/a76/layouts_csv/customs_brokers/common/mappers.py +++ b/backend/api/v1/modules/a76/layouts_csv/customs_brokers/common/mappers.py @@ -1,8 +1,11 @@ """ Mapeo fila CSV → datos para CustomsBroker. +Clarion: TIPO normalizado a M/A con parse_tipo_agente_aduanal; CURP/personal_id máx 19. """ from typing import Dict, Any, Optional +from .common_validators import parse_tipo_agente_aduanal, CURP_MAX + MAX_LEN = { "broker_key": 5, "type": 9, @@ -49,11 +52,13 @@ def row_to_customs_broker_data( clave = _str_or_none(row_norm.get("CLAVE"), MAX_LEN["broker_key"]) if not clave: return {} + tipo_raw = (row_norm.get("TIPO") or "").strip() + tipo_normalized = parse_tipo_agente_aduanal(tipo_raw) if tipo_raw else None return { "tenant_id": tenant_id, "company_id": company_id, "broker_key": clave, - "type": _str_or_none(row_norm.get("TIPO"), MAX_LEN["type"]), + "type": tipo_normalized or _str_or_none(row_norm.get("TIPO"), MAX_LEN["type"]), "name": _str_or_none(row_norm.get("NOMBRE"), MAX_LEN["name"]), "address": _str_or_none(row_norm.get("DIRECCION"), MAX_LEN["address"]), "postal_code": _str_or_none(row_norm.get("CODIGO POSTAL"), MAX_LEN["postal_code"]), @@ -64,7 +69,7 @@ def row_to_customs_broker_data( "email": _str_or_none(row_norm.get("EMAIL"), MAX_LEN["email"]), "country": _str_or_none(row_norm.get("PAIS"), MAX_LEN["country"]), "tax_id": _str_or_none(row_norm.get("RFC"), MAX_LEN["tax_id"]), - "personal_id": _str_or_none(row_norm.get("PERSONAL_ID"), MAX_LEN["personal_id"]), + "personal_id": _str_or_none(row_norm.get("PERSONAL_ID"), CURP_MAX), "position": _str_or_none(row_norm.get("POSICION"), MAX_LEN["position"]), "license": _license_value(row_norm), "company": _str_or_none(row_norm.get("EMPRESA"), MAX_LEN["company"]), diff --git a/backend/api/v1/modules/a76/layouts_csv/customs_brokers/tasks.py b/backend/api/v1/modules/a76/layouts_csv/customs_brokers/tasks.py index 987bd736..2f58f124 100644 --- a/backend/api/v1/modules/a76/layouts_csv/customs_brokers/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/customs_brokers/tasks.py @@ -6,7 +6,7 @@ Usa layouts_csv.common (storage, normalize, meta, responses, csv_reader). import json import logging import os -from typing import Dict, Any, Optional, List +from typing import Dict, Any, Optional, List, Set from core.celery_app import celery_app from core.database import CoreSessionLocal @@ -16,9 +16,14 @@ from ..common import normalize as common_normalize from ..common import meta as common_meta from ..common import responses as common_responses from ..common import csv_reader as common_csv_reader -from .template_config import row_from_template +from .template_config import ( + row_from_template, + CUSTOMS_BROKERS_FIELDNAMES_ORDER, + CUSTOMS_BROKERS_HEADERLESS_FIRST_CELL, +) from .validators import validate_row_customs_broker from .common.mappers import row_to_customs_broker_data +from .common.fk_loader import load_customs_brokers_fk_sets logger = logging.getLogger(__name__) @@ -49,6 +54,28 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str, except ValueError as e: return {"status": "failed", "error": str(e)} + meta = common_meta.load_meta(file_path) or {} + actualizar = meta.get("actualizar", False) + existing_broker_keys: Set[str] = set() + if actualizar: + try: + from api.v1.modules.a76.customs_brokers.models import CustomsBroker + with CoreSessionLocal() as session: + for b in ( + session.query(CustomsBroker) + .filter( + CustomsBroker.tenant_id == tenant_id, + CustomsBroker.company_id == company_id, + ) + .all() + ): + if (b.broker_key or "").strip(): + existing_broker_keys.add((b.broker_key or "").strip()) + except Exception as e: + logger.warning("CB import: could not load existing broker_keys for ACT: %s", e) + + valid_country_m3 = load_customs_brokers_fk_sets() + error_count = 0 processed_rows = 0 errors_detail: List[Dict[str, Any]] = [] @@ -56,12 +83,22 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str, try: with open(error_path, "w", encoding="utf-8") as f_err: - for i, row in common_csv_reader.iter_csv_rows(file_path): + for i, row in common_csv_reader.iter_csv_rows( + file_path, + fieldnames=CUSTOMS_BROKERS_FIELDNAMES_ORDER, + headerless_first_cell_values=CUSTOMS_BROKERS_HEADERLESS_FIRST_CELL, + ): if progress_callback and i % 500 == 0: progress_callback(i, total_rows, error_count) row_norm = row_from_template(row, common_normalize.normalize_header) - err = validate_row_customs_broker(row_norm, i) + err = validate_row_customs_broker( + row_norm, + i, + actualizar=actualizar, + existing_broker_keys=existing_broker_keys, + valid_country_m3=valid_country_m3, + ) if err: error_count += 1 error_lines_list.append(err["line"]) @@ -111,6 +148,10 @@ def _do_commit(job_id: str) -> Dict[str, Any]: except ValueError as e: return {"status": "failed", "error": str(e)} + meta = common_meta.load_meta(file_path) or {} + actualizar = meta.get("actualizar", False) + valid_country_m3 = load_customs_brokers_fk_sets() + from api.v1.modules.a76.customs_brokers.models import CustomsBroker inserted_count = 0 @@ -131,12 +172,24 @@ def _do_commit(job_id: str) -> Dict[str, Any]: ): existing_by_key[b.broker_key] = b - for i, row in common_csv_reader.iter_csv_rows(file_path): + existing_broker_keys = set(existing_by_key.keys()) + + for i, row in common_csv_reader.iter_csv_rows( + file_path, + fieldnames=CUSTOMS_BROKERS_FIELDNAMES_ORDER, + headerless_first_cell_values=CUSTOMS_BROKERS_HEADERLESS_FIRST_CELL, + ): if i in error_lines: continue row_norm = row_from_template(row, common_normalize.normalize_header) - err = validate_row_customs_broker(row_norm, i) + err = validate_row_customs_broker( + row_norm, + i, + actualizar=actualizar, + existing_broker_keys=existing_broker_keys, + valid_country_m3=valid_country_m3, + ) if err: skipped_invalid += 1 skipped_details.append({ 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 f4801729..b656b59a 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 @@ -1,5 +1,8 @@ """ Configuración de plantilla CSV para Agentes Aduanales (EstructuraCatAgenteAduanal.xls). +Layout Clarion: A=TIPO, B=CLAVE AADUANAL, C=PATENTE, D=NOMBRE, E=RFC, F=DIRECCION, G=CODIGO POSTAL, +H=CIUDAD, I=ESTADO, J=PAIS, K=TELEFONO, L=NUMERO FAX, M=CORREO ELECTRONICO, N=CURP, O=COL_EXTRA (desfase). +Encabezado ejemplo: "TIPO(MEX=Mexicano,AME=AMERICANO)",CLAVE AADUANAL,PATENTE,NOMBRE,RFC,... Solo se leen columnas definidas aquí; el resto se ignora. """ @@ -7,26 +10,66 @@ from typing import Dict, List, Any, Optional TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { "customs_brokers": [ - {"canonical": "CLAVE", "aliases": ["BROKER_KEY", "CLAVE AGENTE", "ID"]}, - {"canonical": "TIPO"}, - {"canonical": "NOMBRE", "aliases": ["NOMBRE COMPLETO", "RAZON SOCIAL"]}, - {"canonical": "DIRECCION", "aliases": ["DOMICILIO"]}, - {"canonical": "CODIGO POSTAL", "aliases": ["CODIGOPOSTAL", "CP", "C.P."]}, - {"canonical": "CIUDAD"}, - {"canonical": "ESTADO"}, - {"canonical": "TELEFONO", "aliases": ["PHONE", "TEL"]}, - {"canonical": "FAX"}, - {"canonical": "EMAIL", "aliases": ["CORREO", "E-MAIL"]}, - {"canonical": "PAIS", "aliases": ["COUNTRY"]}, - {"canonical": "RFC", "aliases": ["TAX_ID", "TAXID"]}, - {"canonical": "PERSONAL_ID", "aliases": ["PERSONALID", "ID PERSONAL"]}, - {"canonical": "POSICION", "aliases": ["CARGO", "PUESTO"]}, + # Col A - TIPO (MEX/Mexicano, AME/Americano) + {"canonical": "TIPO", "aliases": ["TIPO(MEX=Mexicano,AME=AMERICANO)"]}, + # Col B - Clave agente aduanal (máx 5) + {"canonical": "CLAVE", "aliases": ["BROKER_KEY", "CLAVE AGENTE", "CLAVE AADUANAL", "CLAVE ADUANAL", "ID"]}, + # Col C - Patente / Licencia (obligatoria si TIPO=MEX) {"canonical": "LICENCIA", "aliases": ["PATENTE", "LICENSE"]}, + # Col D - Nombre + {"canonical": "NOMBRE", "aliases": ["NOMBRE COMPLETO", "RAZON SOCIAL"]}, + # Col E - RFC (máx 30) + {"canonical": "RFC", "aliases": ["TAX_ID", "TAXID"]}, + # Col F - Dirección + {"canonical": "DIRECCION", "aliases": ["DOMICILIO"]}, + # Col G - Código postal + {"canonical": "CODIGO POSTAL", "aliases": ["CODIGOPOSTAL", "CP", "C.P."]}, + # Col H - Ciudad + {"canonical": "CIUDAD"}, + # Col I - Estado + {"canonical": "ESTADO"}, + # Col J - País (clave M3) + {"canonical": "PAIS", "aliases": ["COUNTRY"]}, + # Col K - Teléfono + {"canonical": "TELEFONO", "aliases": ["PHONE", "TEL"]}, + # Col L - Fax + {"canonical": "FAX", "aliases": ["NUMERO FAX"]}, + # Col M - Email + {"canonical": "EMAIL", "aliases": ["CORREO", "E-MAIL", "CORREO ELECTRONICO"]}, + # Col N - CURP / Personal ID (máx 19) + {"canonical": "PERSONAL_ID", "aliases": ["PERSONALID", "ID PERSONAL", "CURP"]}, + # Col O - Desfase (si tiene valor → advertencia) + {"canonical": "COL_EXTRA", "aliases": ["COLUMNA EXTRA", "DESFASE"]}, + # Otros opcionales (no en encabezado oficial) + {"canonical": "POSICION", "aliases": ["CARGO", "PUESTO"]}, {"canonical": "EMPRESA", "aliases": ["COMPANY"]}, {"canonical": "CONTACTO", "aliases": ["CONTACT"]}, ], } +# Orden oficial de columnas (para CSV sin encabezado o detección) +CUSTOMS_BROKERS_FIELDNAMES_ORDER = [ + "TIPO", + "CLAVE", + "LICENCIA", + "NOMBRE", + "RFC", + "DIRECCION", + "CODIGO POSTAL", + "CIUDAD", + "ESTADO", + "PAIS", + "TELEFONO", + "FAX", + "EMAIL", + "PERSONAL_ID", +] + +# Si la primera celda de la primera fila está en este set, se trata como CSV sin encabezado +CUSTOMS_BROKERS_HEADERLESS_FIRST_CELL = frozenset( + {"MEX", "AME", "M", "A", "MEXICANO", "AMERICANO"} +) + def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]: """normalized_header -> canonical_name para plantilla customs_brokers.""" @@ -39,6 +82,8 @@ def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]: lookup[normalize_header_fn(canonical)] = canonical for alias in item.get("aliases") or []: lookup[normalize_header_fn(alias)] = canonical + for idx, name in enumerate(CUSTOMS_BROKERS_FIELDNAMES_ORDER): + lookup[normalize_header_fn(f"_COL_{idx}_")] = name return lookup diff --git a/backend/api/v1/modules/a76/layouts_csv/customs_brokers/validators/common.py b/backend/api/v1/modules/a76/layouts_csv/customs_brokers/validators/common.py index 492414a1..867bbe70 100644 --- a/backend/api/v1/modules/a76/layouts_csv/customs_brokers/validators/common.py +++ b/backend/api/v1/modules/a76/layouts_csv/customs_brokers/validators/common.py @@ -1,23 +1,126 @@ """ Validaciones comunes de fila para import CSV de agentes aduanales. +Paridad Clarion: VALIDA_TODA_AGENTE_ADUANAL, VALIDA_PARCIAL_AGENTE_ADUANAL, VALIDACIONES_AGENTE_ADUANAL. """ -from typing import Dict, Any, Optional +from typing import Dict, Any, Optional, Set from ..common.common_validators import ( check_required_broker_key, check_optional_license, + check_desfase, + check_tipo_mex_ame, + check_patente_obligatoria_si_mex, + check_rfc_max, + check_curp_max, + check_pais_catalogo, ) +MSG_CLAVE_NO_EXISTE = "Error: (Col. B) Clave de A. Aduanal No Existe en el Catalogo." +MSG_CLAVE_NO_EXISTE_SOLUCION = "La clave debe existir en el catálogo cuando el modo es Actualizar." +MSG_CAMPOS_OBLIGATORIOS = "Existen campos vacíos que son obligatorios, es la {campos}." +MSG_CAMPOS_OBLIGATORIOS_SOLUCION = "Revisar la línea del archivo y capturar los campos con la información correcta." -def validate_row_customs_broker(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + +def validaciones_agente_aduanal( + row: Dict[str, Any], + line_num: int, + valid_country_m3: Optional[Set[str]] = None, +) -> Optional[Dict[str, Any]]: """ - Valida una fila de CSV de agentes aduanales. - CLAVE requerida (max 5, alfanumérica); LICENCIA opcional (max 4 dígitos). + VALIDACIONES_AGENTE_ADUANAL: reglas de dominio compartidas. + Tipo MEX/AME + coherencia País, patente si MEX, RFC/CURP máx, país en catálogo, licencia formato. """ - err = check_required_broker_key(row, line_num) + err = check_tipo_mex_ame(row, line_num) + if err: + return err + err = check_patente_obligatoria_si_mex(row, line_num) if err: return err err = check_optional_license(row, line_num) + if err: + return err + err = check_rfc_max(row, line_num) + if err: + return err + err = check_curp_max(row, line_num) + if err: + return err + err = check_pais_catalogo(row, line_num, valid_country_m3) + if err: + return err + return None + + +def valida_toda_agente_aduanal( + row: Dict[str, Any], + line_num: int, + valid_country_m3: Optional[Set[str]] = None, +) -> Optional[Dict[str, Any]]: + """ + VALIDA_TODA: TIPO (Col A) y NOMBRE (Col D) obligatorios; luego VALIDACIONES_AGENTE_ADUANAL. + """ + campos_oblig = [] + if not (row.get("TIPO") or "").strip(): + campos_oblig.append("(Col.A) Tipo") + if not (row.get("NOMBRE") or "").strip(): + campos_oblig.append("(Col.D) Nombre") + if campos_oblig: + return { + "line": line_num, + "col": "TIPO" if not (row.get("TIPO") or "").strip() else "NOMBRE", + "msg": MSG_CAMPOS_OBLIGATORIOS.format(campos=", ".join(campos_oblig)), + "solution": MSG_CAMPOS_OBLIGATORIOS_SOLUCION, + } + return validaciones_agente_aduanal(row, line_num, valid_country_m3) + + +def valida_parcial_agente_aduanal( + row: Dict[str, Any], + line_num: int, + valid_country_m3: Optional[Set[str]] = None, +) -> Optional[Dict[str, Any]]: + """VALIDA_PARCIAL: solo VALIDACIONES_AGENTE_ADUANAL (no exige TIPO ni NOMBRE).""" + return validaciones_agente_aduanal(row, line_num, valid_country_m3) + + +def validate_row_customs_broker( + row: Dict[str, Any], + line_num: int, + actualizar: bool = False, + existing_broker_keys: Optional[Set[str]] = None, + valid_country_m3: Optional[Set[str]] = None, +) -> Optional[Dict[str, Any]]: + """ + Valida una fila de CSV de agentes aduanales. + 1. Desfase (COL_EXTRA) primero. + 2. Clave vacía → error. + 3. Si actualizar y clave no existe en existing_broker_keys → error. + 4. Si actualizar y clave existe → valida_parcial_agente_aduanal. + 5. Si no actualizar o clave no existe → valida_toda_agente_aduanal. + """ + err = check_desfase(row, line_num) + if err: + return err + err = check_required_broker_key(row, line_num) + if err: + return err + + clave = (row.get("CLAVE") or "").strip() + existing = existing_broker_keys or set() + use_partial = actualizar and clave in existing + + if actualizar and clave and clave not in existing: + return { + "line": line_num, + "col": "CLAVE", + "msg": MSG_CLAVE_NO_EXISTE, + "solution": MSG_CLAVE_NO_EXISTE_SOLUCION, + } + + if use_partial: + err = valida_parcial_agente_aduanal(row, line_num, valid_country_m3) + else: + err = valida_toda_agente_aduanal(row, line_num, valid_country_m3) if err: return err return None