feature/tabs-partidas-jerarquia-alt-n

backend/api/v1/common/catalog_validation_errors.py

@@ -0,0 +1,12 @@
+"""Errores de validación alineados a reglas CSV / catálogos (HTTP 422)."""
+
+from typing import Any, Dict, List
+
+
+class CatalogValidationError(Exception):
+    """Lista de errores tipo {line, col, msg} como en import CSV."""
+
+    def __init__(self, errors: List[Dict[str, Any]]):
+        self.errors = errors or []
+        first = self.errors[0].get("msg", "Validación de catálogo") if self.errors else "Validación de catálogo"
+        super().__init__(first)

backend/api/v1/common/tenant_crud_routes.py

@@ -5,6 +5,8 @@ import inspect
 from core.database import get_core_db
 from core.security import get_current_user, validate_access_to_resource
 from fastapi import APIRouter, Body, Depends, HTTPException, Path, Query, Request
+
+from api.v1.common.catalog_validation_errors import CatalogValidationError
 from pydantic import BaseModel
 from sqlalchemy.orm import Session
 
@@ -376,6 +378,14 @@ class TenantCRUDRoutes(
                 try:
                     resource = self.service.create(db, data, tenant_id, company_id)
                     return resource
+                except CatalogValidationError as e:
+                    raise HTTPException(
+                        status_code=422,
+                        detail={
+                            "message": str(e),
+                            "errors": e.errors,
+                        },
+                    )
                 except ValueError as e:
                     # Capturar errores de validación (como duplicados)
                     raise HTTPException(status_code=400, detail=str(e))
@@ -412,6 +422,14 @@ class TenantCRUDRoutes(
                 try:
                     resource = self.service.create(db, data, tenant_id, company_id)
                     return resource
+                except CatalogValidationError as e:
+                    raise HTTPException(
+                        status_code=422,
+                        detail={
+                            "message": str(e),
+                            "errors": e.errors,
+                        },
+                    )
                 except ValueError as e:
                     # Capturar errores de validación (como duplicados)
                     raise HTTPException(status_code=400, detail=str(e))
@@ -454,6 +472,14 @@ class TenantCRUDRoutes(
                     resource = self.service.update(
                         db, parent_id, tenant_id, data, company_id
                     )
+                except CatalogValidationError as e:
+                    raise HTTPException(
+                        status_code=422,
+                        detail={
+                            "message": str(e),
+                            "errors": e.errors,
+                        },
+                    )
                 except ValueError as e:
                     # Capturar errores de validación (como duplicados)
                     raise HTTPException(status_code=400, detail=str(e))
@@ -498,6 +524,14 @@ class TenantCRUDRoutes(
                     resource = self.service.update(
                         db, resource_id, tenant_id, data, company_id
                     )
+                except CatalogValidationError as e:
+                    raise HTTPException(
+                        status_code=422,
+                        detail={
+                            "message": str(e),
+                            "errors": e.errors,
+                        },
+                    )
                 except ValueError as e:
                     # Capturar errores de validación (como duplicados)
                     raise HTTPException(status_code=400, detail=str(e))

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

@@ -7,7 +7,8 @@ from typing import Dict, Any, Optional, Set
 
 
 MAX_LEN = {
-    "transporter_key": 5,
+    # Alineado a a76.transporter.transporter_key (23); CSV Clarion histórico usaba claves cortas
+    "transporter_key": 23,
     "driver_name": 80,
     "license_number": 29,
     "express_line_id": 17,

backend/api/v1/modules/a76/transportation/catalog_parity.py

@@ -0,0 +1,379 @@
+"""
+Validación de paridad con import CSV (mismas reglas y fk_loader) para CRUD Transportes.
+"""
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional, Set, Tuple
+
+from sqlalchemy.orm import Session
+
+from api.v1.common.catalog_validation_errors import CatalogValidationError
+
+LINE = 1
+
+
+def _raise_if_errors(errors: List[Dict[str, Any]]) -> None:
+    if errors:
+        raise CatalogValidationError(errors)
+
+
+# --- Trailers ---
+
+
+def trailer_fields_to_csv_row(d: Dict[str, Any]) -> Dict[str, Any]:
+    return {
+        "NUMERO TRAILER": (d.get("trailer_number") or "").strip(),
+        "CLAVE ACE": (d.get("ace_trailer_number") or "").strip() if d.get("ace_trailer_number") is not None else "",
+        "TIPO TRAILER": (d.get("trailer_type_key") or "").strip() if d.get("trailer_type_key") is not None else "",
+        "PRECINTO": (d.get("seal") or "").strip() if d.get("seal") is not None else "",
+        "CODIGO DE ENTIDAD": (d.get("entity_code") or "").strip() if d.get("entity_code") is not None else "",
+        "PLACAS": (d.get("plate_number") or "").strip() if d.get("plate_number") is not None else "",
+        "ESTADO": (d.get("state") or "").strip() if d.get("state") is not None else "",
+        "PAIS": (d.get("country") or "").strip() if d.get("country") is not None else "",
+        "CLAVE CONTENEDOR": (d.get("container_key") or "").strip() if d.get("container_key") is not None else "",
+    }
+
+
+def trailer_model_to_row(tr) -> Dict[str, Any]:
+    return trailer_fields_to_csv_row(
+        {
+            "trailer_number": tr.trailer_number,
+            "ace_trailer_number": tr.ace_trailer_number,
+            "trailer_type_key": tr.trailer_type_key,
+            "seal": tr.seal,
+            "entity_code": tr.entity_code,
+            "plate_number": tr.plate_number,
+            "state": tr.state,
+            "country": tr.country,
+            "container_key": tr.container_key,
+        }
+    )
+
+
+def validate_trailer_row_for_api(
+    tenant_id: int,
+    company_id: int,
+    row: Dict[str, Any],
+    *,
+    is_update: bool,
+    existing_trailer_numbers: Set[str],
+) -> None:
+    from api.v1.modules.a76.layouts_csv.trailers.common.fk_loader import load_trailers_fk_sets
+    from api.v1.modules.a76.layouts_csv.trailers.validators.create import validate_row_trailer
+
+    (
+        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)
+
+    clave = (row.get("NUMERO TRAILER") or "").strip()
+    existing_norm = {x.strip() for x in existing_trailer_numbers if x}
+    actualizar = is_update and bool(clave and clave in existing_norm)
+
+    errs = validate_row_trailer(
+        row,
+        LINE,
+        actualizar=actualizar,
+        existing_trailer_numbers=existing_norm,
+        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,
+    )
+    _raise_if_errors(errs)
+
+
+# --- Vehicles ---
+
+
+def _fmt_insurance_date(val: Any) -> str:
+    if val is None:
+        return ""
+    if isinstance(val, int):
+        return str(val)
+    return str(val).strip()
+
+
+def _fmt_monto(val: Any) -> str:
+    if val is None:
+        return ""
+    if isinstance(val, float):
+        return str(val)
+    return str(val).strip()
+
+
+def vehicle_fields_to_csv_row(d: Dict[str, Any]) -> Dict[str, Any]:
+    return {
+        "CLAVE": (d.get("vehicle_key") or "").strip(),
+        "CLAVE ACE": (d.get("ace_vehicle_key") or "").strip() if d.get("ace_vehicle_key") is not None else "",
+        "CLAVE TRANSPORTE": (d.get("transporter_key") or "").strip() if d.get("transporter_key") is not None else "",
+        "VIN": (d.get("series") or "").strip() if d.get("series") is not None else "",
+        "TIPO TRANSPORTE": (d.get("transport_type") or "").strip() if d.get("transport_type") is not None else "",
+        "CODIGO DE ENTIDAD": (d.get("entity_code") or "").strip() if d.get("entity_code") is not None else "",
+        "TRANSPONDEDOR": (d.get("transponder_number") or "").strip() if d.get("transponder_number") is not None else "",
+        "NUMERO DOT": (d.get("dot_number") or "").strip() if d.get("dot_number") is not None else "",
+        "PLACAS": (d.get("plate_number") or "").strip() if d.get("plate_number") is not None else "",
+        "CIUDAD": (d.get("city") or "").strip() if d.get("city") is not None else "",
+        "ESTADO": (d.get("state") or "").strip() if d.get("state") is not None else "",
+        "PAIS": (d.get("country") or "").strip() if d.get("country") is not None else "",
+        "PRECINTO": (d.get("seal") or "").strip() if d.get("seal") is not None else "",
+        "EMPRESA ASEGURADORA": (d.get("insurance_company_name") or "").strip()
+        if d.get("insurance_company_name") is not None
+        else "",
+        "NUM. ASEGURADORA": (d.get("insurance_number") or "").strip() if d.get("insurance_number") is not None else "",
+        "MONTO ASEGURADO": _fmt_monto(d.get("insurance_amount")),
+        "FECHA DE ASEGURADORA": _fmt_insurance_date(d.get("insurance_date")),
+    }
+
+
+def vehicle_model_to_row(v) -> Dict[str, Any]:
+    return vehicle_fields_to_csv_row(
+        {
+            "vehicle_key": v.vehicle_key,
+            "ace_vehicle_key": v.ace_vehicle_key,
+            "transporter_key": v.transporter_key,
+            "series": v.series,
+            "transport_type": v.transport_type,
+            "entity_code": v.entity_code,
+            "transponder_number": v.transponder_number,
+            "dot_number": v.dot_number,
+            "plate_number": v.plate_number,
+            "city": v.city,
+            "state": v.state,
+            "country": v.country,
+            "seal": v.seal,
+            "insurance_company_name": v.insurance_company_name,
+            "insurance_number": v.insurance_number,
+            "insurance_amount": float(v.insurance_amount) if v.insurance_amount is not None else None,
+            "insurance_date": v.insurance_date,
+        }
+    )
+
+
+def validate_vehicle_transporter_key(
+    db: Session,
+    tenant_id: int,
+    company_id: int,
+    transporter_key: Optional[str],
+) -> None:
+    if transporter_key is None or not str(transporter_key).strip():
+        return
+    from api.v1.modules.a76.transportation.transporters.services import TransporterService
+
+    t = TransporterService.get_by_id_ignore_case(
+        db, str(transporter_key).strip(), tenant_id, company_id
+    )
+    if not t:
+        _raise_if_errors(
+            [
+                {
+                    "line": LINE,
+                    "col": "CLAVE TRANSPORTE",
+                    "msg": f"El transportista '{transporter_key}' no existe en el catálogo de esta empresa.",
+                }
+            ]
+        )
+
+
+def validate_vehicle_row_for_api(
+    db: Session,
+    tenant_id: int,
+    company_id: int,
+    row: Dict[str, Any],
+    *,
+    is_update: bool,
+    existing_vehicle_keys: Set[str],
+) -> None:
+    from api.v1.modules.a76.layouts_csv.vehicles.common.fk_loader import load_vehicles_fk_sets
+    from api.v1.modules.a76.layouts_csv.vehicles.validators.create import validate_row_vehicle
+
+    tk = (row.get("CLAVE TRANSPORTE") or "").strip()
+    if tk:
+        validate_vehicle_transporter_key(db, tenant_id, company_id, tk)
+
+    (
+        valid_transport_codes,
+        valid_country_ame,
+        state_descriptions_upper,
+        state_country_set,
+    ) = load_vehicles_fk_sets(tenant_id, company_id)
+
+    clave = (row.get("CLAVE") or "").strip()
+    existing_norm = {x.strip() for x in existing_vehicle_keys if x}
+    actualizar = is_update and bool(clave and clave in existing_norm)
+
+    errs = validate_row_vehicle(
+        row,
+        LINE,
+        actualizar=actualizar,
+        existing_vehicle_keys=existing_norm,
+        valid_transport_codes=valid_transport_codes,
+        valid_country_ame=valid_country_ame,
+        state_descriptions_upper=state_descriptions_upper,
+        state_country_set=state_country_set,
+    )
+    _raise_if_errors(errs)
+
+
+# --- Transporters ---
+
+
+def transporter_fields_to_csv_row(d: Dict[str, Any]) -> Dict[str, Any]:
+    def s(k: str) -> str:
+        v = d.get(k)
+        if v is None:
+            return ""
+        return str(v).strip()
+
+    return {
+        "CLAVE TRANSPORTISTA": s("transporter_key"),
+        "NOMBRE": s("name"),
+        "NOMBRE CORTO": s("short_name"),
+        "RESPONSABLE": s("responsible"),
+        "RFC": s("rfc"),
+        "CALLES": s("streets"),
+        "CODIGO POSTAL": s("postal_code"),
+        "CIUDAD": s("city"),
+        "ESTADO": s("state"),
+        "PAIS": s("country"),
+        "CODIGO CARGADOR": s("loader_code"),
+        "CODIGO CAAT": s("caat_code"),
+        "CODIGO TRANS": s("transport_code"),
+        "TIPO INTERFASE TRANS": s("transport_interface_type"),
+        "SERVIDOR FTP": s("ftp_server"),
+        "USUARIO FTP": s("ftp_user"),
+        "CLAVE ACCESO FTP": s("ftp_password"),
+        "DIRECTORIO FTP": s("ftp_directory"),
+    }
+
+
+def transporter_model_to_row(t) -> Dict[str, Any]:
+    return transporter_fields_to_csv_row(
+        {
+            "transporter_key": t.transporter_key,
+            "name": t.name,
+            "short_name": t.short_name,
+            "responsible": t.responsible,
+            "rfc": t.rfc,
+            "streets": t.streets,
+            "postal_code": t.postal_code,
+            "city": t.city,
+            "state": t.state,
+            "country": t.country,
+            "loader_code": t.loader_code,
+            "caat_code": t.caat_code,
+            "transport_code": t.transport_code,
+            "transport_interface_type": t.transport_interface_type,
+            "ftp_server": t.ftp_server,
+            "ftp_user": t.ftp_user,
+            "ftp_password": t.ftp_password,
+            "ftp_directory": t.ftp_directory,
+        }
+    )
+
+
+def validate_transporter_row_for_api(
+    tenant_id: int,
+    company_id: int,
+    row: Dict[str, Any],
+    *,
+    is_update: bool,
+    existing_transporter_keys: Set[str],
+) -> None:
+    from api.v1.modules.a76.layouts_csv.transportistas.common.fk_loader import load_transportistas_fk_sets
+    from api.v1.modules.a76.layouts_csv.transportistas.validators.create import validate_row_transporter
+
+    (
+        existing_keys_loaded,
+        valid_country_ame,
+        state_descriptions_upper,
+        state_country_set,
+    ) = load_transportistas_fk_sets(tenant_id, company_id)
+
+    clave = (row.get("CLAVE TRANSPORTISTA") or "").strip().upper()
+    # existing set from loader is uppercased keys for this company
+    existing_norm = existing_keys_loaded | {x.strip().upper() for x in existing_transporter_keys if x}
+    actualizar = is_update and bool(clave and clave in existing_norm)
+
+    errs = validate_row_transporter(
+        row,
+        LINE,
+        actualizar=actualizar,
+        existing_transporter_keys=existing_norm,
+        valid_country_ame=valid_country_ame,
+        state_descriptions_upper=state_descriptions_upper,
+        state_country_set=state_country_set,
+    )
+    _raise_if_errors(errs)
+
+
+# --- Drivers ---
+
+
+def driver_fields_to_csv_row(d: Dict[str, Any]) -> Dict[str, Any]:
+    def s(k: str) -> str:
+        v = d.get(k)
+        if v is None:
+            return ""
+        return str(v).strip()
+
+    line = d.get("line")
+    line_s = str(line) if line is not None else ""
+
+    return {
+        "TRANSPORTISTA": s("transporter_key"),
+        "LINEA": line_s,
+        "CLAVE CONDUCTOR": s("driver_name"),
+        "LICENCIA": s("license_number"),
+        "PERMISO LINEA EXPRESS": s("express_line_id"),
+        "IDENTIFICACION ACE": s("ace_id"),
+        "FECHA NACIMIENTO": str(d.get("birth_date")) if d.get("birth_date") is not None else "",
+        "SEXO": s("gender"),
+        "PAIS NACIMIENTO": s("birth_country"),
+        "TRANSPORTA MAT. PELIGROSO?": s("hazardous_material_auth"),
+        "PERMISO MAT. PELIGROSO": s("hazardous_material_state"),
+        "NOMBRE(S)": s("first_name"),
+        "APELLIDO PATERNO": s("last_name"),
+        "FORMA IDENTIFICACION 1": s("id_key1"),
+        "NUM. IDENTIFICACION 1": s("id_number1"),
+        "ESTADO": s("id_state1"),
+        "PAIS": s("id_country1"),
+        "FORMA IDENTIFICACION 2": s("id_key2"),
+        "NUM. IDENTIFICACION 2": s("id_number2"),
+        "ESTADO 2": s("id_state2"),
+        "PAIS 2": s("id_country2"),
+    }
+
+
+def validate_driver_row_for_api(
+    tenant_id: int,
+    company_id: int,
+    row: Dict[str, Any],
+    *,
+    is_update: bool,
+    existing_driver_keys: Set[Tuple[str, int]],
+) -> None:
+    from api.v1.modules.a76.layouts_csv.drivers.common.fk_loader import load_drivers_fk_sets
+    from api.v1.modules.a76.layouts_csv.drivers.validators.create import validate_row_driver
+
+    valid_transporter_keys, valid_country_ame, _ = load_drivers_fk_sets(tenant_id, company_id)
+
+    transporter_key = (row.get("TRANSPORTISTA") or "").strip().upper()
+    from api.v1.modules.a76.layouts_csv.drivers.common.common_validators import parse_int
+
+    line = parse_int(row.get("LINEA"))
+    actualizar = is_update and bool(transporter_key and line is not None) and (transporter_key, line) in existing_driver_keys
+
+    errs = validate_row_driver(
+        row,
+        LINE,
+        actualizar=actualizar,
+        existing_driver_keys=existing_driver_keys,
+        valid_transporter_keys=valid_transporter_keys,
+        valid_country_ame=valid_country_ame,
+    )
+    _raise_if_errors(errs)

backend/api/v1/modules/a76/transportation/drivers/routes.py

@@ -3,6 +3,8 @@ from typing import Any, Dict, List
 from core.database import get_core_db
 from core.security import get_current_user, validate_access_to_resource
 from fastapi import APIRouter, Depends, HTTPException, Query, status
+
+from api.v1.common.catalog_validation_errors import CatalogValidationError
 from sqlalchemy.orm import Session
 
 from .dto import DriverCreateDTO, DriverResponseDTO, DriverUpdateDTO
@@ -104,7 +106,13 @@ async def create_driver(
         )
     # Usar la clave tal como está en BD (mismo caso)
     driver_data.transporter_key = transporter.transporter_key
-    return DriverService.create_driver(db, driver_data)
+    try:
+        return DriverService.create_driver(db, driver_data)
+    except CatalogValidationError as e:
+        raise HTTPException(
+            status_code=422,
+            detail={"message": str(e), "errors": e.errors},
+        )
 
 
 @router.put("/{transporter_key}/{line}", response_model=DriverResponseDTO)
@@ -117,14 +125,20 @@ async def update_driver(
     current_user: dict = Depends(get_current_user),
 ):
     tenant_id = validate_access_to_resource(db, company_id, current_user)
-    driver = DriverService.update_driver(
-        db,
-        transporter_key,
-        line,
-        str(company_id),
-        tenant_id,
-        driver_data,
-    )
+    try:
+        driver = DriverService.update_driver(
+            db,
+            transporter_key,
+            line,
+            str(company_id),
+            tenant_id,
+            driver_data,
+        )
+    except CatalogValidationError as e:
+        raise HTTPException(
+            status_code=422,
+            detail={"message": str(e), "errors": e.errors},
+        )
     if not driver:
         raise HTTPException(status_code=404, detail="Driver not found")
     return driver

backend/api/v1/modules/a76/transportation/drivers/services.py

@@ -4,6 +4,10 @@ from sqlalchemy import text
 from sqlalchemy.orm import Session
 
 from . import dto, models
+from api.v1.modules.a76.transportation.catalog_parity import (
+    driver_fields_to_csv_row,
+    validate_driver_row_for_api,
+)
 
 DRIVER_ID_SEQ = "a76.driver_driver_id_seq"
 
@@ -43,6 +47,13 @@ class DriverService:
     @staticmethod
     def create_driver(db: Session, driver_data: dto.DriverCreateDTO):
         data = driver_data.model_dump()
+        validate_driver_row_for_api(
+            int(driver_data.tenant_id),
+            int(driver_data.company_id),
+            driver_fields_to_csv_row(data),
+            is_update=False,
+            existing_driver_keys=set(),
+        )
         if data.get("driver_id") is None:
             data["driver_id"] = allocate_driver_id(db)
         new_driver = models.Driver(**data)
@@ -66,6 +77,42 @@ class DriverService:
         if not driver:
             return None
         update_data = data.model_dump(exclude_unset=True)
+        merged = {
+            "transporter_key": driver.transporter_key,
+            "line": driver.line,
+            "driver_name": driver.driver_name,
+            "license_number": driver.license_number,
+            "express_line_id": driver.express_line_id,
+            "ace_id": driver.ace_id,
+            "birth_date": driver.birth_date,
+            "gender": driver.gender,
+            "birth_country": driver.birth_country,
+            "hazardous_material_auth": driver.hazardous_material_auth,
+            "hazardous_material_state": driver.hazardous_material_state,
+            "first_name": driver.first_name,
+            "last_name": driver.last_name,
+            "id_key1": driver.id_key1,
+            "id_number1": driver.id_number1,
+            "id_state1": driver.id_state1,
+            "id_country1": driver.id_country1,
+            "id_key2": driver.id_key2,
+            "id_number2": driver.id_number2,
+            "id_state2": driver.id_state2,
+            "id_country2": driver.id_country2,
+            "badge_number": driver.badge_number,
+            "class_type": driver.class_type,
+            "unique_badge_number": driver.unique_badge_number,
+        }
+        merged.update(update_data)
+        tid = int(driver.tenant_id)
+        cid = int(driver.company_id)
+        validate_driver_row_for_api(
+            tid,
+            cid,
+            driver_fields_to_csv_row(merged),
+            is_update=True,
+            existing_driver_keys={(driver.transporter_key.strip().upper(), driver.line)},
+        )
         for key, value in update_data.items():
             setattr(driver, key, value)
         db.commit()

backend/api/v1/modules/a76/transportation/trailers/dto.py

@@ -1,6 +1,6 @@
 from typing import Optional
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator
 
 
 class TrailerBaseDTO(BaseModel):
@@ -15,6 +15,44 @@ class TrailerBaseDTO(BaseModel):
     country: Optional[str] = None
     container_key: Optional[str] = None
 
+    @field_validator("trailer_number", mode="before")
+    @classmethod
+    def strip_trailer_number(cls, v: object) -> object:
+        if isinstance(v, str):
+            return v.strip()
+        return v
+
+    @field_validator("trailer_type_key", mode="before")
+    @classmethod
+    def trailer_type_key_optional_fk(cls, v: object) -> Optional[str]:
+        """Empty string from JSON must become NULL for FK; normalize case for catalog match."""
+        if v is None:
+            return None
+        if not isinstance(v, str):
+            return None
+        s = v.strip().upper()
+        return None if s == "" else s
+
+    @field_validator(
+        "ace_trailer_number",
+        "seal",
+        "entity_code",
+        "plate_number",
+        "state",
+        "country",
+        "container_key",
+        mode="before",
+    )
+    @classmethod
+    def empty_optional_str_to_none(cls, v: object) -> Optional[str]:
+        """JSON often sends ''; nullable columns should get NULL, not ''."""
+        if v is None:
+            return None
+        if not isinstance(v, str):
+            return None
+        s = v.strip()
+        return None if s == "" else s
+
 
 class TrailerCreateDTO(TrailerBaseDTO):
     """Schema for creating a trailer"""

backend/api/v1/modules/a76/transportation/trailers/services.py

@@ -4,6 +4,10 @@ from sqlalchemy.orm import Session
 from sqlalchemy import text
 
 from . import dto, models
+from api.v1.modules.a76.transportation.catalog_parity import (
+    trailer_fields_to_csv_row,
+    validate_trailer_row_for_api,
+)
 
 TRAILER_ID_SEQ = "a76.trailer_trailer_id_seq"
 
@@ -75,6 +79,13 @@ class TrailerService:
     ) -> models.Trailer:
         """Create a new trailer"""
         data = trailer_data.model_dump()
+        validate_trailer_row_for_api(
+            tenant_id,
+            company_id,
+            trailer_fields_to_csv_row(data),
+            is_update=False,
+            existing_trailer_numbers=set(),
+        )
         if data.get("trailer_id") is None:
             data["trailer_id"] = allocate_trailer_id(db)
         new_trailer = models.Trailer(
@@ -102,6 +113,25 @@ class TrailerService:
         update_data = trailer_data.model_dump(
             exclude_unset=True, exclude={"trailer_number"}
         )
+        merged = {
+            "trailer_number": trailer.trailer_number,
+            "ace_trailer_number": trailer.ace_trailer_number,
+            "trailer_type_key": trailer.trailer_type_key,
+            "seal": trailer.seal,
+            "entity_code": trailer.entity_code,
+            "plate_number": trailer.plate_number,
+            "state": trailer.state,
+            "country": trailer.country,
+            "container_key": trailer.container_key,
+        }
+        merged.update(update_data)
+        validate_trailer_row_for_api(
+            tenant_id,
+            company_id,
+            trailer_fields_to_csv_row(merged),
+            is_update=True,
+            existing_trailer_numbers={trailer_number.strip()},
+        )
         for field, value in update_data.items():
             setattr(trailer, field, value)
 

backend/api/v1/modules/a76/transportation/transporters/services.py

@@ -5,6 +5,10 @@ from sqlalchemy.orm import Session
 from sqlalchemy import func, text
 
 from . import dto, models
+from api.v1.modules.a76.transportation.catalog_parity import (
+    transporter_fields_to_csv_row,
+    validate_transporter_row_for_api,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -99,6 +103,13 @@ class TransporterService:
     ) -> models.Transporter:
         """Create a new transporter"""
         data = transporter_data.model_dump()
+        validate_transporter_row_for_api(
+            tenant_id,
+            company_id,
+            transporter_fields_to_csv_row(data),
+            is_update=False,
+            existing_transporter_keys=set(),
+        )
         if data.get("transporter_id") is None:
             data["transporter_id"] = allocate_transporter_id(db)
         new_transporter = models.Transporter(
@@ -128,6 +139,35 @@ class TransporterService:
         update_data = transporter_data.model_dump(
             exclude_unset=True, exclude={"transporter_key"}
         )
+        merged = {
+            "transporter_key": transporter.transporter_key,
+            "name": transporter.name,
+            "short_name": transporter.short_name,
+            "responsible": transporter.responsible,
+            "rfc": transporter.rfc,
+            "streets": transporter.streets,
+            "postal_code": transporter.postal_code,
+            "city": transporter.city,
+            "state": transporter.state,
+            "country": transporter.country,
+            "loader_code": transporter.loader_code,
+            "caat_code": transporter.caat_code,
+            "transport_code": transporter.transport_code,
+            "transport_interface_type": transporter.transport_interface_type,
+            "ftp_server": transporter.ftp_server,
+            "ftp_user": transporter.ftp_user,
+            "ftp_password": transporter.ftp_password,
+            "ftp_directory": transporter.ftp_directory,
+            "filler_code": transporter.filler_code,
+        }
+        merged.update(update_data)
+        validate_transporter_row_for_api(
+            tenant_id,
+            company_id,
+            transporter_fields_to_csv_row(merged),
+            is_update=True,
+            existing_transporter_keys={transporter_key.strip().upper()},
+        )
         for field, value in update_data.items():
             setattr(transporter, field, value)
 

backend/api/v1/modules/a76/transportation/vehicles/services.py

@@ -4,6 +4,10 @@ from sqlalchemy.orm import Session
 from sqlalchemy import text
 
 from . import dto, models
+from api.v1.modules.a76.transportation.catalog_parity import (
+    vehicle_fields_to_csv_row,
+    validate_vehicle_row_for_api,
+)
 
 VEHICLE_ID_SEQ = "a76.vehicle_vehicle_id_seq"
 
@@ -75,6 +79,14 @@ class VehicleService:
     ) -> models.Vehicle:
         """Create a new vehicle"""
         data = vehicle_data.model_dump()
+        validate_vehicle_row_for_api(
+            db,
+            tenant_id,
+            company_id,
+            vehicle_fields_to_csv_row(data),
+            is_update=False,
+            existing_vehicle_keys=set(),
+        )
         if data.get("vehicle_id") is None:
             data["vehicle_id"] = allocate_vehicle_id(db)
         new_vehicle = models.Vehicle(
@@ -100,6 +112,45 @@ class VehicleService:
 
         # Update fields (excluding vehicle_key as it's the primary key)
         update_data = vehicle_data.model_dump(exclude_unset=True, exclude={"vehicle_key"})
+        merged = {
+            "vehicle_key": vehicle.vehicle_key,
+            "ace_vehicle_key": vehicle.ace_vehicle_key,
+            "transporter_key": vehicle.transporter_key,
+            "transport_identifier": vehicle.transport_identifier,
+            "transport_type": vehicle.transport_type,
+            "entity_code": vehicle.entity_code,
+            "transponder_number": vehicle.transponder_number,
+            "dot_number": vehicle.dot_number,
+            "plate_number": vehicle.plate_number,
+            "city": vehicle.city,
+            "state": vehicle.state,
+            "country": vehicle.country,
+            "seal": vehicle.seal,
+            "insurance_company_name": vehicle.insurance_company_name,
+            "insurance_number": vehicle.insurance_number,
+            "insurance_amount": float(vehicle.insurance_amount)
+            if vehicle.insurance_amount is not None
+            else None,
+            "insurance_date": vehicle.insurance_date,
+            "box_number": vehicle.box_number,
+            "brand": vehicle.brand,
+            "year": vehicle.year,
+            "series": vehicle.series,
+            "description": vehicle.description,
+            "engine_number": vehicle.engine_number,
+            "sct_permission": vehicle.sct_permission,
+            "color": vehicle.color,
+            "container_key": vehicle.container_key,
+        }
+        merged.update(update_data)
+        validate_vehicle_row_for_api(
+            db,
+            tenant_id,
+            company_id,
+            vehicle_fields_to_csv_row(merged),
+            is_update=True,
+            existing_vehicle_keys={vehicle_key.strip()},
+        )
         for field, value in update_data.items():
             setattr(vehicle, field, value)
 

backend/api/v1/modules/public/reference_data/trailer_types/dto.py

@@ -1,6 +1,15 @@
 from typing import Optional
 
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class TrailerTypeListItemDTO(BaseModel):
+    """Catálogo público GTipoTrailer (listado UI)."""
+
+    trailer_type_key: str = Field(..., max_length=2)
+    description: Optional[str] = None
+
+    model_config = ConfigDict(from_attributes=True)
 
 
 class TrailerTypeBaseDTO(BaseModel):

backend/api/v1/modules/public/reference_data/trailer_types/routes.py

@@ -1,5 +1,7 @@
+from typing import Any, Dict
+
 from core.database import get_core_db
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Query
 from sqlalchemy.orm import Session
 
 from . import dto, services
@@ -7,6 +9,22 @@ from . import dto, services
 router = APIRouter()
 
 
+@router.get("/trailer-types/", response_model=Dict[str, Any])
+def list_trailer_types(
+    page: int = Query(1, ge=1, description="Número de página"),
+    page_size: int = Query(50, ge=1, le=100, description="Tamaño de página"),
+    db: Session = Depends(get_core_db),
+):
+    skip = (page - 1) * page_size
+    items, total = services.TrailerTypeService.list_trailer_types(db, skip, page_size)
+    return {
+        "items": [dto.TrailerTypeListItemDTO.model_validate(obj) for obj in items],
+        "total": total,
+        "page": page,
+        "page_size": page_size,
+    }
+
+
 @router.get(
     "/trailer-types/{trailer_type_key}", response_model=dto.TrailerTypeResponseDTO
 )

backend/api/v1/modules/public/reference_data/trailer_types/services.py

@@ -1,9 +1,20 @@
+from typing import List, Tuple
+
 from sqlalchemy.orm import Session
 
 from . import dto, models
 
 
 class TrailerTypeService:
+    @staticmethod
+    def list_trailer_types(
+        db: Session, skip: int = 0, limit: int = 50
+    ) -> Tuple[List[models.TrailerType], int]:
+        q = db.query(models.TrailerType).order_by(models.TrailerType.trailer_type_key)
+        total = q.count()
+        items = q.offset(skip).limit(limit).all()
+        return items, total
+
     @staticmethod
     def get_trailer_type_by_key(db: Session, trailer_type_key: str):
         return (