From 2df4a029e06a530d8bc77e1a25ad7362eaec460d Mon Sep 17 00:00:00 2001 From: hreyes Date: Fri, 6 Mar 2026 09:03:28 -0700 Subject: [PATCH] feature/trailes-validaciones-clarion-csv --- .../layouts_csv/trailers/common/__init__.py | 2 +- .../trailers/common/common_validators.py | 146 +++++++++++++++++- .../layouts_csv/trailers/common/fk_loader.py | 78 ++++++++++ .../layouts_csv/trailers/common/mappers.py | 65 +++++++- .../a76/layouts_csv/trailers/routes.py | 2 + .../modules/a76/layouts_csv/trailers/tasks.py | 107 +++++++++++-- .../layouts_csv/trailers/template_config.py | 1 + .../trailers/validators/__init__.py | 4 +- .../layouts_csv/trailers/validators/common.py | 92 ++++++++++- .../layouts_csv/trailers/validators/create.py | 60 ++++++- 10 files changed, 533 insertions(+), 24 deletions(-) create mode 100644 backend/api/v1/modules/a76/layouts_csv/trailers/common/fk_loader.py diff --git a/backend/api/v1/modules/a76/layouts_csv/trailers/common/__init__.py b/backend/api/v1/modules/a76/layouts_csv/trailers/common/__init__.py index 55e6cb8b..47ef2168 100644 --- a/backend/api/v1/modules/a76/layouts_csv/trailers/common/__init__.py +++ b/backend/api/v1/modules/a76/layouts_csv/trailers/common/__init__.py @@ -1 +1 @@ -# common_validators, mappers (no fk_loader for trailers) +# common_validators, mappers, fk_loader diff --git a/backend/api/v1/modules/a76/layouts_csv/trailers/common/common_validators.py b/backend/api/v1/modules/a76/layouts_csv/trailers/common/common_validators.py index ca5c6231..6948bf3c 100644 --- a/backend/api/v1/modules/a76/layouts_csv/trailers/common/common_validators.py +++ b/backend/api/v1/modules/a76/layouts_csv/trailers/common/common_validators.py @@ -1,7 +1,8 @@ """ Validadores reutilizables para import CSV de trailers (longitudes, requerido). +Paridad Clarion: VALIDACIONES_TRAILER, código entidad C/I/A/B, catálogos tipo/país/estado, desfase. """ -from typing import Dict, Any, Optional +from typing import Dict, Any, Optional, Set, Tuple # Max lengths from Trailer model (a76.trailer) MAX_LEN = { @@ -16,6 +17,8 @@ MAX_LEN = { "container_key": 3, } +CODIGO_ENTIDAD_VALIDOS = {"C", "I", "A", "B"} + def check_required(row: Dict[str, Any], col: str, max_len: int, line_num: int) -> Optional[Dict[str, Any]]: val = (row.get(col) or "").strip() @@ -37,3 +40,144 @@ def check_max_length( if len(val) > max_len: return {"line": line_num, "col": col, "msg": f"Máximo {max_len} caracteres"} return None + + +def _get_codigo_entidad(row: Dict[str, Any]) -> str: + return (row.get("CODIGO DE ENTIDAD") or row.get("CODIGO ENTIDAD") or "").strip() + + +def check_codigo_entidad_valores(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + """Col E: Si CODIGO DE ENTIDAD no vacío, debe ser C, I, A o B (Clarion).""" + val = _get_codigo_entidad(row).upper() + if not val: + return None + if val in CODIGO_ENTIDAD_VALIDOS: + return None + return { + "line": line_num, + "col": "CODIGO DE ENTIDAD", + "msg": f"Error: (Col. E) El Codigo de Entidad: {val} es incorrecto.", + "solution": "Capturar en columna E el Codigo de Entidad correcto, C, I, A ó B.", + } + + +def check_trailer_type_catalog( + row: Dict[str, Any], + line_num: int, + valid_trailer_type_keys: Optional[Set[str]] = None, +) -> Optional[Dict[str, Any]]: + """Col C: Si TIPO TRAILER no vacío, debe existir en catálogo (GTipoTrailer).""" + val = (row.get("TIPO TRAILER") or "").strip() + if not val or valid_trailer_type_keys is None: + return None + if val.upper() in valid_trailer_type_keys: + return None + return { + "line": line_num, + "col": "TIPO TRAILER", + "msg": f"Error: (Col. C) El Tipo de Trailer: {val} es incorrecto.", + "solution": "Capturar en columna C un Tipo del Trailer existente.", + } + + +def check_pais_catalog_trailers( + row: Dict[str, Any], + line_num: int, + valid_country_ame: Optional[Set[str]] = None, +) -> Optional[Dict[str, Any]]: + """Col H: Si PAIS no vacío, debe ser clave americana (≤2 chars) y existir en catálogo (GPaises.Pais_Ame).""" + val = (row.get("PAIS") or "").strip() + if not val or valid_country_ame is None: + return None + val_upper = val.upper() + if len(val) > 2: + return { + "line": line_num, + "col": "PAIS", + "msg": f"Error: (Celda H{line_num}) El Pais: {val} Es Incorrecto", + "solution": "Capturar en columna H un Pais en Clave Americana(US = Estados Unidos, MX = Mexico, ES = España, etc).", + } + if val_upper in valid_country_ame: + return None + return { + "line": line_num, + "col": "PAIS", + "msg": f"Error: (Col. H) El Pais: {val} es incorrecto.", + "solution": "Capturar en columna H el Pais del Transporte en Clave Americana.", + } + + +def check_estado_catalog_trailers( + row: Dict[str, Any], + line_num: int, + state_descriptions_upper: Optional[Set[str]] = None, + state_ame_to_description: Optional[Dict[str, str]] = None, +) -> Tuple[Optional[Dict[str, Any]], Optional[str]]: + """ + Col G: Si ESTADO no vacío, debe existir en catálogo (GEstados). Si hay estado, PAIS no puede estar vacío. + Devuelve (error, estado_resuelto_upper). estado_resuelto_upper es la descripción en mayúsculas para estado-país. + """ + state_ame_to_description = state_ame_to_description or {} + val = (row.get("ESTADO") or "").strip() + if not val or state_descriptions_upper is None: + return None, None + val_upper = val.upper() + resolved_upper: Optional[str] = None + if val_upper in state_descriptions_upper: + resolved_upper = val_upper + elif len(val) <= 2 and val_upper in state_ame_to_description: + resolved_upper = state_ame_to_description[val_upper] + if resolved_upper is None: + return { + "line": line_num, + "col": "ESTADO", + "msg": f"Error: (Celda G{line_num}) El Estado: {val} es incorrecto.", + "solution": "Capturar en columna G el Estado del Trailer en Clave Americana o Nombre Completo.", + }, None + # Si existe estado, PAIS no puede estar vacío (Clarion) + pais = (row.get("PAIS") or "").strip() + if not pais: + return { + "line": line_num, + "col": "PAIS", + "msg": f"Error: (Celda G{line_num}) El Estado: {val} No esta ligado a ningun Pais.", + "solution": "Capturar en columna H un Pais en Clave Americana(US = Estados Unidos, MX = Mexico, ES = España, etc).", + }, None + return None, resolved_upper + + +def check_estado_pais_consistency_trailers( + row: Dict[str, Any], + line_num: int, + state_country_set: Optional[Set[Tuple[str, str]]] = None, + estado_resuelto_upper: Optional[str] = None, +) -> Optional[Dict[str, Any]]: + """Si hay ESTADO y PAIS, validar que el estado pertenezca al país (Clarion GEstados-GPaises).""" + estado = (row.get("ESTADO") or "").strip() + pais = (row.get("PAIS") or "").strip().upper() + if not estado or not pais or state_country_set is None: + return None + desc_upper = estado_resuelto_upper or estado.upper() + key = (pais, desc_upper) + if key in state_country_set: + return None + return { + "line": line_num, + "col": "ESTADO", + "msg": f"Error: (Col. G) EL Estado: {estado} no pertenece al Pais: {pais}.", + "solution": "Capturar en columna G un Estado que pertenesca al Pais de la columna H.", + } + + +def check_desfase_trailers(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + """Si COL_EXTRA tiene valor → advertencia de desfase (Clarion, no bloqueante).""" + 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 este en la posicion correcta.", + "warning": True, + } diff --git a/backend/api/v1/modules/a76/layouts_csv/trailers/common/fk_loader.py b/backend/api/v1/modules/a76/layouts_csv/trailers/common/fk_loader.py new file mode 100644 index 00000000..c87adcaa --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/trailers/common/fk_loader.py @@ -0,0 +1,78 @@ +""" +Carga de conjuntos FK para validación de import CSV de trailers/cajas. +Clarion: GTipoTrailer, GPaises (Pais_Ame), GEstados (Descripcion / Clave_Ame), relación Estado-País. +""" +from typing import Set, Tuple, Optional, Dict +import logging + +from core.database import CoreSessionLocal + +logger = logging.getLogger(__name__) + + +def load_trailers_fk_sets( + tenant_id: Optional[int] = None, + company_id: Optional[int] = None, +) -> Tuple[ + Set[str], + Set[str], + Set[str], + Set[Tuple[str, str]], + Dict[str, str], +]: + """ + Carga conjuntos para validación CSV de trailers (paridad Clarion). + Devuelve: + - valid_trailer_type_keys: códigos de trailer_type (GTipoTrailer) + - valid_country_ame: claves americana de países (GPaises.Pais_Ame), mayúsculas + - state_descriptions_upper: descripciones de estados en mayúsculas (GEstados) + - state_country_set: set de (ame_key_pais, description_estado_upper) para validar "estado pertenece a país" + - state_ame_to_description: dict clave_ame_upper -> description_upper (opcional; vacío si State no tiene ame_key) + """ + valid_trailer_type_keys: Set[str] = set() + valid_country_ame: Set[str] = set() + state_descriptions_upper: Set[str] = set() + state_country_set: Set[Tuple[str, str]] = set() + state_ame_to_description: Dict[str, str] = {} + + try: + with CoreSessionLocal() as session: + from api.v1.modules.public.reference_data.trailer_types.models import TrailerType + from api.v1.modules.public.reference_data.countries.models import Country + from api.v1.modules.public.reference_data.states.models import State + + for row in session.query(TrailerType.trailer_type_key).all(): + if row[0]: + valid_trailer_type_keys.add((row[0] or "").strip().upper()) + + for row in session.query(Country.ame_key).all(): + if row[0]: + valid_country_ame.add((row[0] or "").strip().upper()) + + for state in session.query(State).all(): + desc = (state.description or "").strip() + if desc: + state_descriptions_upper.add(desc.upper()) + country = ( + session.query(Country) + .filter(Country.m3_key == state.m3_key) + .first() + ) + if country and (country.ame_key or "").strip(): + state_country_set.add( + ((country.ame_key or "").strip().upper(), desc.upper()) + ) + ame = getattr(state, "ame_key", None) + if ame and (ame or "").strip(): + state_ame_to_description[(ame or "").strip().upper()] = desc.upper() if desc else (ame or "").strip().upper() + + except Exception as e: + logger.warning("Trailers import: could not load FK sets: %s", e) + + return ( + valid_trailer_type_keys, + valid_country_ame, + state_descriptions_upper, + state_country_set, + state_ame_to_description, + ) diff --git a/backend/api/v1/modules/a76/layouts_csv/trailers/common/mappers.py b/backend/api/v1/modules/a76/layouts_csv/trailers/common/mappers.py index 62d3ec97..032a8edb 100644 --- a/backend/api/v1/modules/a76/layouts_csv/trailers/common/mappers.py +++ b/backend/api/v1/modules/a76/layouts_csv/trailers/common/mappers.py @@ -1,7 +1,8 @@ """ Mapeo fila CSV → datos para Trailer. +Paridad Clarion LLENA_TRAILER: en actualización, campo vacío en CSV usa valor existente del trailer. """ -from typing import Dict, Any, Optional +from typing import Dict, Any, Optional, Union from .common_validators import MAX_LEN @@ -17,9 +18,28 @@ def _str_or_none(val: Any, max_len: Optional[int] = None) -> Optional[str]: return s +def _get_existing_val(existing: Union[Any, Dict[str, Any]], key: str) -> Any: + """Obtiene valor del trailer existente (modelo ORM o dict).""" + if existing is None: + return None + if isinstance(existing, dict): + return existing.get(key) + return getattr(existing, key, None) + + +def _get_entity_code(row_norm: Dict[str, Any]) -> Optional[str]: + return _str_or_none( + row_norm.get("CODIGO ENTIDAD") or row_norm.get("CODIGO DE ENTIDAD"), + MAX_LEN["entity_code"], + ) + + def row_to_trailer_data(row_norm: Dict[str, Any]) -> Dict[str, Any]: """Build dict for TrailerCreateDTO / TrailerUpdateDTO from normalized row.""" - trailer_number = _str_or_none(row_norm.get("NUMERO TRAILER"), MAX_LEN["trailer_number"]) + trailer_number = _str_or_none( + row_norm.get("NUMERO TRAILER") or row_norm.get("CLAVE TRAILER"), + MAX_LEN["trailer_number"], + ) if not trailer_number: return {} return { @@ -27,9 +47,48 @@ def row_to_trailer_data(row_norm: Dict[str, Any]) -> Dict[str, Any]: "ace_trailer_number": _str_or_none(row_norm.get("CLAVE ACE"), MAX_LEN["ace_trailer_number"]), "trailer_type_key": _str_or_none(row_norm.get("TIPO TRAILER"), MAX_LEN["trailer_type_key"]), "seal": _str_or_none(row_norm.get("PRECINTO"), MAX_LEN["seal"]), - "entity_code": _str_or_none(row_norm.get("CODIGO ENTIDAD"), MAX_LEN["entity_code"]), + "entity_code": _get_entity_code(row_norm), "plate_number": _str_or_none(row_norm.get("PLACAS"), MAX_LEN["plate_number"]), "state": _str_or_none(row_norm.get("ESTADO"), MAX_LEN["state"]), "country": _str_or_none(row_norm.get("PAIS"), MAX_LEN["country"]), "container_key": _str_or_none(row_norm.get("CLAVE CONTENEDOR"), MAX_LEN["container_key"]), } + + +def row_to_trailer_data_for_update( + row_norm: Dict[str, Any], + existing_trailer: Union[Any, Dict[str, Any]], +) -> Dict[str, Any]: + """ + Build dict for TrailerUpdateDTO: CSV value if non-empty, else existing trailer value (Clarion VALIDA_PARCIAL / LLENA_TRAILER). + """ + trailer_number = _str_or_none( + row_norm.get("NUMERO TRAILER") or row_norm.get("CLAVE TRAILER"), + MAX_LEN["trailer_number"], + ) or _get_existing_val(existing_trailer, "trailer_number") + if not trailer_number: + return {} + + def _csv_or_existing(csv_key: str, dto_key: str, max_len: Optional[int] = None): + v = _str_or_none(row_norm.get(csv_key), max_len) + if v is not None and v != "": + return v + return _get_existing_val(existing_trailer, dto_key) + + def _entity_code_or_existing(): + v = _get_entity_code(row_norm) + if v is not None and v != "": + return v + return _get_existing_val(existing_trailer, "entity_code") + + return { + "trailer_number": trailer_number, + "ace_trailer_number": _csv_or_existing("CLAVE ACE", "ace_trailer_number", MAX_LEN["ace_trailer_number"]), + "trailer_type_key": _csv_or_existing("TIPO TRAILER", "trailer_type_key", MAX_LEN["trailer_type_key"]), + "seal": _csv_or_existing("PRECINTO", "seal", MAX_LEN["seal"]), + "entity_code": _entity_code_or_existing(), + "plate_number": _csv_or_existing("PLACAS", "plate_number", MAX_LEN["plate_number"]), + "state": _csv_or_existing("ESTADO", "state", MAX_LEN["state"]), + "country": _csv_or_existing("PAIS", "country", MAX_LEN["country"]), + "container_key": _csv_or_existing("CLAVE CONTENEDOR", "container_key", MAX_LEN["container_key"]), + } diff --git a/backend/api/v1/modules/a76/layouts_csv/trailers/routes.py b/backend/api/v1/modules/a76/layouts_csv/trailers/routes.py index 64dc86f1..0c9475fd 100644 --- a/backend/api/v1/modules/a76/layouts_csv/trailers/routes.py +++ b/backend/api/v1/modules/a76/layouts_csv/trailers/routes.py @@ -43,6 +43,7 @@ def _get_redis(): async def upload_import_file( file: UploadFile = File(...), company_id: int = Query(..., description="Company ID"), + actualizar: bool = Query(False, description="Modo Agregar/Actualizar (ACT); si True, validación parcial para claves existentes"), db: Session = Depends(get_core_db), current_user: Dict[str, Any] = Depends(get_current_user), ): @@ -63,6 +64,7 @@ async def upload_import_file( "company_id": company_id, "user_id": current_user.get("id"), "template_id": "trailers", + "actualizar": actualizar, } try: diff --git a/backend/api/v1/modules/a76/layouts_csv/trailers/tasks.py b/backend/api/v1/modules/a76/layouts_csv/trailers/tasks.py index 448788ee..3102971d 100644 --- a/backend/api/v1/modules/a76/layouts_csv/trailers/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/trailers/tasks.py @@ -7,7 +7,7 @@ import csv 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 @@ -17,8 +17,9 @@ from ..common import normalize as common_normalize from ..common import meta as common_meta from ..common import responses as common_responses from .template_config import row_from_template -from .validators import validate_row_trailer -from .common.mappers import row_to_trailer_data +from .validators import validate_row_trailer, validate_row_trailer_desfase +from .common.mappers import row_to_trailer_data, row_to_trailer_data_for_update +from .common.fk_loader import load_trailers_fk_sets logger = logging.getLogger(__name__) @@ -68,6 +69,34 @@ 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_trailer_numbers: Set[str] = set() + if actualizar: + try: + from api.v1.modules.a76.transportation.trailers.models import Trailer + with CoreSessionLocal() as session: + for t in ( + session.query(Trailer.trailer_number) + .filter( + Trailer.tenant_id == tenant_id, + Trailer.company_id == company_id, + ) + .all() + ): + if t[0] and (t[0] or "").strip(): + existing_trailer_numbers.add((t[0] or "").strip()) + except Exception as e: + logger.warning("Trailers import: could not load existing trailer_numbers for actualizar: %s", e) + + ( + valid_trailer_type_keys, + valid_country_ame, + state_descriptions_upper, + state_country_set, + state_ame_to_description, + ) = load_trailers_fk_sets(tenant_id, company_id) + error_count = 0 processed_rows = 0 errors_detail: List[Dict[str, Any]] = [] @@ -96,7 +125,19 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str, progress_callback(i, total_rows, error_count) row_norm = row_from_template(row, common_normalize.normalize_header) - err = validate_row_trailer(row_norm, i) + # Desfase: advertencia no bloqueante (no se añade a error_lines) + _ = validate_row_trailer_desfase(row_norm, i) + err = validate_row_trailer( + row_norm, + i, + actualizar=actualizar, + existing_trailer_numbers=existing_trailer_numbers, + valid_trailer_type_keys=valid_trailer_type_keys, + valid_country_ame=valid_country_ame, + state_descriptions_upper=state_descriptions_upper, + state_country_set=state_country_set, + state_ame_to_description=state_ame_to_description, + ) if err: error_count += 1 error_lines_list.append(err["line"]) @@ -170,6 +211,34 @@ 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) + existing_trailer_numbers: Set[str] = set() + if actualizar: + try: + from api.v1.modules.a76.transportation.trailers.models import Trailer + with CoreSessionLocal() as session: + for t in ( + session.query(Trailer.trailer_number) + .filter( + Trailer.tenant_id == tenant_id, + Trailer.company_id == company_id, + ) + .all() + ): + if t[0] and (t[0] or "").strip(): + existing_trailer_numbers.add((t[0] or "").strip()) + except Exception as e: + logger.warning("Trailers import: could not load existing trailer_numbers for actualizar: %s", e) + + ( + valid_trailer_type_keys, + valid_country_ame, + state_descriptions_upper, + state_country_set, + state_ame_to_description, + ) = load_trailers_fk_sets(tenant_id, company_id) + from api.v1.modules.a76.transportation.trailers.services import TrailerService from api.v1.modules.a76.transportation.trailers.dto import TrailerCreateDTO, TrailerUpdateDTO @@ -203,10 +272,20 @@ def _do_commit(job_id: str) -> Dict[str, Any]: continue row_norm = row_from_template(row, common_normalize.normalize_header) - err = validate_row_trailer(row_norm, i) + err = validate_row_trailer( + row_norm, + i, + actualizar=actualizar, + existing_trailer_numbers=existing_trailer_numbers, + valid_trailer_type_keys=valid_trailer_type_keys, + valid_country_ame=valid_country_ame, + state_descriptions_upper=state_descriptions_upper, + state_country_set=state_country_set, + state_ame_to_description=state_ame_to_description, + ) if err: skipped_invalid += 1 - tn = (row_norm.get("NUMERO TRAILER") or "").strip()[:20] or "-" + tn = (row_norm.get("NUMERO TRAILER") or row_norm.get("CLAVE TRAILER") or "").strip()[:20] or "-" skipped_details.append({ "line": i, "trailer_number": tn, @@ -215,12 +294,11 @@ def _do_commit(job_id: str) -> Dict[str, Any]: }) continue - data = row_to_trailer_data(row_norm) - if not data or not data.get("trailer_number"): + tn = (row_norm.get("NUMERO TRAILER") or row_norm.get("CLAVE TRAILER") or "").strip()[:20] or "" + if not tn: skipped_invalid += 1 continue - tn = data["trailer_number"] if tn in seen_keys_in_file: skipped_duplicate += 1 skipped_details.append({ @@ -235,10 +313,21 @@ def _do_commit(job_id: str) -> Dict[str, Any]: existing = TrailerService.get_by_id(session, tn, tenant_id, company_id) try: if existing: + if actualizar: + data = row_to_trailer_data_for_update(row_norm, existing) + else: + data = row_to_trailer_data(row_norm) + if not data or not data.get("trailer_number"): + skipped_invalid += 1 + continue update_data = TrailerUpdateDTO(**{k: v for k, v in data.items() if k != "trailer_number"}) TrailerService.update(session, tn, tenant_id, update_data, company_id) updated_count += 1 else: + data = row_to_trailer_data(row_norm) + if not data or not data.get("trailer_number"): + skipped_invalid += 1 + continue create_data = TrailerCreateDTO(**data) TrailerService.create(session, create_data, tenant_id, company_id) inserted_count += 1 diff --git a/backend/api/v1/modules/a76/layouts_csv/trailers/template_config.py b/backend/api/v1/modules/a76/layouts_csv/trailers/template_config.py index be0eb4a7..31c35f12 100644 --- a/backend/api/v1/modules/a76/layouts_csv/trailers/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/trailers/template_config.py @@ -16,6 +16,7 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { {"canonical": "ESTADO", "aliases": ["STATE"]}, {"canonical": "PAIS", "aliases": ["COUNTRY", "Pais"]}, {"canonical": "CLAVE CONTENEDOR", "aliases": ["CONTAINER", "CONTAINER KEY", "CONTENEDOR"]}, + {"canonical": "COL_EXTRA", "aliases": ["COLUMNA I", "COL I", "COL 9"]}, ], } diff --git a/backend/api/v1/modules/a76/layouts_csv/trailers/validators/__init__.py b/backend/api/v1/modules/a76/layouts_csv/trailers/validators/__init__.py index 406c2f15..456065a3 100644 --- a/backend/api/v1/modules/a76/layouts_csv/trailers/validators/__init__.py +++ b/backend/api/v1/modules/a76/layouts_csv/trailers/validators/__init__.py @@ -1,3 +1,3 @@ -from .create import validate_row_trailer +from .create import validate_row_trailer, validate_row_trailer_desfase -__all__ = ["validate_row_trailer"] +__all__ = ["validate_row_trailer", "validate_row_trailer_desfase"] diff --git a/backend/api/v1/modules/a76/layouts_csv/trailers/validators/common.py b/backend/api/v1/modules/a76/layouts_csv/trailers/validators/common.py index d10b3760..056ce6f2 100644 --- a/backend/api/v1/modules/a76/layouts_csv/trailers/validators/common.py +++ b/backend/api/v1/modules/a76/layouts_csv/trailers/validators/common.py @@ -1,12 +1,18 @@ """ Validaciones comunes de fila para import CSV de trailers. +Paridad Clarion: VALIDACIONES_TRAILER, VALIDA_TODA_TRAILER, VALIDA_PARCIAL_TRAILER. """ -from typing import Dict, Any, Optional +from typing import Dict, Any, Optional, Set, Tuple from ..common.common_validators import ( MAX_LEN, check_required, check_max_length, + check_codigo_entidad_valores, + check_trailer_type_catalog, + check_pais_catalog_trailers, + check_estado_catalog_trailers, + check_estado_pais_consistency_trailers, ) @@ -20,6 +26,7 @@ def validate_row_trailer_lengths(row: Dict[str, Any], line_num: int) -> Optional ("TIPO TRAILER", MAX_LEN["trailer_type_key"]), ("PRECINTO", MAX_LEN["seal"]), ("CODIGO ENTIDAD", MAX_LEN["entity_code"]), + ("CODIGO DE ENTIDAD", MAX_LEN["entity_code"]), ("PLACAS", MAX_LEN["plate_number"]), ("ESTADO", MAX_LEN["state"]), ("PAIS", MAX_LEN["country"]), @@ -30,3 +37,86 @@ def validate_row_trailer_lengths(row: Dict[str, Any], line_num: int) -> Optional if err: return err return None + + +def validaciones_trailer( + row: Dict[str, Any], + line_num: int, + valid_trailer_type_keys: Optional[Set[str]] = None, + valid_country_ame: Optional[Set[str]] = None, + state_descriptions_upper: Optional[Set[str]] = None, + state_country_set: Optional[Set[Tuple[str, str]]] = None, + state_ame_to_description: Optional[Dict[str, str]] = None, +) -> Optional[Dict[str, Any]]: + """ + VALIDACIONES_TRAILER: reglas compartidas (longitudes, tipo trailer, código entidad, + país, estado, estado-país). No exige CODIGO DE ENTIDAD obligatorio. + """ + err = validate_row_trailer_required(row, line_num) + if err: + return err + err = validate_row_trailer_lengths(row, line_num) + if err: + return err + err = check_codigo_entidad_valores(row, line_num) + if err: + return err + err = check_trailer_type_catalog(row, line_num, valid_trailer_type_keys) + if err: + return err + err = check_pais_catalog_trailers(row, line_num, valid_country_ame) + if err: + return err + err_estado, estado_resuelto_upper = check_estado_catalog_trailers( + row, line_num, state_descriptions_upper, state_ame_to_description + ) + if err_estado: + return err_estado + err = check_estado_pais_consistency_trailers( + row, line_num, state_country_set, estado_resuelto_upper + ) + if err: + return err + return None + + +def valida_toda_trailer( + row: Dict[str, Any], + line_num: int, + valid_trailer_type_keys: Optional[Set[str]] = None, + valid_country_ame: Optional[Set[str]] = None, + state_descriptions_upper: Optional[Set[str]] = None, + state_country_set: Optional[Set[Tuple[str, str]]] = None, + state_ame_to_description: Optional[Dict[str, str]] = None, +) -> Optional[Dict[str, Any]]: + """VALIDA_TODA_TRAILER: mismo que validaciones_trailer (Col A ya validada antes).""" + return validaciones_trailer( + row, + line_num, + valid_trailer_type_keys=valid_trailer_type_keys, + valid_country_ame=valid_country_ame, + state_descriptions_upper=state_descriptions_upper, + state_country_set=state_country_set, + state_ame_to_description=state_ame_to_description, + ) + + +def valida_parcial_trailer( + row: Dict[str, Any], + line_num: int, + valid_trailer_type_keys: Optional[Set[str]] = None, + valid_country_ame: Optional[Set[str]] = None, + state_descriptions_upper: Optional[Set[str]] = None, + state_country_set: Optional[Set[Tuple[str, str]]] = None, + state_ame_to_description: Optional[Dict[str, str]] = None, +) -> Optional[Dict[str, Any]]: + """VALIDA_PARCIAL_TRAILER: solo validaciones_trailer (actualizar registro existente).""" + return validaciones_trailer( + row, + line_num, + valid_trailer_type_keys=valid_trailer_type_keys, + valid_country_ame=valid_country_ame, + state_descriptions_upper=state_descriptions_upper, + state_country_set=state_country_set, + state_ame_to_description=state_ame_to_description, + ) diff --git a/backend/api/v1/modules/a76/layouts_csv/trailers/validators/create.py b/backend/api/v1/modules/a76/layouts_csv/trailers/validators/create.py index d01491b5..cc8bdf39 100644 --- a/backend/api/v1/modules/a76/layouts_csv/trailers/validators/create.py +++ b/backend/api/v1/modules/a76/layouts_csv/trailers/validators/create.py @@ -1,22 +1,68 @@ """ Punto de entrada de validación para import de una fila trailer. +Paridad Clarion: desfase (advertencia), CLAVE vacía, VALIDA_TODA vs VALIDA_PARCIAL según actualizar y clave existente. """ -from typing import Dict, Any, Optional +from typing import Dict, Any, Optional, Set, Tuple from .common import ( validate_row_trailer_required, - validate_row_trailer_lengths, + valida_toda_trailer, + valida_parcial_trailer, ) +from ..common.common_validators import check_desfase_trailers -def validate_row_trailer(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: +def validate_row_trailer( + row: Dict[str, Any], + line_num: int, + actualizar: bool = False, + existing_trailer_numbers: Optional[Set[str]] = None, + valid_trailer_type_keys: Optional[Set[str]] = None, + valid_country_ame: Optional[Set[str]] = None, + state_descriptions_upper: Optional[Set[str]] = None, + state_country_set: Optional[Set[Tuple[str, str]]] = None, + state_ame_to_description: Optional[Dict[str, str]] = None, +) -> Optional[Dict[str, Any]]: """ Valida una fila de CSV de trailers. + 1. Clave (NUMERO TRAILER) vacía → error. + 2. Si actualizar y clave en existing_trailer_numbers → valida_parcial_trailer. + 3. Si no actualizar o clave no existe → valida_toda_trailer. + Desfase (COL_EXTRA) no se valida aquí; el caller puede llamar validate_row_trailer_desfase para advertencias no bloqueantes. """ err = validate_row_trailer_required(row, line_num) if err: return err - err = validate_row_trailer_lengths(row, line_num) - if err: - return err - return None + + existing = existing_trailer_numbers or set() + clave = (row.get("NUMERO TRAILER") or row.get("CLAVE TRAILER") or "").strip() + use_partial = actualizar and bool(clave and clave in existing) + + if use_partial: + err = valida_parcial_trailer( + row, + line_num, + valid_trailer_type_keys=valid_trailer_type_keys, + valid_country_ame=valid_country_ame, + state_descriptions_upper=state_descriptions_upper, + state_country_set=state_country_set, + state_ame_to_description=state_ame_to_description, + ) + else: + err = valida_toda_trailer( + row, + line_num, + valid_trailer_type_keys=valid_trailer_type_keys, + valid_country_ame=valid_country_ame, + state_descriptions_upper=state_descriptions_upper, + state_country_set=state_country_set, + state_ame_to_description=state_ame_to_description, + ) + return err + + +def validate_row_trailer_desfase(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + """ + Advertencia de desfase si COL_EXTRA tiene valor. No bloqueante; el caller puede acumular en warnings. + """ + return check_desfase_trailers(row, line_num)