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

feature/validaciones-clarion-vehiculos

Browse Source
This commit is contained in:
hreyes
2026-03-06 07:55:04 -07:00
parent cbfb2554cf
commit 6a902a414a
11 changed files with 532 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

148
backend/api/v1/modules/a76/layouts_csv/vehicles/common/common_validators.py
View File

@@ -1,11 +1,12 @@
"""
Validadores reutilizables para import CSV de vehículos (longitudes, decimal, fecha aseguradora).
Paridad Clarion: VALIDACIONES_TRANSPORTE, código entidad C/I/A/B, catálogos tipo/país/estado, desfase.
"""
import re
from decimal import Decimal
from typing import Dict, Any, Optional
from typing import Dict, Any, Optional, Set, Tuple
# Max lengths from Vehicle model (a76.vehicle)
# Max lengths from Vehicle model (a76.vehicle). Clarion Col A máx 15; modelo actual 14.
MAX_LEN = {
"vehicle_key": 14,
"ace_vehicle_key": 10,
@@ -102,3 +103,146 @@ def check_optional_insurance_date(row: Dict[str, Any], col: str, line_num: int)
if parse_insurance_date(val) is None:
return {"line": line_num, "col": col, "msg": "Formato de fecha inválido (use YYYYMMDD o DD/MM/YYYY)"}
return None
# --- Clarion VALIDACIONES_TRANSPORTE: dominio y catálogos ---
CODIGO_ENTIDAD_VALIDOS = {"C", "I", "A", "B"}
def check_codigo_entidad_valores(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Col F: Si CODIGO DE ENTIDAD no vacío, debe ser C, I, A o B (Clarion)."""
val = (row.get("CODIGO DE ENTIDAD") or "").strip().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. F) El Codigo de Entidad: {val} es incorrecto.",
"solution": "Capturar en columna F el Codigo de Entidad correcto, C, I, A ó B.",
}
def check_codigo_entidad_required(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""VALIDA_TODA: Col F (Codigo de Entidad) obligatorio cuando registro es nuevo."""
val = (row.get("CODIGO DE ENTIDAD") or "").strip()
if val:
return None
return {
"line": line_num,
"col": "CODIGO DE ENTIDAD",
"msg": "Existen campos vacios que son obligatorios, es la (Col.F) Codigo de Entidad.",
"solution": "Revisar la línea del archivo y capturar los campos con la información correcta.",
}
def check_transport_type_catalog(
row: Dict[str, Any],
line_num: int,
valid_transport_codes: Optional[Set[str]] = None,
) -> Optional[Dict[str, Any]]:
"""Col E: Si TIPO TRANSPORTE no vacío, debe existir en catálogo (GTipoTransportes)."""
val = (row.get("TIPO TRANSPORTE") or "").strip()
if not val or valid_transport_codes is None:
return None
if val.upper() in valid_transport_codes:
return None
return {
"line": line_num,
"col": "TIPO TRANSPORTE",
"msg": f"Error: (Col. E) El Tipo de Transporte: {val} es incorrecto.",
"solution": "Capturar en columna E algun Tipo de Transporte correcto.",
}
def check_pais_catalog_vehicles(
row: Dict[str, Any],
line_num: int,
valid_country_ame: Optional[Set[str]] = None,
) -> Optional[Dict[str, Any]]:
"""Col L: 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 L{line_num}) El Pais: {val} Es Incorrecto",
"solution": "Capturar en columna L 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. L) El Pais: {val} es incorrecto.",
"solution": "Capturar en columna L el Pais del Transporte en Clave Americana.",
}
def check_estado_catalog(
row: Dict[str, Any],
line_num: int,
state_descriptions_upper: Optional[Set[str]] = None,
) -> Optional[Dict[str, Any]]:
"""Col K: Si ESTADO no vacío, debe existir en catálogo (GEstados). Si hay estado, PAIS no puede estar vacío."""
val = (row.get("ESTADO") or "").strip()
if not val or state_descriptions_upper is None:
return None
val_upper = val.upper()
if val_upper not in state_descriptions_upper:
return {
"line": line_num,
"col": "ESTADO",
"msg": f"Error: (Celda K{line_num}) El Estado: {val} es incorrecto.",
"solution": "Capturar en columna K el Estado del Transporte en Clave Americana o Nombre Completo.",
}
# 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 K{line_num}) El Estado: {val} No esta ligado a ningun Pais.",
"solution": "Capturar en columna L un Pais en Clave Americana(US = Estados Unidos, MX = Mexico, ES = España, etc).",
}
return None
def check_estado_pais_consistency(
row: Dict[str, Any],
line_num: int,
state_country_set: Optional[Set[Tuple[str, 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
key = (pais, estado.upper())
if key in state_country_set:
return None
return {
"line": line_num,
"col": "ESTADO",
"msg": f"Error: (Col. K) EL Estado: {estado} no pertenece al Pais: {pais}.",
"solution": "Capturar en columna K un Estado que pertenesca al Pais de la columna L.",
}
def check_desfase_vehicles(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Si COL_EXTRA (columna 18 / Col R) tiene valor → advertencia de desfase (Clarion, no bloqueante por defecto)."""
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,
}

73
backend/api/v1/modules/a76/layouts_csv/vehicles/common/fk_loader.py Normal file
View File

@@ -0,0 +1,73 @@
"""
Carga de conjuntos FK para validación de import CSV de vehículos (transportes).
Clarion: GTipoTransportes, GPaises (Pais_Ame), GEstados (Descripcion / Clave_Ame), relación Estado-País.
"""
from typing import Set, Tuple, Optional
import logging
from core.database import CoreSessionLocal
logger = logging.getLogger(__name__)
def load_vehicles_fk_sets(
tenant_id: Optional[int] = None,
company_id: Optional[int] = None,
) -> Tuple[
Set[str],
Set[str],
Set[str],
Set[Tuple[str, str]],
]:
"""
Carga conjuntos para validación CSV de vehículos (paridad Clarion).
Devuelve:
- valid_transport_codes: códigos de transport_types (GTipoTransportes)
- valid_country_ame: claves americana de países (GPaises.Pais_Ame), mayúsculas
- state_descriptions_upper: descripciones de estados en mayúsculas (GEstados), para "estado existe"
- state_country_set: set de (ame_key_pais, description_estado_upper) para validar "estado pertenece a país"
"""
valid_transport_codes: Set[str] = set()
valid_country_ame: Set[str] = set()
state_descriptions_upper: Set[str] = set()
state_country_set: Set[Tuple[str, str]] = set()
try:
with CoreSessionLocal() as session:
from api.v1.modules.public.reference_data.transport_types.models import TransportType
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(TransportType.transport_code).all():
if row[0]:
valid_transport_codes.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())
# States: description (GEstados.Descripcion); State.m3_key = Country.m3_key
for state in session.query(State).all():
desc = (state.description or "").strip()
if desc:
state_descriptions_upper.add(desc.upper())
# País para este estado vía m3_key
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())
)
except Exception as e:
logger.warning("Vehicles import: could not load FK sets: %s", e)
return (
valid_transport_codes,
valid_country_ame,
state_descriptions_upper,
state_country_set,
)

72
backend/api/v1/modules/a76/layouts_csv/vehicles/common/mappers.py
View File

@@ -1,7 +1,8 @@
"""
Mapeo fila CSV → datos para Vehicle.
Paridad Clarion LLENA_TRANSPORTE: en actualización, campo vacío en CSV usa valor existente del vehículo.
"""
from typing import Dict, Any, Optional
from typing import Dict, Any, Optional, Union
from .common_validators import (
MAX_LEN,
@@ -21,6 +22,15 @@ 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 vehículo 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 row_to_vehicle_data(row_norm: Dict[str, Any]) -> Dict[str, Any]:
"""Build dict suitable for VehicleCreateDTO / VehicleUpdateDTO from normalized row."""
vehicle_key = _str_or_none(row_norm.get("CLAVE"), MAX_LEN["vehicle_key"])
@@ -45,3 +55,63 @@ def row_to_vehicle_data(row_norm: Dict[str, Any]) -> Dict[str, Any]:
"insurance_amount": parse_decimal(row_norm.get("MONTO ASEGURADO") or row_norm.get("MONTO")),
"insurance_date": parse_insurance_date(row_norm.get("FECHA DE ASEGURADORA") or row_norm.get("FECHA ASEGURADORA")),
}
def row_to_vehicle_data_for_update(
row_norm: Dict[str, Any],
existing_vehicle: Union[Any, Dict[str, Any]],
) -> Dict[str, Any]:
"""
Build dict for VehicleUpdateDTO: CSV value if non-empty, else existing vehicle value (Clarion VALIDA_PARCIAL / LLENA_TRANSPORTE).
"""
vehicle_key = _str_or_none(row_norm.get("CLAVE"), MAX_LEN["vehicle_key"]) or _get_existing_val(existing_vehicle, "vehicle_key")
if not vehicle_key:
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_vehicle, dto_key)
def _csv_decimal_or_existing(csv_key: str, dto_key: str):
raw = row_norm.get("MONTO ASEGURADO") or row_norm.get("MONTO") if csv_key == "MONTO ASEGURADO" else row_norm.get(csv_key)
if raw is not None and str(raw).strip():
d = parse_decimal(raw)
if d is not None:
return float(d)
val = _get_existing_val(existing_vehicle, dto_key)
if val is not None and hasattr(val, "__float__"):
try:
return float(val)
except (TypeError, ValueError):
pass
return val
def _csv_date_or_existing(csv_key: str, dto_key: str):
raw = row_norm.get("FECHA DE ASEGURADORA") or row_norm.get("FECHA ASEGURADORA")
if raw is not None and str(raw).strip():
d = parse_insurance_date(raw)
if d is not None:
return d
return _get_existing_val(existing_vehicle, dto_key)
return {
"vehicle_key": vehicle_key,
"ace_vehicle_key": _csv_or_existing("CLAVE ACE", "ace_vehicle_key", MAX_LEN["ace_vehicle_key"]),
"transporter_key": _csv_or_existing("CLAVE TRANSPORTE", "transporter_key", MAX_LEN["transporter_key"]),
"series": _csv_or_existing("VIN", "series", MAX_LEN["series"]),
"transport_type": _csv_or_existing("TIPO TRANSPORTE", "transport_type", MAX_LEN["transport_type"]),
"entity_code": _csv_or_existing("CODIGO DE ENTIDAD", "entity_code", MAX_LEN["entity_code"]),
"transponder_number": _csv_or_existing("TRANSPONDEDOR", "transponder_number", MAX_LEN["transponder_number"]),
"dot_number": _csv_or_existing("NUMERO DOT", "dot_number", MAX_LEN["dot_number"]),
"plate_number": _csv_or_existing("PLACAS", "plate_number", MAX_LEN["plate_number"]),
"city": _csv_or_existing("CIUDAD", "city", MAX_LEN["city"]),
"state": _csv_or_existing("ESTADO", "state", MAX_LEN["state"]),
"country": _csv_or_existing("PAIS", "country", MAX_LEN["country"]),
"seal": _csv_or_existing("PRECINTO", "seal", MAX_LEN["seal"]),
"insurance_company_name": _csv_or_existing("EMPRESA ASEGURADORA", "insurance_company_name", MAX_LEN["insurance_company_name"]),
"insurance_number": _csv_or_existing("NUM. ASEGURADORA", "insurance_number", MAX_LEN["insurance_number"]),
"insurance_amount": _csv_decimal_or_existing("MONTO ASEGURADO", "insurance_amount"),
"insurance_date": _csv_date_or_existing("FECHA DE ASEGURADORA", "insurance_date"),
}

2
backend/api/v1/modules/a76/layouts_csv/vehicles/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": "vehicles",
"actualizar": actualizar,
}
try:

102
backend/api/v1/modules/a76/layouts_csv/vehicles/tasks.py
View File

@@ -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
@@ -17,8 +17,9 @@ 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 .validators import validate_row_vehicle
from .common.mappers import row_to_vehicle_data
from .validators import validate_row_vehicle, validate_row_vehicle_desfase
from .common.mappers import row_to_vehicle_data, row_to_vehicle_data_for_update
from .common.fk_loader import load_vehicles_fk_sets
logger = logging.getLogger(__name__)
@@ -56,6 +57,33 @@ 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_vehicle_keys: Set[str] = set()
if actualizar:
try:
from api.v1.modules.a76.transportation.vehicles.models import Vehicle
with CoreSessionLocal() as session:
for v in (
session.query(Vehicle.vehicle_key)
.filter(
Vehicle.tenant_id == tenant_id,
Vehicle.company_id == company_id,
)
.all()
):
if v[0] and (v[0] or "").strip():
existing_vehicle_keys.add((v[0] or "").strip())
except Exception as e:
logger.warning("Vehicles import: could not load existing vehicle_keys for actualizar: %s", e)
(
valid_transport_codes,
valid_country_ame,
state_descriptions_upper,
state_country_set,
) = load_vehicles_fk_sets(tenant_id, company_id)
error_count = 0
processed_rows = 0
errors_detail: List[Dict[str, Any]] = []
@@ -68,7 +96,18 @@ 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_vehicle(row_norm, i)
# Desfase: advertencia no bloqueante (no se añade a error_lines)
_ = validate_row_vehicle_desfase(row_norm, i)
err = validate_row_vehicle(
row_norm,
i,
actualizar=actualizar,
existing_vehicle_keys=existing_vehicle_keys,
valid_transport_codes=valid_transport_codes,
valid_country_ame=valid_country_ame,
state_descriptions_upper=state_descriptions_upper,
state_country_set=state_country_set,
)
if err:
error_count += 1
error_lines_list.append(err["line"])
@@ -142,6 +181,33 @@ 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_vehicle_keys: Set[str] = set()
if actualizar:
try:
from api.v1.modules.a76.transportation.vehicles.models import Vehicle
with CoreSessionLocal() as session:
for v in (
session.query(Vehicle.vehicle_key)
.filter(
Vehicle.tenant_id == tenant_id,
Vehicle.company_id == company_id,
)
.all()
):
if v[0] and (v[0] or "").strip():
existing_vehicle_keys.add((v[0] or "").strip())
except Exception as e:
logger.warning("Vehicles import: could not load existing vehicle_keys for actualizar: %s", e)
(
valid_transport_codes,
valid_country_ame,
state_descriptions_upper,
state_country_set,
) = load_vehicles_fk_sets(tenant_id, company_id)
from api.v1.modules.a76.transportation.vehicles.services import VehicleService
from api.v1.modules.a76.transportation.vehicles.dto import VehicleCreateDTO, VehicleUpdateDTO
@@ -160,7 +226,16 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
continue
row_norm = row_from_template(row, common_normalize.normalize_header)
err = validate_row_vehicle(row_norm, i)
err = validate_row_vehicle(
row_norm,
i,
actualizar=actualizar,
existing_vehicle_keys=existing_vehicle_keys,
valid_transport_codes=valid_transport_codes,
valid_country_ame=valid_country_ame,
state_descriptions_upper=state_descriptions_upper,
state_country_set=state_country_set,
)
if err:
skipped_invalid += 1
vk = (row_norm.get("CLAVE") or "").strip()[:14] or "-"
@@ -172,12 +247,10 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
})
continue
data = row_to_vehicle_data(row_norm)
if not data or not data.get("vehicle_key"):
vk = (row_norm.get("CLAVE") or "").strip()[:14] or ""
if not vk:
skipped_invalid += 1
continue
vk = data["vehicle_key"]
if vk in seen_keys_in_file:
skipped_duplicate += 1
skipped_details.append({
@@ -192,10 +265,21 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
existing = VehicleService.get_by_id(session, vk, tenant_id, company_id)
try:
if existing:
if actualizar:
data = row_to_vehicle_data_for_update(row_norm, existing)
else:
data = row_to_vehicle_data(row_norm)
if not data or not data.get("vehicle_key"):
skipped_invalid += 1
continue
update_data = VehicleUpdateDTO(**{k: v for k, v in data.items() if k != "vehicle_key"})
VehicleService.update(session, vk, tenant_id, update_data, company_id)
updated_count += 1
else:
data = row_to_vehicle_data(row_norm)
if not data or not data.get("vehicle_key"):
skipped_invalid += 1
continue
create_data = VehicleCreateDTO(**data)
VehicleService.create(session, create_data, tenant_id, company_id)
inserted_count += 1

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

@@ -24,6 +24,7 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
{"canonical": "NUM. ASEGURADORA", "aliases": ["NUM ASEGURADORA", "POLIZA", "INSURANCE NUMBER"]},
{"canonical": "MONTO ASEGURADO", "aliases": ["MONTO", "INSURANCE AMOUNT"]},
{"canonical": "FECHA DE ASEGURADORA", "aliases": ["FECHA ASEGURADORA", "INSURANCE DATE"]},
{"canonical": "COL_EXTRA", "aliases": ["COLUMNA R", "COL R"]},
],
}

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

@@ -1,3 +1,3 @@
from .create import validate_row_vehicle
from .create import validate_row_vehicle, validate_row_vehicle_desfase
__all__ = ["validate_row_vehicle"]
__all__ = ["validate_row_vehicle", "validate_row_vehicle_desfase"]

90
backend/api/v1/modules/a76/layouts_csv/vehicles/validators/common.py
View File

@@ -1,12 +1,19 @@
"""
Validaciones comunes de fila para import CSV de vehículos.
Paridad Clarion: VALIDACIONES_TRANSPORTE, VALIDA_TODA_TRANSPORTE, VALIDA_PARCIAL_TRANSPORTE.
"""
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_codigo_entidad_required,
check_transport_type_catalog,
check_pais_catalog_vehicles,
check_estado_catalog,
check_estado_pais_consistency,
)
@@ -56,3 +63,84 @@ def validate_row_vehicle_insurance_date(row: Dict[str, Any], line_num: int) -> O
if parse_insurance_date(fecha) is None:
return {"line": line_num, "col": "FECHA DE ASEGURADORA", "msg": "Formato de fecha inválido (use YYYYMMDD o DD/MM/YYYY)"}
return None
def validaciones_transporte(
row: Dict[str, Any],
line_num: int,
valid_transport_codes: 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,
) -> Optional[Dict[str, Any]]:
"""
VALIDACIONES_TRANSPORTE: reglas compartidas (longitudes, tipo transporte, código entidad dominio,
país, estado, estado-país). No exige CODIGO DE ENTIDAD obligatorio (eso es solo VALIDA_TODA).
"""
err = validate_row_vehicle_required(row, line_num)
if err:
return err
err = validate_row_vehicle_lengths(row, line_num)
if err:
return err
err = validate_row_vehicle_amount(row, line_num)
if err:
return err
err = validate_row_vehicle_insurance_date(row, line_num)
if err:
return err
err = check_codigo_entidad_valores(row, line_num)
if err:
return err
err = check_transport_type_catalog(row, line_num, valid_transport_codes)
if err:
return err
err = check_pais_catalog_vehicles(row, line_num, valid_country_ame)
if err:
return err
err = check_estado_catalog(row, line_num, state_descriptions_upper)
if err:
return err
err = check_estado_pais_consistency(row, line_num, state_country_set)
if err:
return err
return None
def valida_toda_transporte(
row: Dict[str, Any],
line_num: int,
valid_transport_codes: 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,
) -> Optional[Dict[str, Any]]:
"""VALIDA_TODA_TRANSPORTE: CODIGO DE ENTIDAD obligatorio + VALIDACIONES_TRANSPORTE (registro nuevo)."""
err = check_codigo_entidad_required(row, line_num)
if err:
return err
return validaciones_transporte(
row, line_num,
valid_transport_codes=valid_transport_codes,
valid_country_ame=valid_country_ame,
state_descriptions_upper=state_descriptions_upper,
state_country_set=state_country_set,
)
def valida_parcial_transporte(
row: Dict[str, Any],
line_num: int,
valid_transport_codes: 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,
) -> Optional[Dict[str, Any]]:
"""VALIDA_PARCIAL_TRANSPORTE: solo VALIDACIONES_TRANSPORTE (actualizar registro existente)."""
return validaciones_transporte(
row, line_num,
valid_transport_codes=valid_transport_codes,
valid_country_ame=valid_country_ame,
state_descriptions_upper=state_descriptions_upper,
state_country_set=state_country_set,
)

64
backend/api/v1/modules/a76/layouts_csv/vehicles/validators/create.py
View File

@@ -1,31 +1,63 @@
"""
Punto de entrada de validación para import de una fila vehículo.
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_vehicle_required,
validate_row_vehicle_lengths,
validate_row_vehicle_amount,
validate_row_vehicle_insurance_date,
valida_toda_transporte,
valida_parcial_transporte,
)
from ..common.common_validators import check_desfase_vehicles
def validate_row_vehicle(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
def validate_row_vehicle(
row: Dict[str, Any],
line_num: int,
actualizar: bool = False,
existing_vehicle_keys: Optional[Set[str]] = None,
valid_transport_codes: 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,
) -> Optional[Dict[str, Any]]:
"""
Valida una fila de CSV de vehículos.
Encadena: requerido (CLAVE) → longitudes → monto opcional → fecha aseguradora opcional.
1. CLAVE vacía → error.
2. Si actualizar y CLAVE en existing_vehicle_keys → valida_parcial_transporte (sin exigir CODIGO DE ENTIDAD).
3. Si no actualizar o CLAVE no existe → valida_toda_transporte (CODIGO DE ENTIDAD obligatorio).
Desfase (COL_EXTRA) no se valida aquí; el caller puede llamar validate_row_vehicle_desfase para advertencias no bloqueantes.
"""
err = validate_row_vehicle_required(row, line_num)
if err:
return err
err = validate_row_vehicle_lengths(row, line_num)
if err:
return err
err = validate_row_vehicle_amount(row, line_num)
if err:
return err
err = validate_row_vehicle_insurance_date(row, line_num)
if err:
return err
return None
existing = existing_vehicle_keys or set()
clave = (row.get("CLAVE") or "").strip()
use_partial = actualizar and bool(clave and clave in existing)
if use_partial:
err = valida_parcial_transporte(
row, line_num,
valid_transport_codes=valid_transport_codes,
valid_country_ame=valid_country_ame,
state_descriptions_upper=state_descriptions_upper,
state_country_set=state_country_set,
)
else:
err = valida_toda_transporte(
row, line_num,
valid_transport_codes=valid_transport_codes,
valid_country_ame=valid_country_ame,
state_descriptions_upper=state_descriptions_upper,
state_country_set=state_country_set,
)
return err
def validate_row_vehicle_desfase(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""
Advertencia de desfase si COL_EXTRA (Col R) tiene valor. No bloqueante; el caller puede acumular en warnings.
"""
return check_desfase_vehicles(row, line_num)

6
frontend/src/lib/api.ts
View File

@@ -509,11 +509,13 @@ export const api = {
// CSV import for Vehículos / Transportes (transportation/vehicles/imports)
vehicleImports: {
upload: (file: File, companyId: number) => {
upload: (file: File, companyId: number, options?: { actualizar?: boolean }) => {
const formData = new FormData();
formData.append('file', file);
const params = new URLSearchParams({ company_id: String(companyId) });
if (options?.actualizar !== undefined) params.set('actualizar', String(options.actualizar));
return fetchApi(
`/v1/a76/transportation/vehicles/imports/upload?company_id=${companyId}`,
`/v1/a76/transportation/vehicles/imports/upload?${params.toString()}`,
{ method: 'POST', body: formData }
);
},

4
frontend/src/routes/dashboard/csv-upload/+page.svelte
View File

@@ -272,7 +272,9 @@
if (useVehicleImport) {
try {
const res = await api.vehicleImports.upload(file, companyId);
const catalogosSettings = allSettings['catalogos'] || {};
const actualizar = catalogosSettings['mode'] === 'update';
const res = await api.vehicleImports.upload(file, companyId, { actualizar });
if (res.data?.job_id) {
currentJobId = res.data.job_id;
pollStatus();