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

feature/validaciones-clarion-csv-drivers

Browse Source
This commit is contained in:
hreyes
2026-03-06 12:21:19 -07:00
parent 8cd299609c
commit dfcf4aad6b
12 changed files with 616 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
backend/api/v1/modules/a76/layouts_csv/drivers/common/__init__.py
View File

@@ -1 +1,4 @@
# common validators, mappers for drivers CSV import
# common validators, mappers, fk_loader for drivers CSV import
from .fk_loader import load_drivers_fk_sets
__all__ = ["load_drivers_fk_sets"]

145
backend/api/v1/modules/a76/layouts_csv/drivers/common/common_validators.py
View File

@@ -1,8 +1,9 @@
"""
Helpers reutilizables para validación de filas CSV (conductores).
Paridad Clarion: VALIDACIONES_CONDUCTOR, obligatorios A/C, catálogos transportista/países, Sexo M/F, Si/No, tipo identificación, desfase.
"""
import re
from typing import Dict, Any, Optional
from typing import Dict, Any, Optional, Set
MAX_LEN = {
@@ -27,6 +28,33 @@ MAX_LEN = {
"id_country2": 3,
}
# Clarion: Col H Sexo M o F
SEXO_VALIDOS = {"M", "F"}
# Clarion: Col J Si o No (comparar en mayúsculas)
MATERIAL_PELIGROSO_VALIDOS = {"SI", "NO"}
# Clarion: Col N y Col R tipo identificación
FORMA_IDENTIFICACION_CLAVES = frozenset(
{"ACW", "ALR", "BCP", "BCN", "CDN", "CON", "OTD", "REP", "RTP", "5J", "5K", "30"}
)
# Mapeo clave CSV → valor guardado en BD (Clarion QueCSV:ColumnaN/R)
FORMA_IDENTIFICACION_MAP = {
"ACW": "ACW-Pasaporte",
"ALR": "ALR-Residencia",
"BCP": "BCP-Permiso Cruce",
"BCN": "BCN-Acta Nacimiento",
"CDN": "CDN-Ciudadania",
"CON": "CON-CertificadoNaturalizacion",
"OTD": "OTD-Otra Identificación",
"REP": "REP-Permiso Reentrada",
"RTP": "RTP-Permiso de Viaje",
"5J": "5J - Licencia",
"5K": "5K -Licencia",
"30": "30 -Visa de EU",
}
def check_required(row: Dict[str, Any], col: str, line_num: int) -> Optional[Dict[str, Any]]:
val = (row.get(col) or "").strip()
@@ -116,3 +144,118 @@ def check_optional_birth_date(row: Dict[str, Any], col: str, line_num: int) -> O
"msg": "Formato de fecha invalido (use YYYYMMDD o DD/MM/YYYY)",
}
return None
def check_transportista_catalog(
row: Dict[str, Any],
line_num: int,
valid_transporter_keys: Optional[Set[str]] = None,
) -> Optional[Dict[str, Any]]:
"""Col A: Si TRANSPORTISTA no vacío, debe existir en catálogo (GTransportista)."""
val = (row.get("TRANSPORTISTA") or "").strip()
if not val or valid_transporter_keys is None:
return None
if val.upper() in valid_transporter_keys:
return None
return {
"line": line_num,
"col": "TRANSPORTISTA",
"msg": f"Error: (Col. A) La Clave de Transportista: {val} es incorrecto.",
"solution": "Capturar en columna A un Transportista existente en catalogo.",
}
def check_sexo_m_f(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Col H: Si SEXO no vacío, debe ser M o F (Clarion)."""
val = (row.get("SEXO") or "").strip()
if not val:
return None
if val.upper() in SEXO_VALIDOS:
return None
conductor = (row.get("CLAVE CONDUCTOR") or "").strip() or "(Conductor)"
return {
"line": line_num,
"col": "SEXO",
"msg": f"Error: (Col. H) El Sexo: {val} del Conductor: {conductor} es incorrecto.",
"solution": "Capturar en columna H el sexo correcto (M o F).",
}
def check_pais_catalog_drivers(
row: Dict[str, Any],
col: str,
line_num: int,
valid_country_ame: Optional[Set[str]] = None,
col_letter: str = "",
) -> Optional[Dict[str, Any]]:
"""Si col (PAIS NACIMIENTO, PAIS, PAIS 2) no vacío, debe ser clave americana en catálogo (GPaises.Pais_Ame)."""
val = (row.get(col) or "").strip()
if not val or valid_country_ame is None:
return None
val_upper = val.upper()
if len(val) > 3:
return {
"line": line_num,
"col": col,
"msg": f"Error: ({col_letter}) El Pais: {val} es incorrecto.",
"solution": "Capturar un Pais en clave americana (US, MX, etc.).",
}
if val_upper in valid_country_ame:
return None
return {
"line": line_num,
"col": col,
"msg": f"Error: ({col_letter}) El Pais: {val} es incorrecto.",
"solution": "Capturar en columna un Pais en clave americana.",
}
def check_material_peligroso_si_no(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Col J: TRANSPORTA MAT. PELIGROSO? debe ser Si o No (Clarion; comparar en mayúsculas)."""
val = (row.get("TRANSPORTA MAT. PELIGROSO?") or "").strip()
if not val:
return None
if val.upper() in MATERIAL_PELIGROSO_VALIDOS:
return None
return {
"line": line_num,
"col": "TRANSPORTA MAT. PELIGROSO?",
"msg": "Error: (Col. J) La Autorizacion para el Manejo de Material Peligroso es incorrecta.",
"solution": "Capturar en columna J Si o No la autorizacion.",
}
def check_tipo_identificacion(
row: Dict[str, Any],
col: str,
line_num: int,
col_letter: str = "",
primera_o_segunda: str = "Primera",
) -> Optional[Dict[str, Any]]:
"""Col N o R: Si FORMA IDENTIFICACION no vacío, debe ser clave válida (ACW, ALR, ...)."""
val = (row.get(col) or "").strip()
if not val:
return None
val_upper = val.upper()
if val_upper in FORMA_IDENTIFICACION_CLAVES:
return None
return {
"line": line_num,
"col": col,
"msg": f"Error: ({col_letter}) El Tipo de Identificacion: {val} de la {primera_o_segunda} Identificacion es incorrecto.",
"solution": f"Capturar en columna {col_letter} una clave de Identificacion valida (ACW, ALR, BCP, BCN, CDN, CON, OTD, REP, RTP, 5J, 5K, 30).",
}
def check_desfase_drivers(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Si COL_EXTRA (Col V) 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: Podria existir un desfase en esta linea.",
"solution": "Revisar esta linea del archivo CSV y verificar cada campo este en la posicion correcta.",
"warning": True,
}

58
backend/api/v1/modules/a76/layouts_csv/drivers/common/fk_loader.py Normal file
View File

@@ -0,0 +1,58 @@
"""
Carga de conjuntos FK para validación de import CSV de conductores.
Clarion: GTransportista (ClaveTrans), GPaises (Pais_Ame).
"""
from typing import Set, Tuple, Optional, Dict
import logging
from core.database import CoreSessionLocal
logger = logging.getLogger(__name__)
def load_drivers_fk_sets(
tenant_id: Optional[int] = None,
company_id: Optional[int] = None,
) -> Tuple[Set[str], Set[str], Dict[str, str]]:
"""
Carga conjuntos para validación CSV de conductores (paridad Clarion).
Devuelve:
- valid_transporter_keys: todas las claves de transportistas en mayúsculas (a76.transporter)
- valid_country_ame: claves americana de países (GPaises.Pais_Ame / Country.ame_key), mayúsculas
- transporter_key_actual: dict clave_upper -> clave real en BD (para insert con mismo caso que en transporter)
"""
valid_transporter_keys: Set[str] = set()
valid_country_ame: Set[str] = set()
transporter_key_actual: Dict[str, str] = {}
try:
with CoreSessionLocal() as session:
from api.v1.modules.a76.transportation.transporters.models import Transporter
from api.v1.modules.public.reference_data.countries.models import Country
# Solo transportistas del tenant/company del upload (paridad con Driver.tenant_id/company_id)
q = session.query(Transporter.transporter_key).filter(
Transporter.tenant_id == tenant_id,
Transporter.company_id == company_id,
)
for row in q.all():
if row[0]:
raw = (row[0] or "").strip()
upper = raw.upper()
valid_transporter_keys.add(upper)
transporter_key_actual[upper] = raw
for row in session.query(Country.ame_key).all():
if row[0]:
valid_country_ame.add((row[0] or "").strip().upper())
except Exception as e:
logger.warning("Drivers import: could not load FK sets: %s", e)
logger.info(
"Drivers import FK: %d transportistas (claves: %s), %d paises",
len(valid_transporter_keys),
sorted(valid_transporter_keys)[:20] if len(valid_transporter_keys) <= 20 else sorted(valid_transporter_keys)[:10] + ["..."],
len(valid_country_ame),
)
return (valid_transporter_keys, valid_country_ame, transporter_key_actual)

21
backend/api/v1/modules/a76/layouts_csv/drivers/common/mappers.py
View File

@@ -1,5 +1,6 @@
"""
Mapeo fila CSV → datos para Driver (conductores).
Clarion: FORMA IDENTIFICACION 1/2 se guardan expandidas (ACW → ACW-Pasaporte, etc.).
"""
from typing import Dict, Any, Optional
@@ -7,6 +8,7 @@ from .common_validators import (
MAX_LEN,
parse_int,
parse_birth_date,
FORMA_IDENTIFICACION_MAP,
)
@@ -21,6 +23,17 @@ def _str_or_none(val: Any, max_len: Optional[int] = None) -> Optional[str]:
return s
def _forma_identificacion_or_raw(val: Any, max_len: int) -> Optional[str]:
"""Si el valor es una clave Clarion (ACW, ALR, ...), devuelve el valor expandido; si no, el valor truncado."""
s = _str_or_none(val, max_len)
if not s:
return None
expanded = FORMA_IDENTIFICACION_MAP.get(s.upper())
if expanded:
return expanded[:max_len] if len(expanded) > max_len else expanded
return s
def row_to_driver_data(
row_norm: Dict[str, Any],
tenant_id: int,
@@ -49,11 +62,15 @@ def row_to_driver_data(
),
"first_name": _str_or_none(row_norm.get("NOMBRE(S)"), MAX_LEN["first_name"]),
"last_name": _str_or_none(row_norm.get("APELLIDO PATERNO"), MAX_LEN["last_name"]),
"id_key1": _str_or_none(row_norm.get("FORMA IDENTIFICACION 1"), MAX_LEN["id_key1"]),
"id_key1": _forma_identificacion_or_raw(
row_norm.get("FORMA IDENTIFICACION 1"), MAX_LEN["id_key1"]
),
"id_number1": _str_or_none(row_norm.get("NUM. IDENTIFICACION 1"), MAX_LEN["id_number1"]),
"id_state1": _str_or_none(row_norm.get("ESTADO"), MAX_LEN["id_state1"]),
"id_country1": _str_or_none(row_norm.get("PAIS"), MAX_LEN["id_country1"]),
"id_key2": _str_or_none(row_norm.get("FORMA IDENTIFICACION 2"), MAX_LEN["id_key2"]),
"id_key2": _forma_identificacion_or_raw(
row_norm.get("FORMA IDENTIFICACION 2"), MAX_LEN["id_key2"]
),
"id_number2": _str_or_none(row_norm.get("NUM. IDENTIFICACION 2"), MAX_LEN["id_number2"]),
"id_state2": _str_or_none(row_norm.get("ESTADO 2"), MAX_LEN["id_state2"]),
"id_country2": _str_or_none(row_norm.get("PAIS 2"), MAX_LEN["id_country2"]),

3
backend/api/v1/modules/a76/layouts_csv/drivers/routes.py
View File

@@ -20,7 +20,6 @@ from core.security import get_current_user, validate_access_to_resource
from .schemas import ImportJobResponse
from .tasks import (
scan_file,
run_scan_sync,
run_commit_sync,
DRV_IMPORT_FILE_PREFIX,
@@ -91,8 +90,6 @@ async def upload_import_file(
except Exception as e:
logger.warning(f"Drivers import: local file save failed: {e}")
scan_file.apply_async(args=[job_id], task_id=job_id)
def run_scan_background():
try:
run_scan_sync(job_id)

163
backend/api/v1/modules/a76/layouts_csv/drivers/tasks.py
View File

@@ -2,23 +2,26 @@
Tareas Celery para importación CSV de Conductores.
Flujo: scan_file (validación) → insert_valid_rows (commit).
Usa layouts_csv.common (storage, normalize, meta, responses); CSV con headers duplicados (dedupe) y clave de estado en Redis.
Paridad Clarion: actualizar, existing_driver_keys, valid_transporter_keys, valid_country_ame.
"""
import csv
import json
import logging
import os
from typing import Dict, Any, Optional, List
from typing import Dict, Any, Optional, List, Set, Tuple
from core.celery_app import celery_app
from core.database import CoreSessionLocal
from sqlalchemy import func
from ..common import storage as common_storage
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_driver
from .validators import validate_row_driver, validate_row_driver_desfase
from .common.mappers import row_to_driver_data
from .common.fk_loader import load_drivers_fk_sets
logger = logging.getLogger(__name__)
@@ -30,6 +33,7 @@ DRV_IMPORT_META_PREFIX = "drv_import_meta:"
DRV_IMPORT_ERROR_LINES_PREFIX = "drv_import_error_lines:"
DRV_IMPORT_STATUS_PREFIX = "drv_import_status:"
DRV_IMPORT_REDIS_TTL = common_storage.IMPORT_REDIS_TTL
DRV_IMPORT_TRANSPORTER_MAP_PREFIX = "drv_import_transporter_map:"
def _get_redis():
@@ -68,6 +72,45 @@ 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_driver_keys: Set[Tuple[str, int]] = set()
if actualizar:
try:
from api.v1.modules.a76.transportation.drivers.models import Driver
with CoreSessionLocal() as session:
for row in (
session.query(Driver.transporter_key, Driver.line)
.filter(
Driver.tenant_id == tenant_id,
Driver.company_id == company_id,
)
.all()
):
if row[0] is not None and row[1] is not None:
existing_driver_keys.add(
((row[0] or "").strip().upper(), int(row[1]))
)
except Exception as e:
logger.warning("Drivers import: could not load existing_driver_keys for actualizar: %s", e)
valid_transporter_keys, valid_country_ame, transporter_key_actual = load_drivers_fk_sets(tenant_id, company_id)
try:
r = _get_redis()
r.set(
f"{DRV_IMPORT_TRANSPORTER_MAP_PREFIX}{job_id}",
json.dumps(transporter_key_actual).encode("utf-8"),
ex=DRV_IMPORT_REDIS_TTL,
)
except Exception as e:
logger.warning("Drivers import: failed to store transporter map in Redis: %s", e)
logger.info(
"Drivers import scan: job_id=%s tenant_id=%s company_id=%s actualizar=%s transportistas=%d",
job_id, tenant_id, company_id, actualizar, len(valid_transporter_keys),
)
error_count = 0
processed_rows = 0
errors_detail: List[Dict[str, Any]] = []
@@ -96,7 +139,15 @@ 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_driver(row_norm, i)
_ = validate_row_driver_desfase(row_norm, i)
err = validate_row_driver(
row_norm,
i,
actualizar=actualizar,
existing_driver_keys=existing_driver_keys,
valid_transporter_keys=valid_transporter_keys,
valid_country_ame=valid_country_ame,
)
if err:
error_count += 1
error_lines_list.append(err["line"])
@@ -170,6 +221,57 @@ 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_driver_keys: Set[Tuple[str, int]] = set()
if actualizar:
try:
from api.v1.modules.a76.transportation.drivers.models import Driver
with CoreSessionLocal() as session:
for row in (
session.query(Driver.transporter_key, Driver.line)
.filter(
Driver.tenant_id == tenant_id,
Driver.company_id == company_id,
)
.all()
):
if row[0] is not None and row[1] is not None:
existing_driver_keys.add(
((row[0] or "").strip().upper(), int(row[1]))
)
except Exception as e:
logger.warning("Drivers import: could not load existing_driver_keys for actualizar: %s", e)
valid_transporter_keys, valid_country_ame, transporter_key_actual = load_drivers_fk_sets(tenant_id, company_id)
transporter_map_from_redis: Optional[Dict[str, str]] = None
try:
r = _get_redis()
map_key = f"{DRV_IMPORT_TRANSPORTER_MAP_PREFIX}{job_id}"
raw = r.get(map_key)
if raw:
transporter_map_from_redis = json.loads(raw.decode("utf-8"))
logger.info(
"Drivers import commit: using transporter map from Redis (job_id=%s, keys=%d)",
job_id, len(transporter_map_from_redis),
)
else:
logger.warning(
"Drivers import commit: no transporter map in Redis for job_id=%s, using DB fallback",
job_id,
)
except Exception as e:
logger.warning(
"Drivers import: could not load transporter map from Redis (job_id=%s): %s",
job_id, e,
)
logger.info(
"Drivers import commit: job_id=%s tenant_id=%s company_id=%s transportistas=%d",
job_id, tenant_id, company_id, len(valid_transporter_keys),
)
from api.v1.modules.a76.transportation.drivers.services import DriverService
from api.v1.modules.a76.transportation.drivers.dto import DriverCreateDTO
@@ -203,7 +305,14 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
continue
row_norm = row_from_template(row, common_normalize.normalize_header)
err = validate_row_driver(row_norm, i)
err = validate_row_driver(
row_norm,
i,
actualizar=actualizar,
existing_driver_keys=existing_driver_keys,
valid_transporter_keys=valid_transporter_keys,
valid_country_ame=valid_country_ame,
)
if err:
skipped_invalid += 1
driver_key = (row_norm.get("CLAVE CONDUCTOR") or "").strip()[:80] or "-"
@@ -220,6 +329,42 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
skipped_invalid += 1
continue
tk = (data["transporter_key"] or "").strip()
# Usar mapa del scan (Redis) si existe; si no, resolver en la sesión del commit (fallback)
if transporter_map_from_redis is not None:
tk_upper = tk.upper()
if tk_upper not in transporter_map_from_redis:
skipped_invalid += 1
skipped_details.append({
"line": i,
"driver_key": f"{tk}:{data.get('line')}",
"invoice": f"{tk}:{data.get('line')}",
"reason": f"El transportista {tk} no existe en el catálogo.",
})
continue
data["transporter_key"] = transporter_map_from_redis[tk_upper]
else:
from api.v1.modules.a76.transportation.transporters.models import Transporter
transporter_row = (
session.query(Transporter.transporter_key)
.filter(
Transporter.tenant_id == tenant_id,
Transporter.company_id == company_id,
func.upper(Transporter.transporter_key) == tk.upper(),
)
.first()
)
if not transporter_row or not transporter_row[0]:
skipped_invalid += 1
skipped_details.append({
"line": i,
"driver_key": f"{tk}:{data.get('line')}",
"invoice": f"{tk}:{data.get('line')}",
"reason": f"El transportista {tk} no existe en el catálogo.",
})
continue
data["transporter_key"] = (transporter_row[0] or "").strip()
key = f"{data['transporter_key']}:{data['line']}"
if key in seen_keys_in_file:
skipped_duplicate += 1
@@ -252,8 +397,15 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
except Exception as db_err:
session.rollback()
skipped_invalid += 1
err_msg = str(db_err)
if "ForeignKeyViolation" in err_msg or "foreign key constraint" in err_msg.lower() or "driver_transporter_key_fkey" in err_msg:
err_msg = f"El transportista {data.get('transporter_key', '')} no existe en el catálogo."
logger.warning(
"Drivers import: FK violation linea %d transporter_key=%r (valid_transporter_keys tiene %d claves)",
i, data.get("transporter_key"), len(valid_transporter_keys),
)
skipped_details.append({
"line": i, "driver_key": key, "invoice": key, "reason": str(db_err),
"line": i, "driver_key": key, "invoice": key, "reason": err_msg,
})
continue
@@ -277,6 +429,7 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
try:
r = _get_redis()
r.delete(f"{DRV_IMPORT_STATUS_PREFIX}{job_id}")
r.delete(f"{DRV_IMPORT_TRANSPORTER_MAP_PREFIX}{job_id}")
except Exception as e:
logger.warning("Drivers import: failed to delete status key: %s", e)

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

@@ -27,6 +27,7 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
{"canonical": "NUM. IDENTIFICACION 2", "aliases": ["NUM IDENTIFICACION 2", "ID NUMERO 2", "ID NUMBER 2"]},
{"canonical": "ESTADO 2", "aliases": ["STATE 2"]},
{"canonical": "PAIS 2", "aliases": ["COUNTRY 2"]},
{"canonical": "COL_EXTRA", "aliases": ["COLUMNA V", "COL V"]},
],
}

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

@@ -1,3 +1,3 @@
from .create import validate_row_driver
from .create import validate_row_driver, validate_row_driver_desfase
__all__ = ["validate_row_driver"]
__all__ = ["validate_row_driver", "validate_row_driver_desfase"]

114
backend/api/v1/modules/a76/layouts_csv/drivers/validators/common.py
View File

@@ -1,20 +1,40 @@
"""
Validaciones comunes de fila para import CSV de conductores.
Paridad Clarion: VALIDACIONES_CONDUCTOR, VALIDA_TODA_CONDUCTOR, VALIDA_PARCIAL_CONDUCTOR.
"""
from typing import Dict, Any, Optional
from typing import Dict, Any, Optional, Set
from ..common.common_validators import (
MAX_LEN,
check_max_length,
check_int_positive,
check_optional_birth_date,
check_transportista_catalog,
check_sexo_m_f,
check_pais_catalog_drivers,
check_material_peligroso_si_no,
check_tipo_identificacion,
)
def validate_row_driver_required(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Obligatorios Clarion: Col A (TRANSPORTISTA) y Col C (CLAVE CONDUCTOR). Si falta uno, no se ejecutan validaciones."""
err = check_max_length(
row, "TRANSPORTISTA", MAX_LEN["transporter_key"], line_num, required=True
)
if err:
return err
err = check_max_length(
row, "CLAVE CONDUCTOR", MAX_LEN["driver_name"], line_num, required=True
)
if err:
return err
return None
def validate_row_driver_required_full(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""VALIDA_TODA: obligatorios A, C y LINEA (Col B) para registro nuevo."""
err = validate_row_driver_required(row, line_num)
if err:
return err
return check_int_positive(row, "LINEA", line_num)
@@ -50,3 +70,95 @@ def validate_row_driver_lengths(row: Dict[str, Any], line_num: int) -> Optional[
def validate_row_driver_date(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
return check_optional_birth_date(row, "FECHA NACIMIENTO", line_num)
def validaciones_conductores(
row: Dict[str, Any],
line_num: int,
valid_transporter_keys: Optional[Set[str]] = None,
valid_country_ame: Optional[Set[str]] = None,
) -> Optional[Dict[str, Any]]:
"""
VALIDACIONES_CONDUCTOR: reglas compartidas (longitudes, fecha, transportista en catálogo,
sexo M/F, país nacimiento, material peligroso Si/No, tipo ID 1/2, país ID1/ID2).
"""
err = validate_row_driver_lengths(row, line_num)
if err:
return err
err = validate_row_driver_date(row, line_num)
if err:
return err
err = check_transportista_catalog(row, line_num, valid_transporter_keys)
if err:
return err
err = check_sexo_m_f(row, line_num)
if err:
return err
err = check_pais_catalog_drivers(
row, "PAIS NACIMIENTO", line_num, valid_country_ame, col_letter="Col. I"
)
if err:
return err
err = check_material_peligroso_si_no(row, line_num)
if err:
return err
err = check_tipo_identificacion(
row,
"FORMA IDENTIFICACION 1",
line_num,
col_letter="Col. N",
primera_o_segunda="Primera",
)
if err:
return err
err = check_pais_catalog_drivers(
row, "PAIS", line_num, valid_country_ame, col_letter="Col. Q"
)
if err:
return err
err = check_tipo_identificacion(
row,
"FORMA IDENTIFICACION 2",
line_num,
col_letter="Col. R",
primera_o_segunda="Segunda",
)
if err:
return err
err = check_pais_catalog_drivers(
row, "PAIS 2", line_num, valid_country_ame, col_letter="Col. U"
)
if err:
return err
return None
def valida_toda_conductor(
row: Dict[str, Any],
line_num: int,
valid_transporter_keys: Optional[Set[str]] = None,
valid_country_ame: Optional[Set[str]] = None,
) -> Optional[Dict[str, Any]]:
"""VALIDA_TODA_CONDUCTOR: obligatorios A, C y LINEA (B) + validaciones_conductores (registro nuevo)."""
err = validate_row_driver_required_full(row, line_num)
if err:
return err
return validaciones_conductores(
row, line_num,
valid_transporter_keys=valid_transporter_keys,
valid_country_ame=valid_country_ame,
)
def valida_parcial_conductor(
row: Dict[str, Any],
line_num: int,
valid_transporter_keys: Optional[Set[str]] = None,
valid_country_ame: Optional[Set[str]] = None,
) -> Optional[Dict[str, Any]]:
"""VALIDA_PARCIAL_CONDUCTOR: solo validaciones_conductores (actualizar registro existente)."""
return validaciones_conductores(
row, line_num,
valid_transporter_keys=valid_transporter_keys,
valid_country_ame=valid_country_ame,
)

61
backend/api/v1/modules/a76/layouts_csv/drivers/validators/create.py
View File

@@ -1,27 +1,64 @@
"""
Punto de entrada de validación para import de una fila conductor.
Paridad Clarion: desfase (advertencia), obligatorios A y C, 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.common_validators import parse_int, check_desfase_drivers
from .common import (
validate_row_driver_required,
validate_row_driver_lengths,
validate_row_driver_date,
valida_toda_conductor,
valida_parcial_conductor,
)
def validate_row_driver(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
def validate_row_driver(
row: Dict[str, Any],
line_num: int,
actualizar: bool = False,
existing_driver_keys: Optional[Set[Tuple[str, int]]] = None,
valid_transporter_keys: Optional[Set[str]] = None,
valid_country_ame: Optional[Set[str]] = None,
) -> Optional[Dict[str, Any]]:
"""
Valida una fila de CSV de conductores.
Encadena: requeridos (TRANSPORTISTA, LINEA) → longitudes → fecha opcional.
1. Obligatorios A (TRANSPORTISTA) y C (CLAVE CONDUCTOR) vacíos → error.
2. Si actualizar y (TRANSPORTISTA, LINEA) en existing_driver_keys → valida_parcial_conductor.
3. Si no actualizar o conductor no existe → valida_toda_conductor (obligatorios A, C y LINEA + validaciones).
Desfase (COL_EXTRA) no se valida aquí; el caller puede llamar validate_row_driver_desfase para advertencias no bloqueantes.
"""
err = validate_row_driver_required(row, line_num)
if err:
return err
err = validate_row_driver_lengths(row, line_num)
if err:
return err
err = validate_row_driver_date(row, line_num)
if err:
return err
return None
existing = existing_driver_keys or set()
transporter_key = (row.get("TRANSPORTISTA") or "").strip().upper()
line = parse_int(row.get("LINEA"))
use_partial = (
actualizar
and bool(transporter_key and line is not None)
and (transporter_key, line) in existing
)
if use_partial:
err = valida_parcial_conductor(
row,
line_num,
valid_transporter_keys=valid_transporter_keys,
valid_country_ame=valid_country_ame,
)
else:
err = valida_toda_conductor(
row,
line_num,
valid_transporter_keys=valid_transporter_keys,
valid_country_ame=valid_country_ame,
)
return err
def validate_row_driver_desfase(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""
Advertencia de desfase si COL_EXTRA (Col V) tiene valor. No bloqueante; el caller puede acumular en warnings.
"""
return check_desfase_drivers(row, line_num)

41
backend/api/v1/modules/a76/transportation/drivers/routes.py
View File

@@ -8,7 +8,13 @@ from sqlalchemy.orm import Session
from .dto import DriverCreateDTO, DriverResponseDTO, DriverUpdateDTO
from .models import Driver
from .services import DriverService
from api.v1.modules.a76.transportation.transporters.services import TransporterService
from api.v1.modules.a76.transportation.transporters.models import Transporter
from api.v1.modules.a76.layouts_csv.drivers.routes import router as imports_router
from sqlalchemy import func
import logging
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/drivers")
@@ -63,6 +69,41 @@ async def create_driver(
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
# Validar acceso a la empresa del cuerpo
tenant_id = validate_access_to_resource(
db, driver_data.company_id, current_user
)
tk = (driver_data.transporter_key or "").strip()
# Buscar transportista: primero exacto, luego ignorando mayúsculas
transporter = TransporterService.get_by_id(
db, tk, tenant_id, driver_data.company_id
)
if not transporter:
transporter = TransporterService.get_by_id_ignore_case(
db, tk, tenant_id, driver_data.company_id
)
if not transporter:
# Diagnóstico: ¿existe ese transportista con otra empresa/tenant?
any_with_key = (
db.query(Transporter)
.filter(func.upper(Transporter.transporter_key) == tk.upper())
.limit(1)
.first()
)
logger.warning(
"Driver create: transporter not found. key=%r tenant_id=%s company_id=%s; "
"any_transporter_with_key=%s (other_tenant=%s other_company=%s)",
tk, tenant_id, driver_data.company_id,
getattr(any_with_key, "transporter_key", None) if any_with_key else None,
getattr(any_with_key, "tenant_id", None) if any_with_key else None,
getattr(any_with_key, "company_id", None) if any_with_key else None,
)
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"El transportista {tk} no existe en el catálogo de esta empresa. Crea primero el transportista o elige uno existente.",
)
# Usar la clave tal como está en BD (mismo caso)
driver_data.transporter_key = transporter.transporter_key
return DriverService.create_driver(db, driver_data)

29
backend/api/v1/modules/a76/transportation/transporters/services.py
View File

@@ -1,9 +1,13 @@
from typing import Optional, Tuple, List, Dict, Any
import logging
from sqlalchemy.orm import Session
from sqlalchemy import func
from . import dto, models
logger = logging.getLogger(__name__)
class TransporterService:
"""Service for Transporter CRUD operations with tenant support"""
@@ -47,11 +51,32 @@ class TransporterService:
def get_by_id(
db: Session, transporter_key: str, tenant_id: int, company_id: int
) -> Optional[models.Transporter]:
"""Get transporter by transporter_key"""
"""Get transporter by transporter_key (exact match after strip)."""
key = (transporter_key or "").strip()
if not key:
return None
return (
db.query(models.Transporter)
.filter(
models.Transporter.transporter_key == transporter_key,
models.Transporter.transporter_key == key,
models.Transporter.tenant_id == tenant_id,
models.Transporter.company_id == company_id,
)
.first()
)
@staticmethod
def get_by_id_ignore_case(
db: Session, transporter_key: str, tenant_id: int, company_id: int
) -> Optional[models.Transporter]:
"""Get transporter by transporter_key (case-insensitive) for given tenant/company."""
key = (transporter_key or "").strip()
if not key:
return None
return (
db.query(models.Transporter)
.filter(
func.upper(models.Transporter.transporter_key) == key.upper(),
models.Transporter.tenant_id == tenant_id,
models.Transporter.company_id == company_id,
)