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

feature/trailes-validaciones-clarion-csv

Browse Source
This commit is contained in:
hreyes
2026-03-06 09:03:28 -07:00
parent 6a902a414a
commit 2df4a029e0
10 changed files with 533 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/api/v1/modules/a76/layouts_csv/trailers/common/__init__.py
View File

@@ -1 +1 @@
# common_validators, mappers (no fk_loader for trailers)
# common_validators, mappers, fk_loader

146
backend/api/v1/modules/a76/layouts_csv/trailers/common/common_validators.py
View File

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

78
backend/api/v1/modules/a76/layouts_csv/trailers/common/fk_loader.py Normal file
View File

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

65
backend/api/v1/modules/a76/layouts_csv/trailers/common/mappers.py
View File

@@ -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"]),
}

2
backend/api/v1/modules/a76/layouts_csv/trailers/routes.py
View File

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

107
backend/api/v1/modules/a76/layouts_csv/trailers/tasks.py
View File

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

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

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

4
backend/api/v1/modules/a76/layouts_csv/trailers/validators/__init__.py
View File

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

92
backend/api/v1/modules/a76/layouts_csv/trailers/validators/common.py
View File

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

60
backend/api/v1/modules/a76/layouts_csv/trailers/validators/create.py
View File

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