merge: integrar development y resolver conflictos en traducciones

backend/api/v1/modules/a76/audit_log/events.py

@@ -2,12 +2,20 @@
 Audit Log Events
 """
 
+import logging
+
 from sqlalchemy import event, inspect
 from sqlalchemy.orm import Session
+
+from api.v1.modules.a76.general_catalogs.company.models import Company
+from core.database import rls_company_var, rls_tenant_var
+
 from .services.service import AuditService
 from .utils.serialization import serialize_for_json
 from core.context import get_user_context
 
+logger = logging.getLogger(__name__)
+
 
 def register_audit_listeners(models_to_audit):
     """
@@ -30,11 +38,37 @@ def _get_current_username():
                 or context.get("sub")
                 or "System"
             )
-    except:
+    except Exception:
         pass
     return "System"
 
 
+def _resolve_audit_company_tenant(session: Session, target) -> tuple:
+    """
+    company_id / tenant_id desde la fila ORM, ContextVars RLS (petición HTTP),
+    o ``Company.tenant_id`` por ``company_id``.
+    """
+    company_id = getattr(target, "company_id", None)
+    if company_id is None and getattr(target, "__tablename__", None) == "company":
+        company_id = getattr(target, "id", None)
+    if company_id is None:
+        company_id = rls_company_var.get()
+
+    tenant_id = getattr(target, "tenant_id", None)
+    if tenant_id is None:
+        tenant_id = rls_tenant_var.get()
+    if tenant_id is None and company_id is not None:
+        row = (
+            session.query(Company.tenant_id)
+            .filter(Company.id == company_id, Company.deleted_at.is_(None))
+            .first()
+        )
+        if row:
+            tenant_id = int(row[0])
+
+    return company_id, tenant_id
+
+
 def after_insert_listener(mapper, connection, target):
     """
     Listener for INSERT operations
@@ -42,12 +76,19 @@ def after_insert_listener(mapper, connection, target):
     table_name = target.__tablename__
     record_data = {c.name: getattr(target, c.name) for c in mapper.columns}
     username = _get_current_username()
-    company_id = getattr(target, "company_id", None) or getattr(target, "id", None)
-    tenant_id = getattr(target, "tenant_id", None)
 
-    # Create a session bound to the connection
     session = Session(bind=connection)
     try:
+        company_id, tenant_id = _resolve_audit_company_tenant(session, target)
+        if company_id is None or tenant_id is None:
+            logger.debug(
+                "Audit skip INSERT %s: missing company_id=%s tenant_id=%s",
+                table_name,
+                company_id,
+                tenant_id,
+            )
+            return
+
         AuditService.log_crud_operation(
             db=session,
             table_name=table_name,
@@ -59,7 +100,7 @@ def after_insert_listener(mapper, connection, target):
             tenant_id=tenant_id,
         )
     except Exception as e:
-        print(f"Error logging insert: {e}")
+        logger.warning("Error logging insert for %s: %s", table_name, e)
     finally:
         session.close()
 
@@ -87,11 +128,19 @@ def after_update_listener(mapper, connection, target):
 
     record_data = {c.name: getattr(target, c.name) for c in mapper.columns}
     username = _get_current_username()
-    company_id = getattr(target, "company_id", None) or getattr(target, "id", None)
-    tenant_id = getattr(target, "tenant_id", None)
 
     session = Session(bind=connection)
     try:
+        company_id, tenant_id = _resolve_audit_company_tenant(session, target)
+        if company_id is None or tenant_id is None:
+            logger.debug(
+                "Audit skip UPDATE %s: missing company_id=%s tenant_id=%s",
+                table_name,
+                company_id,
+                tenant_id,
+            )
+            return
+
         AuditService.log_crud_operation(
             db=session,
             table_name=table_name,
@@ -105,7 +154,7 @@ def after_update_listener(mapper, connection, target):
             tenant_id=tenant_id,
         )
     except Exception as e:
-        print(f"Error logging update: {e}")
+        logger.warning("Error logging update for %s: %s", table_name, e)
     finally:
         session.close()
 
@@ -117,11 +166,19 @@ def after_delete_listener(mapper, connection, target):
     table_name = target.__tablename__
     record_data = {c.name: getattr(target, c.name) for c in mapper.columns}
     username = _get_current_username()
-    company_id = getattr(target, "company_id", None) or getattr(target, "id", None)
-    tenant_id = getattr(target, "tenant_id", None)
 
     session = Session(bind=connection)
     try:
+        company_id, tenant_id = _resolve_audit_company_tenant(session, target)
+        if company_id is None or tenant_id is None:
+            logger.debug(
+                "Audit skip DELETE %s: missing company_id=%s tenant_id=%s",
+                table_name,
+                company_id,
+                tenant_id,
+            )
+            return
+
         AuditService.log_crud_operation(
             db=session,
             table_name=table_name,
@@ -133,6 +190,6 @@ def after_delete_listener(mapper, connection, target):
             tenant_id=tenant_id,
         )
     except Exception as e:
-        print(f"Error logging delete: {e}")
+        logger.warning("Error logging delete for %s: %s", table_name, e)
     finally:
         session.close()

backend/api/v1/modules/a76/audit_log/router.py

@@ -25,6 +25,19 @@ from api.v1.modules.a76.general_catalogs.company.models import Company
 
 router = APIRouter()
 
+
+def _audit_scope_tenant_id(db: Session, company_id: int) -> int:
+    """Tenant_id de la fila ``Company`` para filtrar ``audit_logs`` (alineado con lo persistido)."""
+    row = (
+        db.query(Company.tenant_id)
+        .filter(Company.id == company_id, Company.deleted_at.is_(None))
+        .first()
+    )
+    if not row:
+        raise HTTPException(status_code=404, detail="Company not found")
+    return int(row[0])
+
+
 _SEGMENT_LABELS = {
     "tenants": "Espacio",
     "companies": "Companias",
@@ -189,16 +202,17 @@ async def get_bitacora(
     """
     Bitácora por compañía. Requiere permiso ``audit_logs.view``.
     """
-    tenant_id = validate_access_to_resource(
+    validate_access_to_resource(
         db,
         company_id,
         current_user,
         required_permissions=["audit_logs.view"],
     )
+    scope_tenant_id = _audit_scope_tenant_id(db, company_id)
 
     query = db.query(AuditLog).filter(
         AuditLog.company_id == company_id,
-        AuditLog.tenant_id == tenant_id,
+        AuditLog.tenant_id == scope_tenant_id,
     )
 
     # Filters
@@ -250,18 +264,19 @@ async def get_procedures(
     """
     Lista de procedimientos para filtros (alcance compañía). Requiere ``audit_logs.view``.
     """
-    tenant_id = validate_access_to_resource(
+    validate_access_to_resource(
         db,
         company_id,
         current_user,
         required_permissions=["audit_logs.view"],
     )
+    scope_tenant_id = _audit_scope_tenant_id(db, company_id)
 
     results = (
         db.query(AuditLog.procedure)
         .filter(
             AuditLog.company_id == company_id,
-            AuditLog.tenant_id == tenant_id,
+            AuditLog.tenant_id == scope_tenant_id,
         )
         .distinct()
         .order_by(AuditLog.procedure)
@@ -279,19 +294,20 @@ async def get_audit_detail(
     """
     Detalle de un registro de bitácora. Requiere ``audit_logs.view``.
     """
-    tenant_id = validate_access_to_resource(
+    validate_access_to_resource(
         db,
         company_id,
         current_user,
         required_permissions=["audit_logs.view"],
     )
+    scope_tenant_id = _audit_scope_tenant_id(db, company_id)
 
     log = (
         db.query(AuditLog)
         .filter(
             AuditLog.spec_id == spec_id,
             AuditLog.company_id == company_id,
-            AuditLog.tenant_id == tenant_id,
+            AuditLog.tenant_id == scope_tenant_id,
         )
         .first()
     )
@@ -316,13 +332,14 @@ async def list_tenant_files(
     if not should_ensure_s3_bucket():
         raise HTTPException(status_code=400, detail="S3 storage is disabled")
 
-    tenant_id = validate_access_to_resource(
+    validate_access_to_resource(
         db,
         company_id,
         current_user,
         required_permissions=["audit_logs.view"],
     )
-    tenant_prefix = _tenant_prefix(tenant_id)
+    scope_tenant_id = _audit_scope_tenant_id(db, company_id)
+    tenant_prefix = _tenant_prefix(scope_tenant_id)
     rel_path = _normalize_relative_path(path)
     list_prefix = f"{tenant_prefix}{rel_path}/" if rel_path else tenant_prefix
 
@@ -342,7 +359,7 @@ async def list_tenant_files(
         continuation_token=continuation_token,
     )
 
-    company_names = _companies_map(db, tenant_id)
+    company_names = _companies_map(db, scope_tenant_id)
     folders: List[AuditFileFolderItem] = []
     for prefix in data.get("prefixes", []):
         rel = _relative_from_tenant_prefix(prefix, tenant_prefix)
@@ -410,13 +427,14 @@ async def download_tenant_file(
     if not should_ensure_s3_bucket():
         raise HTTPException(status_code=400, detail="S3 storage is disabled")
 
-    tenant_id = validate_access_to_resource(
+    validate_access_to_resource(
         db,
         company_id,
         current_user,
         required_permissions=["audit_logs.view"],
     )
-    tenant_prefix = _tenant_prefix(tenant_id)
+    scope_tenant_id = _audit_scope_tenant_id(db, company_id)
+    tenant_prefix = _tenant_prefix(scope_tenant_id)
     rel_path = _normalize_relative_path(path)
     if not rel_path or rel_path.endswith("/"):
         raise HTTPException(status_code=400, detail="A file path is required")

backend/api/v1/modules/a76/audit_log/services/service.py

@@ -1,6 +1,7 @@
 """
 Audit Log Service
 """
+import logging
 from datetime import datetime, timedelta, date, time
 from decimal import Decimal
 import uuid
@@ -9,8 +10,8 @@ from typing import Optional, List, Dict, Any
 from sqlalchemy.orm import Session
 from ..models import AuditLog
 from .core import AuditMapper, ReferenceGenerator
-from core.security import verify_token # keep if needed or simpler just remove if unused
-# We don't need security import here anymore as context is passed explicitly or handled by events
+
+logger = logging.getLogger(__name__)
 
 
 def _make_json_safe(obj: Any) -> Any:
@@ -188,49 +189,62 @@ class AuditService:
         )
 
     @staticmethod
-    def log_login(db: Session, username: str, ip_address: str = None, user_agent: str = None):
-        
-        # Prevent duplicate login logs (debounce 5 seconds)
-        # This handles cases where frontend might submit twice or redirects trigger re-auth
-        try:
-             # Timezone handling: Use UTC for consistency
-            now = datetime.now(pytz.UTC)
+    def log_login(
+        db: Session,
+        username: str,
+        ip_address: str = None,
+        user_agent: str = None,
+        company_id: Optional[int] = None,
+        tenant_id: Optional[int] = None,
+    ):
+        """
+        ``audit_logs`` exige ``tenant_id`` y ``company_id``. El login vía Hub no
+        define compañía activa; sin ambos argumentos no se inserta fila (antes fallaba NOT NULL).
+        """
+        if company_id is None or tenant_id is None:
+            return None
 
+        try:
+            now = datetime.now(pytz.UTC)
             five_seconds_ago = now - timedelta(seconds=5)
-            
-            # Check for recent login from same user
             existing = db.query(AuditLog).filter(
                 AuditLog.username == username,
                 AuditLog.operation_type == "LOGIN",
-                # Compare against timestamp (timezone aware)
-                AuditLog.timestamp >= five_seconds_ago
+                AuditLog.company_id == company_id,
+                AuditLog.tenant_id == tenant_id,
+                AuditLog.timestamp >= five_seconds_ago,
             ).first()
-            
             if existing:
                 return existing
-                
         except Exception as e:
-            import traceback
-            traceback.print_exc()
-        
+            logger.warning("Login audit debounce query failed: %s", e)
+
         return AuditService.create_audit_log(
-             db=db,
-             reference="LOGIN",
-             procedure="SYSTEM SCAF",
-             movement="SYSTEM LOGIN",
-             username=username,
-             operation_type="LOGIN",
-             ip_address=ip_address,
-             user_agent=user_agent
+            db=db,
+            reference="LOGIN",
+            procedure="SYSTEM SCAF",
+            movement="SYSTEM LOGIN",
+            username=username,
+            operation_type="LOGIN",
+            ip_address=ip_address,
+            user_agent=user_agent,
+            company_id=company_id,
+            tenant_id=tenant_id,
         )
 
     @staticmethod
-    def log_logout(db: Session, username: str, ip_address: str = None, user_agent: str = None):
-        """
-        Registra un evento de cierre de sesión
-        """
+    def log_logout(
+        db: Session,
+        username: str,
+        ip_address: str = None,
+        user_agent: str = None,
+        company_id: Optional[int] = None,
+        tenant_id: Optional[int] = None,
+    ):
+        """Misma condición que ``log_login``: sin alcance compañía/tenant no se escribe."""
+        if company_id is None or tenant_id is None:
+            return None
         try:
-            # Reutilizamos create_audit_log para mantener consistencia
             AuditService.create_audit_log(
                 db=db,
                 reference="LOGOUT",
@@ -239,8 +253,9 @@ class AuditService:
                 username=username,
                 operation_type="LOGOUT",
                 ip_address=ip_address,
-                user_agent=user_agent
+                user_agent=user_agent,
+                company_id=company_id,
+                tenant_id=tenant_id,
             )
         except Exception as e:
-            # No re-lanzamos la excepción para no interrumpir el flujo de logout
-            pass
+            logger.warning("Logout audit insert failed: %s", e)

backend/api/v1/modules/a76/csv_templates/registry.py

@@ -4,11 +4,10 @@ Construido a partir de los TEMPLATE_COLUMNS de cada módulo de imports.
 """
 import csv
 import io
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Set
 
 # Importar configs de cada módulo
 from api.v1.modules.a76.layouts_csv.facturas.template_config import (
-    TEMPLATE_COLUMNS as IMPORTS_TEMPLATE_COLUMNS,
     _resolve_template_columns as resolve_imports_template,
 )
 from api.v1.modules.a76.layouts_csv.parts.template_config import (
@@ -140,6 +139,7 @@ ALWAYS_REQUIRED_HEADERS_BY_TEMPLATE: Dict[str, set[str]] = {
     "imp_temp_series": {"NUMERO FACTURA", "LINEA FACTURA"},
     "imp_def_series": {"NUMERO FACTURA", "LINEA FACTURA"},
     "cmex_series": {"NUMERO FACTURA", "LINEA FACTURA"},
+    "exp_def_series": {"NUMERO FACTURA", "LINEA FACTURA"},
     # Catálogos y transportes
     "customs_brokers": {"TIPO", "CLAVE", "NOMBRE"},
     "clients_providers": {"PROCEDENCIA", "SHORT_NAME", "NOMBRE", "RFC"},
@@ -191,23 +191,35 @@ ALWAYS_REQUIRED_HEADERS_BY_TEMPLATE: Dict[str, set[str]] = {
 }
 
 
-def _apply_required_prefix(template_id: str, headers: List[str]) -> List[str]:
-    """Prefija '* ' a cabeceras siempre obligatorias para la plantilla."""
-    required_headers = ALWAYS_REQUIRED_HEADERS_BY_TEMPLATE.get(template_id)
-    if not required_headers:
-        return headers
+def _compute_required_indices_map(base_rows: Dict[str, List[str]]) -> Dict[str, Set[int]]:
+    """Índices de columnas obligatorias (según plantilla ES/base), para marcar * en cualquier idioma."""
+    out: Dict[str, Set[int]] = {}
+    for tid, headers in base_rows.items():
+        req = ALWAYS_REQUIRED_HEADERS_BY_TEMPLATE.get(tid)
+        if not req:
+            continue
+        req_norm = {_normalize_header_for_match(h) for h in req}
+        idx_set: Set[int] = set()
+        for i, h in enumerate(headers):
+            if not h:
+                continue
+            if _normalize_header_for_match(h) in req_norm:
+                idx_set.add(i)
+        out[tid] = idx_set
+    return out
 
-    required_norm = {_normalize_header_for_match(h) for h in required_headers}
+
+def _apply_required_prefix_indices(template_id: str, headers: List[str]) -> List[str]:
+    """Prefija '* ' usando índices fijos de la plantilla base (independiente del idioma de cabecera)."""
+    idx_set = _REQUIRED_INDICES.get(template_id)
+    if not idx_set:
+        return headers
     out: List[str] = []
-    for header in headers:
-        norm_header = _normalize_header_for_match(header)
-        if norm_header and norm_header in required_norm:
-            if header.startswith("* "):
-                out.append(header)
-            else:
-                out.append(f"* {header}")
+    for i, h in enumerate(headers):
+        if i in idx_set and h and not str(h).startswith("* "):
+            out.append(f"* {h}")
         else:
-            out.append(header)
+            out.append(h)
     return out
 
 
@@ -231,6 +243,7 @@ def _build_registry() -> Dict[str, List[str]]:
         "imp_def_series",
         "exp_def_header",
         "exp_def_details",
+        "exp_def_series",
         "cmex_header",
         "cmex_details",
         "cmex_series",
@@ -292,7 +305,9 @@ def _build_registry() -> Dict[str, List[str]]:
     return registry
 
 
-_TEMPLATE_HEADERS: Dict[str, List[str]] = _build_registry()
+_REGISTRY_BASE_ES: Dict[str, List[str]] = _build_registry()
+_REQUIRED_INDICES: Dict[str, Set[int]] = _compute_required_indices_map(_REGISTRY_BASE_ES)
+_TEMPLATE_HEADERS: Dict[str, List[str]] = _REGISTRY_BASE_ES
 
 # Nombre de archivo sugerido para descarga (sin path)
 TEMPLATE_FILENAMES: Dict[str, str] = {
@@ -320,15 +335,65 @@ TEMPLATE_FILENAMES: Dict[str, str] = {
     "cmex_series": "EstructuraSeriesFacComprasMex.csv",
     "exp_def_header": "EstructuraEncFacExpoCamReg.csv",
     "exp_def_details": "EstructuraParExpoCamReg.csv",
+    "exp_def_series": "EstructuraSeriesFacExpoCamReg.csv",
 }
 
 
+def get_download_headers(template_id: str, locale: Optional[str] = "es") -> Optional[List[str]]:
+    """
+    Cabeceras de descarga para la plantilla (sin prefijo *).
+    locale: es | en (default es).
+    """
+    from api.v1.modules.a76.layouts_csv.common.template_locale import (
+        download_header_cell,
+        normalize_locale,
+    )
+
+    loc = normalize_locale(locale)
+
+    cols = resolve_imports_template(template_id)
+    if cols is not None:
+        return [download_header_cell(c, loc) for c in cols]
+
+    if template_id in ("part_numbers", "items"):
+        from api.v1.modules.a76.layouts_csv.parts import template_config as parts_tc
+
+        return parts_tc.download_headers_for_locale(loc)
+
+    if template_id == "material_classes":
+        from api.v1.modules.a76.layouts_csv.classes import template_config as classes_tc
+
+        return classes_tc.download_headers_for_locale(loc)
+
+    # template_id del registry → (module_columns, clave interna del dict de columnas)
+    catalog_resolvers: list[tuple[str, dict, str]] = [
+        ("boms", BOMS_TEMPLATE_COLUMNS, "boms"),
+        ("customs_brokers", CUSTOMS_BROKERS_TEMPLATE_COLUMNS, "customs_brokers"),
+        ("clients_providers", CLIENTS_PROVIDERS_TEMPLATE_COLUMNS, "client_providers"),
+        ("exchange_rates", EXCHANGE_RATE_TEMPLATE_COLUMNS, "exchange_rates"),
+        ("american_fractions", US_TARIFF_FRACTIONS_TEMPLATE_COLUMNS, "us_tariff_fractions"),
+        ("pedimentos", PEDIMENTOS_TEMPLATE_COLUMNS, "pedimentos"),
+        ("transports", VEHICLES_TEMPLATE_COLUMNS, "vehicles"),
+        ("drivers", DRIVERS_TEMPLATE_COLUMNS, "drivers"),
+        ("trailers", TRAILERS_TEMPLATE_COLUMNS, "trailers"),
+        ("transporters", TRANSPORTERS_TEMPLATE_COLUMNS, "transporters"),
+    ]
+    for tid, mapping, inner_key in catalog_resolvers:
+        if tid != template_id:
+            continue
+        cols = mapping.get(inner_key)
+        if cols:
+            return [download_header_cell(col, loc) for col in cols]
+
+    return None
+
+
 def get_template_headers(template_id: str) -> Optional[List[str]]:
-    """Devuelve la lista de cabeceras canónicas para el template_id, o None si no existe."""
-    headers = _TEMPLATE_HEADERS.get(template_id)
-    if headers is None:
+    """Cabeceras como en descarga ES histórica, con prefijo * en obligatorias."""
+    raw = get_download_headers(template_id, "es")
+    if raw is None:
         return None
-    return _apply_required_prefix(template_id, headers)
+    return _apply_required_prefix_indices(template_id, raw)
 
 
 def get_template_filename(template_id: str) -> str:
@@ -336,14 +401,19 @@ def get_template_filename(template_id: str) -> str:
     return TEMPLATE_FILENAMES.get(template_id, f"plantilla_{template_id}.csv")
 
 
-def generate_csv_content(template_id: str, include_bom: bool = True) -> Optional[bytes]:
+def generate_csv_content(
+    template_id: str,
+    locale: str = "es",
+    include_bom: bool = True,
+) -> Optional[bytes]:
     """
     Genera el contenido CSV (solo fila de cabeceras) para el template_id.
     UTF-8, opcionalmente con BOM para Excel.
     """
-    headers = get_template_headers(template_id)
+    headers = get_download_headers(template_id, locale)
     if not headers:
         return None
+    headers = _apply_required_prefix_indices(template_id, headers)
     buf = io.StringIO()
     writer = csv.writer(buf, lineterminator="\n")
     writer.writerow(headers)

backend/api/v1/modules/a76/csv_templates/routes.py

@@ -1,9 +1,9 @@
 """
 Rutas para descargar plantillas CSV generadas desde código (sin archivos XLS/XLSX).
 """
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Query
 from fastapi.responses import Response
 
 from core.security import get_current_user
@@ -12,17 +12,23 @@ from .registry import generate_csv_content, get_template_filename
 
 router = APIRouter()
 
+_LOCALE_ALLOWED = frozenset({"es", "en"})
+
 
 @router.get("/{template_id}", response_class=Response)
 async def download_csv_template(
     template_id: str,
+    locale: Optional[str] = Query("es", description="es | en: idioma de las cabeceras del CSV"),
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
     """
     Devuelve un CSV con solo la fila de cabeceras para la plantilla indicada.
-    Las cabeceras son los nombres canónicos definidos en cada template_config.
+    Query `locale`: es (defecto) o en — cabeceras localizadas cuando estén definidas.
     """
-    content = generate_csv_content(template_id, include_bom=True)
+    loc = (locale or "es").lower().strip()
+    if loc not in _LOCALE_ALLOWED:
+        raise HTTPException(status_code=400, detail="locale must be 'es' or 'en'")
+    content = generate_csv_content(template_id, locale=loc, include_bom=True)
     if content is None:
         raise HTTPException(status_code=404, detail=f"Plantilla desconocida: {template_id}")
     filename = get_template_filename(template_id)
@@ -31,5 +37,7 @@ async def download_csv_template(
         media_type="text/csv; charset=utf-8",
         headers={
             "Content-Disposition": f'attachment; filename="{filename}"',
+            "Cache-Control": "private, no-store, max-age=0, must-revalidate",
+            "Pragma": "no-cache",
         },
     )

backend/api/v1/modules/a76/general_catalogs/company/routes.py

@@ -19,6 +19,8 @@ from core.database import get_core_db
 from core.s3_keys import company_certificate_key, company_logo_key
 from core.storage_s3 import delete_object_if_exists, get_object_bytes, put_object_bytes
 from core.security import (
+    collect_company_ids_from_app_membership,
+    collect_user_role_names,
     get_current_user,
     get_tenant_from_token,
     resolve_effective_tenant_id_from_user,
@@ -37,6 +39,48 @@ MAX_FILE_SIZE = 5 * 1024 * 1024  # 5MB
 logger = logging.getLogger(__name__)
 
 
+def _user_is_admin(current_user: dict) -> bool:
+    return "admin" in collect_user_role_names(current_user)
+
+
+def _assert_permission_any_company(
+    db: Session,
+    current_user: dict,
+    permission_code: str,
+) -> None:
+    """
+    Enforces a permission when endpoint has no explicit company_id param.
+    """
+    if _user_is_admin(current_user):
+        return
+
+    user_id = current_user.get("sub") or current_user.get("id")
+    if not user_id:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Permission denied")
+
+    company_ids = collect_company_ids_from_app_membership(db, str(user_id))
+    if not company_ids:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Permission denied")
+
+    from api.v1.modules.core.permissions.service import PermissionService
+
+    permission_service = PermissionService(db)
+    for company_id in company_ids:
+        if permission_service.has_any_permission(str(user_id), int(company_id), [permission_code]):
+            return
+
+    raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Permission denied")
+
+
+def _assert_permission_for_company(
+    db: Session,
+    company_id: int,
+    current_user: dict,
+    permission_code: str,
+) -> int:
+    return validate_access_to_resource(db, company_id, current_user, [permission_code])
+
+
 def _resolve_tenant_id_int(current_user: dict) -> int:
     """Misma lógica que validate_access_to_resource: entero estable para BD y claves S3."""
     tid = get_tenant_from_token(current_user)
@@ -87,6 +131,7 @@ async def create_company(
     db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ):
+    _assert_permission_any_company(db, current_user, "cat_company.create")
 
     tenant_id = current_user.get("tenant_id")
     if not tenant_id:
@@ -114,6 +159,8 @@ async def list_companies(
     current_user: dict = Depends(get_current_user),
 ):
     """Get paginated list of companies for current tenant with optional filters"""
+    _assert_permission_any_company(db, current_user, "cat_company.view")
+
     tenant_id = current_user.get("tenant_id")
     if not tenant_id:
         raise HTTPException(
@@ -142,7 +189,7 @@ async def list_companies(
 
     return {
         "items": [
-            CompanyResponseDTO.model_validate(service.flatten_company_dto(item)) 
+            CompanyResponseDTO.model_validate(service.flatten_company_dto(item))
             for item in items
         ],
         "total": total,
@@ -167,6 +214,8 @@ async def get_my_companies(
     Un usuario solo con roles de app y sin ``tenant_id`` en /auth/me sigue pudiendo
     listar sus compañías asignadas.
     """
+    _assert_permission_any_company(db, current_user, "cat_company.view")
+
     user_id = current_user.get("sub") or current_user.get("id")
     tenant_id = resolve_effective_tenant_id_from_user(current_user)
 
@@ -194,6 +243,8 @@ async def get_company(
     current_user: dict = Depends(get_current_user),
 ):
     """Get a specific company by ID"""
+    _assert_permission_for_company(db, company_id, current_user, "cat_company.view")
+
     tenant_id = current_user.get("tenant_id")
     if not tenant_id:
         raise HTTPException(
@@ -224,6 +275,8 @@ async def update_company(
     current_user: dict = Depends(get_current_user),
 ):
     """Update a company"""
+    _assert_permission_for_company(db, company_id, current_user, "cat_company.edit")
+
     tenant_id = current_user.get("tenant_id")
     if not tenant_id:
         raise HTTPException(
@@ -299,6 +352,8 @@ async def delete_company(
     current_user: dict = Depends(get_current_user),
 ):
     """Delete a company"""
+    _assert_permission_for_company(db, company_id, current_user, "cat_company.delete")
+
     tenant_id = current_user.get("tenant_id")
     if not tenant_id:
         raise HTTPException(
@@ -328,6 +383,8 @@ async def upload_company_logo(
     current_user: dict = Depends(get_current_user),
 ):
     """Upload a logo for a company"""
+    _assert_permission_for_company(db, company_id, current_user, "cat_company.edit")
+
     tenant_id = _resolve_tenant_id_int(current_user)
 
     # Validar que la empresa existe
@@ -409,6 +466,8 @@ async def upload_company_certificate(
     Upload a certificate for a company
     certificate_type: fiel_cer, fiel_key, cfdi_cert_cer, cfdi_cert_key, cancel_cer, cancel_key
     """
+    _assert_permission_for_company(db, company_id, current_user, "cat_company.edit")
+
     tenant_id = _resolve_tenant_id_int(current_user)
 
     # Validar que la empresa existe

backend/api/v1/modules/a76/general_catalogs/depreciation_catalog/routes.py

@@ -30,6 +30,10 @@ base_router = TenantCRUDRoutes(
     enable_filters=False,
     default_page_size=100,
     max_page_size=1000,
+    get_permissions=["cat_depreciation_catalog.view"],
+    create_permissions=["cat_depreciation_catalog.create"],
+    update_permissions=["cat_depreciation_catalog.edit"],
+    delete_permissions=["cat_depreciation_catalog.delete"],
 )
 
 router = APIRouter(prefix="/depreciation-catalog", tags=["a76 / general catalogs / depreciation catalog"])
@@ -49,7 +53,9 @@ async def list_depreciation_catalog(
     db: Session = Depends(get_core_db),
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(
+        db, company_id, current_user, ["cat_depreciation_catalog.view"]
+    )
     
     items, total = DepreciationCatalogService.get_all(
         db, tenant_id, company_id, page, page_size, search

backend/api/v1/modules/a76/general_catalogs/doda/routes.py

@@ -254,7 +254,9 @@ async def print_doda_pdf(
     Genera o reutiliza el PDF almacenado en S3 cuando el contenido no ha cambiado
     (huella SHA-256 de DODA + hijos).
     """
-    tenant_id = int(validate_access_to_resource(db, company_id, current_user))
+    tenant_id = int(
+        validate_access_to_resource(db, company_id, current_user, ["cat_doda.view"])
+    )
     doda = DodaService.get_by_id(db, doda_id, tenant_id, company_id)
     if not doda:
         raise HTTPException(
@@ -327,9 +329,12 @@ async def print_doda_pdf(
 )
 async def get_doda_containers(
     doda_id: int,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """Get all containers for a specific DODA"""
+    validate_access_to_resource(db, company_id, current_user, ["cat_doda.view"])
     containers = DodaService.get_containers(db, doda_id)
     return [DodaContainerResponseDTO.model_validate(c) for c in containers]
 
@@ -343,9 +348,12 @@ async def get_doda_containers(
 async def add_container(
     doda_id: int,
     container_data: DodaContainerCreateDTO,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """Add a new container to a DODA"""
+    validate_access_to_resource(db, company_id, current_user, ["cat_doda.edit"])
     container = DodaService.add_container(db, doda_id, container_data)
     if not container:
         raise HTTPException(
@@ -364,9 +372,12 @@ async def update_container(
     doda_id: int,
     container_line: int,
     container_data: DodaContainerUpdateDTO,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """Update a container"""
+    validate_access_to_resource(db, company_id, current_user, ["cat_doda.edit"])
     container = DodaService.update_container(
         db, doda_id, container_line, container_data
     )
@@ -386,12 +397,15 @@ async def update_container(
 async def delete_container(
     doda_id: int,
     container_line: int,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """
     Elimina un contenedor del DODA.
     Devuelve 409 si el contenedor tiene precintos (candados) asignados.
     """
+    validate_access_to_resource(db, company_id, current_user, ["cat_doda.delete"])
     DodaService.delete_container(db, doda_id, container_line)
     return None
 
@@ -415,8 +429,11 @@ def _seal_to_response(seal) -> DodaContainerSealResponseDTO:
 async def get_container_seals(
     doda_id: int,
     container_line: int,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
+    current_user: dict = Depends(get_current_user),
 ):
+    validate_access_to_resource(db, company_id, current_user, ["cat_doda.view"])
     seals = DodaService.get_seals_for_container(db, doda_id, container_line)
     return [_seal_to_response(s) for s in seals]
 
@@ -431,8 +448,11 @@ async def add_container_seal(
     doda_id: int,
     container_line: int,
     seal_data: DodaContainerSealCreateDTO,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
+    current_user: dict = Depends(get_current_user),
 ):
+    validate_access_to_resource(db, company_id, current_user, ["cat_doda.edit"])
     seal = DodaService.add_seal(db, doda_id, container_line, seal_data)
     return _seal_to_response(seal)
 
@@ -446,8 +466,11 @@ async def delete_container_seal(
     doda_id: int,
     container_line: int,
     seal_line: int,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
+    current_user: dict = Depends(get_current_user),
 ):
+    validate_access_to_resource(db, company_id, current_user, ["cat_doda.delete"])
     DodaService.delete_seal(db, doda_id, container_line, seal_line)
     return None
 
@@ -460,9 +483,12 @@ async def delete_container_seal(
 )
 async def get_american_pedimentos(
     doda_id: int,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """Get all American pedimentos for a specific DODA"""
+    validate_access_to_resource(db, company_id, current_user, ["cat_doda.view"])
     pedimentos = DodaService.get_american_pedimentos(db, doda_id)
     return [DodaAmericanPedimentoResponseDTO.model_validate(p) for p in pedimentos]
 
@@ -476,9 +502,12 @@ async def get_american_pedimentos(
 async def add_american_pedimento(
     doda_id: int,
     pedimento_data: DodaAmericanPedimentoCreateDTO,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """Add a new American pedimento to a DODA"""
+    validate_access_to_resource(db, company_id, current_user, ["cat_doda.edit"])
     pedimento = DodaService.add_american_pedimento(db, doda_id, pedimento_data)
     if not pedimento:
         raise HTTPException(
@@ -497,9 +526,12 @@ async def add_american_pedimento(
 async def delete_american_pedimento(
     doda_id: int,
     pedimento_line: int,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """Elimina un pedimento americano del DODA."""
+    validate_access_to_resource(db, company_id, current_user, ["cat_doda.delete"])
     DodaService.delete_american_pedimento(db, doda_id, pedimento_line)
     return None
 
@@ -512,9 +544,12 @@ async def delete_american_pedimento(
 )
 async def get_doda_pedimentos(
     doda_id: int,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """Get all pedimentos for a specific DODA"""
+    validate_access_to_resource(db, company_id, current_user, ["cat_doda.view"])
     pedimentos = DodaService.get_pedimentos(db, doda_id)
     return [DodaPedimentoResponseDTO.model_validate(p) for p in pedimentos]
 
@@ -528,9 +563,12 @@ async def get_doda_pedimentos(
 async def add_pedimento(
     doda_id: int,
     pedimento_data: DodaPedimentoCreateDTO,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """Add a new pedimento to a DODA"""
+    validate_access_to_resource(db, company_id, current_user, ["cat_doda.edit"])
     pedimento = DodaService.add_pedimento(db, doda_id, pedimento_data)
     if not pedimento:
         raise HTTPException(
@@ -548,9 +586,12 @@ async def add_pedimento(
 async def delete_pedimento(
     doda_id: int,
     pedimento_line: int,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """Elimina un pedimento del DODA."""
+    validate_access_to_resource(db, company_id, current_user, ["cat_doda.delete"])
     DodaService.delete_pedimento(db, doda_id, pedimento_line)
     return None
 
@@ -575,7 +616,7 @@ async def get_doda_elegibilidad(
     Porta las validaciones del sistema legacy (campos requeridos, max 4 contenedores,
     gafete si DODA, patente vs agente, certificados DODA en VU).
     """
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(db, company_id, current_user, ["cat_doda.view"])
     user_email = (
         current_user.get("email")
         or current_user.get("preferred_username")
@@ -614,7 +655,7 @@ async def post_doda_alta(
     Verifica elegibilidad, construye el payload desde los datos del DODA y su VU,
     y envía el alta al servicio externo. Retorna {task_id, status, message} para polling.
     """
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(db, company_id, current_user, ["cat_doda.edit"])
     user_email = (
         current_user.get("email")
         or current_user.get("preferred_username")
@@ -694,11 +735,14 @@ async def post_doda_alta(
 )
 async def get_doda_alta_status(
     task_id: str,
+    company_id: int = Query(..., description="Company ID"),
+    db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ) -> Any:
     """
     Proxy transparente al servicio externo para consultar el estado de una tarea de alta DODA.
     """
+    validate_access_to_resource(db, company_id, current_user, ["cat_doda.view"])
     try:
         ext = DodaExternalService()
         return ext.get_status(task_id)
@@ -727,7 +771,7 @@ async def post_doda_consulta(
     db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ) -> Dict[str, Any]:
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(db, company_id, current_user, ["cat_doda.edit"])
     user_email = (
         current_user.get("email")
         or current_user.get("preferred_username")
@@ -788,8 +832,11 @@ async def post_doda_consulta(
 )
 async def get_doda_consulta_status(
     task_id: str,
+    company_id: int = Query(..., description="Company ID"),
+    db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ) -> Any:
+    validate_access_to_resource(db, company_id, current_user, ["cat_doda.view"])
     try:
         ext = DodaExternalService()
         return ext.get_consulta_status(task_id)
@@ -818,7 +865,9 @@ async def post_doda_consulta_apply(
     db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ) -> Dict[str, Any]:
-    tenant_id = int(validate_access_to_resource(db, company_id, current_user))
+    tenant_id = int(
+        validate_access_to_resource(db, company_id, current_user, ["cat_doda.edit"])
+    )
     doda_record = DodaService.get_by_id(db, doda_id, tenant_id, company_id)
     if not doda_record:
         raise HTTPException(status_code=404, detail="DODA no encontrado.")
@@ -880,7 +929,7 @@ async def post_doda_eliminar(
     db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ) -> Dict[str, Any]:
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(db, company_id, current_user, ["cat_doda.delete"])
     user_email = (
         current_user.get("email")
         or current_user.get("preferred_username")
@@ -960,8 +1009,11 @@ async def post_doda_eliminar(
 )
 async def get_doda_eliminar_status(
     task_id: str,
+    company_id: int = Query(..., description="Company ID"),
+    db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ) -> Any:
+    validate_access_to_resource(db, company_id, current_user, ["cat_doda.view"])
     try:
         ext = DodaExternalService()
         return ext.get_eliminar_status(task_id)
@@ -996,7 +1048,7 @@ async def list_doda_alta_logs(
     db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ):
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(db, company_id, current_user, ["cat_doda.view"])
     return DodaAltaLogService.list(
         db, company_id, int(tenant_id), page, page_size, doda_id, search
     )
@@ -1014,7 +1066,7 @@ async def get_doda_alta_log(
     db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ):
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(db, company_id, current_user, ["cat_doda.view"])
     record = DodaAltaLogService.get(db, log_id, company_id, int(tenant_id))
     if not record:
         raise HTTPException(status_code=404, detail="Registro de alta DODA no encontrado.")
@@ -1034,7 +1086,7 @@ async def create_doda_alta_log(
     db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ):
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(db, company_id, current_user, ["cat_doda.create"])
     record = DodaAltaLogService.create(db, dto, company_id, int(tenant_id))
     return DodaAltaLogResponseDTO.model_validate(record)
 
@@ -1052,7 +1104,7 @@ async def update_doda_alta_log(
     db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ):
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(db, company_id, current_user, ["cat_doda.edit"])
     record = DodaAltaLogService.get(db, log_id, company_id, int(tenant_id))
     if not record:
         raise HTTPException(status_code=404, detail="Registro de alta DODA no encontrado.")
@@ -1072,7 +1124,7 @@ async def delete_doda_alta_log(
     db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ):
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(db, company_id, current_user, ["cat_doda.delete"])
     record = DodaAltaLogService.get(db, log_id, company_id, int(tenant_id))
     if not record:
         raise HTTPException(status_code=404, detail="Registro de alta DODA no encontrado.")

backend/api/v1/modules/a76/general_catalogs/error_catalogs/routes.py

@@ -4,12 +4,12 @@ Rutas para gestión de catálogos de errores
 
 from typing import List
 
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, HTTPException, Query, status
 from sqlalchemy.orm import Session
 
 from core.database import get_core_db
 from core.security import get_current_user
-from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
+from api.v1.common.tenant_crud_routes import TenantCRUDRoutes, validate_access_to_resource
 from .dto import (
     ErrorClassificationCreateDTO,
     ErrorClassificationResponseDTO,
@@ -35,6 +35,11 @@ classification_crud = TenantCRUDRoutes(
     tags=["error-classifications"],
     resource_name="ErrorClassification",
     enable_list=True,
+    list_permissions=["cat_errors.view"],
+    get_permissions=["cat_errors.view"],
+    create_permissions=["cat_errors.create"],
+    update_permissions=["cat_errors.edit"],
+    delete_permissions=["cat_errors.delete"],
 )
 
 # Add custom endpoints for classifications
@@ -47,15 +52,17 @@ classification_crud = TenantCRUDRoutes(
 )
 async def get_classification_by_code(
     code: str,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ):
     """Get an error classification by its code with all related errors"""
+    tenant_id = validate_access_to_resource(db, company_id, current_user, ["cat_errors.view"])
     classification = ErrorClassificationService.get_by_code(
         db,
         code,
-        tenant_id=current_user["tenant_id"],
-        company_id=current_user["company_id"]
+        tenant_id=tenant_id,
+        company_id=company_id,
     )
     if not classification:
         raise HTTPException(
@@ -74,15 +81,17 @@ async def get_classification_by_code(
 )
 async def get_classification(
     id: int,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ):
     """Get an error classification by its ID with all related errors"""
+    tenant_id = validate_access_to_resource(db, company_id, current_user, ["cat_errors.view"])
     classification = ErrorClassificationService.get_by_id(
         db,
         id,
-        tenant_id=current_user["tenant_id"],
-        company_id=current_user["company_id"]
+        tenant_id=tenant_id,
+        company_id=company_id,
     )
     if not classification:
         raise HTTPException(
@@ -104,6 +113,11 @@ catalog_crud = TenantCRUDRoutes(
     tags=["error-catalogs"],
     resource_name="ErrorCatalog",
     enable_list=True,
+    list_permissions=["cat_errors.view"],
+    get_permissions=["cat_errors.view"],
+    create_permissions=["cat_errors.create"],
+    update_permissions=["cat_errors.edit"],
+    delete_permissions=["cat_errors.delete"],
 )
 
 # Add custom endpoints for catalogs
@@ -116,15 +130,17 @@ catalog_crud = TenantCRUDRoutes(
 )
 async def get_error_by_code(
     code: str,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ):
     """Get an error by its code with classification details"""
+    tenant_id = validate_access_to_resource(db, company_id, current_user, ["cat_errors.view"])
     error = ErrorCatalogService.get_by_code(
         db,
         code,
-        tenant_id=current_user["tenant_id"],
-        company_id=current_user["company_id"]
+        tenant_id=tenant_id,
+        company_id=company_id,
     )
     if not error:
         raise HTTPException(
@@ -141,15 +157,17 @@ async def get_error_by_code(
 )
 async def get_errors_by_classification(
     classification_id: int,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ):
     """Get all errors for a specific classification"""
+    tenant_id = validate_access_to_resource(db, company_id, current_user, ["cat_errors.view"])
     errors = ErrorCatalogService.get_by_classification(
         db,
         classification_id,
-        tenant_id=current_user["tenant_id"],
-        company_id=current_user["company_id"]
+        tenant_id=tenant_id,
+        company_id=company_id,
     )
     return [ErrorCatalogResponseDTO.model_validate(error) for error in errors]
 
@@ -163,15 +181,17 @@ async def get_errors_by_classification(
 )
 async def get_error(
     id: int,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ):
     """Get an error by its ID with classification details"""
+    tenant_id = validate_access_to_resource(db, company_id, current_user, ["cat_errors.view"])
     error = ErrorCatalogService.get_by_id(
         db,
         id,
-        tenant_id=current_user["tenant_id"],
-        company_id=current_user["company_id"]
+        tenant_id=tenant_id,
+        company_id=company_id,
     )
     if not error:
         raise HTTPException(

backend/api/v1/modules/a76/general_catalogs/fractions/canadian_tariff_fractions/routes.py

@@ -25,9 +25,9 @@ def list_canadian_fractions(
     db: Session = Depends(get_core_db),
     current_user = Depends(get_current_user)
 ):
-    from core.security import validate_access_to_resource as validate_perm
-    validate_perm(db, company_id, current_user, ["frac_canadian.view"])
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(
+        db, company_id, current_user, ["frac_canadian.view"]
+    )
     skip = (page - 1) * page_size
     service = CanadianTariffFractionService(db)
     items, total = service.get_multi(
@@ -52,9 +52,9 @@ def get_canadian_fraction(
     db: Session = Depends(get_core_db),
     current_user = Depends(get_current_user)
 ):
-    from core.security import validate_access_to_resource as validate_perm
-    validate_perm(db, company_id, current_user, ["frac_canadian.view"])
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(
+        db, company_id, current_user, ["frac_canadian.view"]
+    )
     service = CanadianTariffFractionService(db)
     item = service.get(id, tenant_id, company_id)
     if not item:
@@ -68,9 +68,9 @@ def create_canadian_fraction(
     db: Session = Depends(get_core_db),
     current_user = Depends(get_current_user)
 ):
-    from core.security import validate_access_to_resource as validate_perm
-    validate_perm(db, company_id, current_user, ["frac_canadian.create"])
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(
+        db, company_id, current_user, ["frac_canadian.create"]
+    )
     service = CanadianTariffFractionService(db)
     return service.create(item_in, tenant_id, company_id)
 
@@ -82,9 +82,9 @@ def update_canadian_fraction(
     db: Session = Depends(get_core_db),
     current_user = Depends(get_current_user)
 ):
-    from core.security import validate_access_to_resource as validate_perm
-    validate_perm(db, company_id, current_user, ["frac_canadian.edit"])
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(
+        db, company_id, current_user, ["frac_canadian.edit"]
+    )
     service = CanadianTariffFractionService(db)
     item = service.get(id, tenant_id, company_id)
     if not item:
@@ -98,9 +98,9 @@ def delete_canadian_fraction(
     db: Session = Depends(get_core_db),
     current_user = Depends(get_current_user)
 ):
-    from core.security import validate_access_to_resource as validate_perm
-    validate_perm(db, company_id, current_user, ["frac_canadian.delete"])
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(
+        db, company_id, current_user, ["frac_canadian.delete"]
+    )
     service = CanadianTariffFractionService(db)
     item = service.get(id, tenant_id, company_id)
     if not item:

backend/api/v1/modules/a76/general_catalogs/fractions/historical_tariff_fractions/routes.py

@@ -20,9 +20,9 @@ def get_historical_fractions(
     db: Session = Depends(get_core_db),
     current_user = Depends(get_current_user)
 ):
-    from core.security import validate_access_to_resource as validate_perm
-    validate_perm(db, company_id, current_user, ["frac_historical.view"])
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(
+        db, company_id, current_user, ["frac_historical.view"]
+    )
     skip = (page - 1) * page_size
     service = HistoricalTariffFractionService(db)
     items, total = service.get_multi(tenant_id, company_id, skip=skip, limit=page_size, historical_fraction=historical_fraction)
@@ -46,9 +46,9 @@ async def get_rate(
     db: Session = Depends(get_core_db),
     current_user = Depends(get_current_user)
 ):
-    from core.security import validate_access_to_resource as validate_perm
-    validate_perm(db, company_id, current_user, ["frac_historical.view"])
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(
+        db, company_id, current_user, ["frac_historical.view"]
+    )
     
     # Parse date
     try:
@@ -80,9 +80,9 @@ def get_historical_fraction(
     db: Session = Depends(get_core_db),
     current_user = Depends(get_current_user)
 ):
-    from core.security import validate_access_to_resource as validate_perm
-    validate_perm(db, company_id, current_user, ["frac_historical.view"])
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(
+        db, company_id, current_user, ["frac_historical.view"]
+    )
     service = HistoricalTariffFractionService(db)
     fraction = service.get(id, tenant_id, company_id)
     if not fraction:
@@ -96,9 +96,9 @@ def create_historical_fraction(
     db: Session = Depends(get_core_db),
     current_user = Depends(get_current_user)
 ):
-    from core.security import validate_access_to_resource as validate_perm
-    validate_perm(db, company_id, current_user, ["frac_historical.create"])
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(
+        db, company_id, current_user, ["frac_historical.create"]
+    )
     service = HistoricalTariffFractionService(db)
     return service.create(fraction_in, tenant_id, company_id)
 
@@ -110,9 +110,9 @@ def update_historical_fraction(
     db: Session = Depends(get_core_db),
     current_user = Depends(get_current_user)
 ):
-    from core.security import validate_access_to_resource as validate_perm
-    validate_perm(db, company_id, current_user, ["frac_historical.edit"])
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(
+        db, company_id, current_user, ["frac_historical.edit"]
+    )
     service = HistoricalTariffFractionService(db)
     fraction = service.get(id, tenant_id, company_id)
     if not fraction:
@@ -126,9 +126,9 @@ def delete_historical_fraction(
     db: Session = Depends(get_core_db),
     current_user = Depends(get_current_user)
 ):
-    from core.security import validate_access_to_resource as validate_perm
-    validate_perm(db, company_id, current_user, ["frac_historical.delete"])
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(
+        db, company_id, current_user, ["frac_historical.delete"]
+    )
     service = HistoricalTariffFractionService(db)
     fraction = service.get(id, tenant_id, company_id)
     if not fraction:

backend/api/v1/modules/a76/general_catalogs/fractions/tariff_fractions/routes.py

@@ -40,8 +40,10 @@ async def list_tariff_fractions(
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
     # Validar permisos según el catálogo solicitado
-    if catalog in ["mex", "usa"]:
+    if catalog == "mex":
         validate_access_to_resource(db, company_id, current_user, ["frac_sitar.view"])
+    elif catalog == "usa":
+        validate_access_to_resource(db, company_id, current_user, ["frac_sitar_us.view"])
     elif catalog == "american":
         validate_access_to_resource(db, company_id, current_user, ["frac_american.view"])
     
@@ -83,12 +85,19 @@ async def list_tariff_fractions(
 )
 async def get_tariff_fraction(
     tariff_fraction_id: int,
+    catalog: str = Query("mex", description="Catalog source: 'mex', 'usa', or 'american'"),
     company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
-    # Por defecto asumimos vista de SITAR para este endpoint de consulta por ID general
-    validate_access_to_resource(db, company_id, current_user, ["frac_sitar.view"])
+    if catalog == "mex":
+        validate_access_to_resource(db, company_id, current_user, ["frac_sitar.view"])
+    elif catalog == "usa":
+        validate_access_to_resource(db, company_id, current_user, ["frac_sitar_us.view"])
+    elif catalog == "american":
+        validate_access_to_resource(db, company_id, current_user, ["frac_american.view"])
+    else:
+        raise HTTPException(status_code=400, detail=f"Unsupported catalog '{catalog}'")
     item = TariffFractionService.get_by_id(db, tariff_fraction_id)
     if not item:
         from fastapi import HTTPException

backend/api/v1/modules/a76/general_catalogs/fractions/us_tariff_fractions/routes.py

@@ -77,11 +77,9 @@ async def list_us_tariff_fractions(
     db: Session = Depends(get_core_db),
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
-    from core.security import validate_access_to_resource as validate_perm
-    validate_perm(db, company_id, current_user, ["frac_american.view"])
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
-    
-    validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(
+        db, company_id, current_user, ["frac_sitar_us.view"]
+    )
 
     skip = (page - 1) * page_size
     try:

backend/api/v1/modules/a76/general_catalogs/location/routes.py

@@ -17,4 +17,9 @@ router = TenantCRUDRoutes(
     enable_list=True,
     enable_filters=True,
     max_page_size=1000,
+    list_permissions=["cat_locations.view"],
+    get_permissions=["cat_locations.view"],
+    create_permissions=["cat_locations.create"],
+    update_permissions=["cat_locations.edit"],
+    delete_permissions=["cat_locations.delete"],
 ).router

backend/api/v1/modules/a76/general_catalogs/prevalidators/routes.py

@@ -4,12 +4,12 @@ Rutas para gestión de prevalidadores
 
 from typing import List
 
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, HTTPException, Query, status
 from sqlalchemy.orm import Session
 
 from core.database import get_core_db
 from core.security import get_current_user
-from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
+from api.v1.common.tenant_crud_routes import TenantCRUDRoutes, validate_access_to_resource
 from .dto import (
     PrevalidatorCreateDTO,
     PrevalidatorResponseDTO,
@@ -29,6 +29,11 @@ prevalidator_crud = TenantCRUDRoutes(
     tags=["Prevalidators"],
     resource_name="Prevalidator",
     enable_list=True,
+    list_permissions=["cat_prevalidators.view"],
+    get_permissions=["cat_prevalidators.view"],
+    create_permissions=["cat_prevalidators.create"],
+    update_permissions=["cat_prevalidators.edit"],
+    delete_permissions=["cat_prevalidators.delete"],
 )
 
 # Add custom endpoints
@@ -41,15 +46,19 @@ prevalidator_crud = TenantCRUDRoutes(
 )
 async def get_prevalidator_by_code(
     code: str,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ):
     """Get a prevalidator by its code"""
+    tenant_id = validate_access_to_resource(
+        db, company_id, current_user, ["cat_prevalidators.view"]
+    )
     prevalidator = PrevalidatorService.get_by_code(
         db,
         code,
-        tenant_id=current_user["tenant_id"],
-        company_id=current_user["company_id"]
+        tenant_id=tenant_id,
+        company_id=company_id,
     )
     if not prevalidator:
         raise HTTPException(
@@ -78,9 +87,12 @@ async def create_prevalidator(
 async def update_prevalidator(
     prevalidator_id: int,
     prevalidator_data: PrevalidatorUpdateDTO,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """Update a prevalidator"""
+    validate_access_to_resource(db, company_id, current_user, ["cat_prevalidators.edit"])
     prevalidator = PrevalidatorService.update(
         db, prevalidator_id, prevalidator_data)
     if not prevalidator:
@@ -98,9 +110,12 @@ async def update_prevalidator(
 )
 async def delete_prevalidator(
     prevalidator_id: int,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """Delete a prevalidator"""
+    validate_access_to_resource(db, company_id, current_user, ["cat_prevalidators.delete"])
     success = PrevalidatorService.delete(db, prevalidator_id)
     if not success:
         raise HTTPException(
@@ -117,10 +132,15 @@ async def delete_prevalidator(
 )
 async def get_prevalidators_by_customs(
     customs: str,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """Get all prevalidators for a specific customs"""
-    prevalidators = PrevalidatorService.get_by_customs(db, customs)
+    tenant_id = validate_access_to_resource(
+        db, company_id, current_user, ["cat_prevalidators.view"]
+    )
+    prevalidators = PrevalidatorService.get_by_customs(db, customs, tenant_id, company_id)
     return [
         PrevalidatorResponseDTO.model_validate(prevalidator)
         for prevalidator in prevalidators

backend/api/v1/modules/a76/general_catalogs/signatures/routes.py

@@ -27,6 +27,11 @@ signature_crud = TenantCRUDRoutes(
     enable_filters=True,  # Enable filtering by code
     default_page_size=50,
     max_page_size=100,
+    list_permissions=["cat_signatures.view"],
+    get_permissions=["cat_signatures.view"],
+    create_permissions=["cat_signatures.create"],
+    update_permissions=["cat_signatures.edit"],
+    delete_permissions=["cat_signatures.delete"],
 )
 
 router = signature_crud.router
@@ -44,7 +49,9 @@ async def get_signature_by_code(
     current_user: Dict[str, Any] = Depends(signature_crud.auth_dependency),
 ):
     """Get a signature by its code"""
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    tenant_id = validate_access_to_resource(
+        db, company_id, current_user, ["cat_signatures.view"]
+    )
     signature = SignatureService.get_by_code(db, code, tenant_id, company_id)
     if not signature:
         raise HTTPException(
@@ -61,9 +68,12 @@ async def get_signature_by_code(
 )
 async def delete_signature(
     signature_id: int,
+    company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
+    current_user: Dict[str, Any] = Depends(signature_crud.auth_dependency),
 ):
     """Delete a signature"""
+    validate_access_to_resource(db, company_id, current_user, ["cat_signatures.delete"])
     success = SignatureService.delete(db, signature_id)
     if not success:
         raise HTTPException(

backend/api/v1/modules/a76/layouts_csv/boms/csv_max_chars_for_headers.py

@@ -0,0 +1,11 @@
+"""Límites alineados a boms/validators/common.py y common/mappers."""
+from typing import Dict
+
+BOMS_CSV_MAX_CHARS: Dict[str, int] = {
+    "NUMPARTE_PADRE": 70,
+    "NUMPARTE_COMPONENTE": 70,
+    "CANTIDAD": 30,
+    "UNIMED": 10,
+    "VERSION_BOM": 20,
+    "VERSION_BILL": 20,
+}

backend/api/v1/modules/a76/layouts_csv/boms/template_config.py

@@ -6,6 +6,10 @@ Placeholders hasta tener el XLS definitivo; ajustar canónicos y aliases según
 from typing import Dict, List, Any
 
 from ..common.cell_value import cell_to_str
+from ..common.csv_hint_inject import inject_max_char_hints_into_columns
+from ..common.inject_csv_labels_en import inject_en_labels_into_template_columns
+from ..common.template_locale import merge_download_header_into_lookup, merge_labels_into_lookup
+from .csv_max_chars_for_headers import BOMS_CSV_MAX_CHARS
 
 TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     "boms": [
@@ -18,6 +22,17 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     ],
 }
 
+inject_en_labels_into_template_columns(TEMPLATE_COLUMNS)
+
+# UNIMED también en parts/catálogo ("commercial UOM"); en BOM es unidad de cantidad genérica.
+_BOM_LABEL_EN_OVERRIDES = {"UNIMED": "Unit of measure"}
+for col in TEMPLATE_COLUMNS.get("boms") or []:
+    canon = col.get("canonical")
+    if canon in _BOM_LABEL_EN_OVERRIDES:
+        col.setdefault("labels", {})["en"] = _BOM_LABEL_EN_OVERRIDES[canon]
+
+inject_max_char_hints_into_columns(TEMPLATE_COLUMNS.get("boms"), BOMS_CSV_MAX_CHARS)
+
 
 def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
     cols = TEMPLATE_COLUMNS.get("boms")
@@ -29,6 +44,8 @@ def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
         lookup[normalize_header_fn(canonical)] = canonical
         for alias in item.get("aliases") or []:
             lookup[normalize_header_fn(alias)] = canonical
+        merge_labels_into_lookup(item, canonical, normalize_header_fn, lookup)
+        merge_download_header_into_lookup(item, canonical, normalize_header_fn, lookup)
     return lookup
 
 

backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/template_config.py

@@ -6,6 +6,8 @@ Por ahora misma estructura que encabezado/partidas de exportación; luego se aju
 from typing import Dict, List, Any, Optional
 
 from ..common.cell_value import cell_to_str
+from ..common.inject_csv_labels_en import inject_en_labels_into_template_columns
+from ..common.template_locale import merge_download_header_into_lookup, merge_labels_into_lookup
 
 # Cambio de régimen: cam_reg_header, cam_reg_details
 # Regularización: regulariz_header, regulariz_details
@@ -108,6 +110,8 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     ],
 }
 
+inject_en_labels_into_template_columns(TEMPLATE_COLUMNS)
+
 
 def _resolve_template_columns(template_id: str) -> Optional[List[Dict[str, Any]]]:
     return TEMPLATE_COLUMNS.get(template_id)
@@ -124,6 +128,8 @@ def build_normalized_lookup(template_id: str, normalize_header_fn) -> Dict[str,
         lookup[normalize_header_fn(canonical)] = canonical
         for alias in item.get("aliases") or []:
             lookup[normalize_header_fn(alias)] = canonical
+        merge_labels_into_lookup(item, canonical, normalize_header_fn, lookup)
+        merge_download_header_into_lookup(item, canonical, normalize_header_fn, lookup)
     return lookup
 
 

backend/api/v1/modules/a76/layouts_csv/classes/csv_max_chars_for_headers.py

@@ -0,0 +1,16 @@
+"""Máximos alineados a classes/validators/common.py."""
+from typing import Dict
+
+MATERIAL_CLASSES_CSV_MAX_CHARS: Dict[str, int] = {
+    "CLASE": 30,
+    "DESCRIPCIONE": 500,
+    "DESCRIPCIONI": 500,
+    "CLAVEMAT": 10,
+    "UNIMED": 5,
+    "FRACCION": 20,
+    "FRACCIONAME": 16,
+    "TASADEPRECIA": 24,
+    "CLAVESUB": 5,
+    "REVFISICA": 10,
+    "FRACCIONEXENTAIVA": 4,
+}

backend/api/v1/modules/a76/layouts_csv/classes/template_config.py

@@ -8,10 +8,19 @@ from typing import Dict, List, Any, Optional, Tuple
 
 from ..common.cell_value import cell_to_str
 from ..common import csv_reader as common_csv_reader
+from ..common.csv_hint_inject import inject_max_char_hints_into_columns, merge_enum_hint_before_max
+from ..common.inject_csv_labels_en import inject_en_labels_into_template_columns
+from ..common.template_locale import (
+    download_header_cell,
+    merge_download_header_into_lookup,
+    merge_labels_into_lookup,
+    normalize_locale,
+)
+from .csv_max_chars_for_headers import MATERIAL_CLASSES_CSV_MAX_CHARS
 
 
 # Valores que indican que la primera fila es cabecera (primera columna normalizada)
-FIRST_COLUMN_HEADER_VALUES = ("CLAVE CLASE", "CLASE")
+FIRST_COLUMN_HEADER_VALUES = ("CLAVE CLASE", "CLASE", "CLASS CODE")
 
 def detect_headers_or_data(
     file_path: str,
@@ -71,6 +80,19 @@ TEMPLATE_DOWNLOAD_HEADERS: List[str] = [
     "CODIGO DE PRODUCTO/SERVICIO CP",
 ]
 
+TEMPLATE_DOWNLOAD_HEADERS_EN: List[str] = [
+    "CLASS CODE",
+    "DESCRIPTION (SPANISH)",
+    "DESCRIPTION (ENGLISH)",
+    "MATERIAL TYPE",
+    "COMMERCIAL UOM",
+    "MX FRACTION",
+    "US FRACTION",
+    "DEPRECIATION RATE",
+    "PHYSICAL REVIEW (1/0)",
+    "PRODUCT/SERVICE CP CODE",
+]
+
 TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     "material_classes": [
         {"canonical": "CLASE", "aliases": ["CLAVE CLASE", "CLASS", "CODIGO", "CLASE CODIGO"]},
@@ -87,6 +109,48 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     ],
 }
 
+inject_en_labels_into_template_columns(TEMPLATE_COLUMNS)
+
+# Mismos canónicos que parts (CLASE, FRACCION, …); restaurar etiquetas de catálogo de clases.
+# Mismo texto que TEMPLATE_DOWNLOAD_HEADERS_EN (plantilla Clarion EN).
+_MATERIAL_CLASS_LABEL_EN_OVERRIDES = {
+    "CLASE": "CLASS CODE",
+    "UNIMED": "COMMERCIAL UOM",
+    "FRACCION": "MX FRACTION",
+    "FRACCIONAME": "US FRACTION",
+}
+for col in TEMPLATE_COLUMNS.get("material_classes") or []:
+    canon = col.get("canonical")
+    if canon in _MATERIAL_CLASS_LABEL_EN_OVERRIDES:
+        col.setdefault("labels", {})["en"] = _MATERIAL_CLASS_LABEL_EN_OVERRIDES[canon]
+
+inject_max_char_hints_into_columns(
+    TEMPLATE_COLUMNS.get("material_classes"), MATERIAL_CLASSES_CSV_MAX_CHARS
+)
+for _mc in TEMPLATE_COLUMNS.get("material_classes") or []:
+    if _mc.get("canonical") == "REVFISICA":
+        merge_enum_hint_before_max(_mc, ["1", "0"])
+
+_MC_DL_ORDER = [
+    "CLASE",
+    "DESCRIPCIONE",
+    "DESCRIPCIONI",
+    "CLAVEMAT",
+    "UNIMED",
+    "FRACCION",
+    "FRACCIONAME",
+    "TASADEPRECIA",
+    "REVFISICA",
+    "FRACCIONEXENTAIVA",
+]
+_MC_BY_CANON = {c["canonical"]: c for c in (TEMPLATE_COLUMNS.get("material_classes") or [])}
+_MATERIAL_CLASSES_DOWNLOAD_DEFS = [dict(_MC_BY_CANON[k]) for k in _MC_DL_ORDER]
+
+
+def download_headers_for_locale(locale: str) -> List[str]:
+    loc = normalize_locale(locale)
+    return [download_header_cell(d, loc) for d in _MATERIAL_CLASSES_DOWNLOAD_DEFS]
+
 
 def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
     cols = TEMPLATE_COLUMNS.get("material_classes")
@@ -98,6 +162,15 @@ def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
         lookup[normalize_header_fn(canonical)] = canonical
         for alias in item.get("aliases") or []:
             lookup[normalize_header_fn(alias)] = canonical
+        merge_labels_into_lookup(item, canonical, normalize_header_fn, lookup)
+        merge_download_header_into_lookup(item, canonical, normalize_header_fn, lookup)
+    for es_h, en_h in zip(TEMPLATE_DOWNLOAD_HEADERS, TEMPLATE_DOWNLOAD_HEADERS_EN):
+        if not es_h or not en_h:
+            continue
+        es_key = normalize_header_fn(es_h)
+        canon = lookup.get(es_key)
+        if canon:
+            lookup[normalize_header_fn(en_h)] = canon
     return lookup
 
 

backend/api/v1/modules/a76/layouts_csv/clients_and_providers/common/csv_max_chars_for_headers.py

@@ -0,0 +1,41 @@
+"""
+Máximos de caracteres por columna canonical (plantilla client_providers) para sufijos en cabecera CSV.
+
+Debe mantenerse alineado con `common/mappers.py` → MAX_LEN y columnas String(n) en
+`clients_and_providers.models`. Sin imports de SQLAlchemy ni Pydantic para poder cargar la plantilla sin arrancar la app.
+"""
+from typing import Dict
+
+CLIENT_PROVIDER_CSV_COLUMN_MAX_CHARS: Dict[str, int] = {
+    "SHORT_NAME": 10,
+    "NOMBRE": 256,
+    "RFC": 30,
+    "DIRECCION": 100,
+    "NUM_EXT": 20,
+    "CODIGO POSTAL": 15,
+    "COLONIA": 40,
+    "CIUDAD": 30,
+    "ESTADO": 30,
+    "PAIS": 3,
+    "TELEFONO": 30,
+    "FAX": 30,
+    "EMAIL": 100,
+    "CURP": 19,
+    "TIPO_PROGRAMA_SECON": 7,
+    "NUM_PROGRAMA_SECON": 40,
+    "NUM_AUT_PROSEC": 20,
+    "REGISTRO_EMPRESA_CERT": 40,
+    "INFORMACION_EXTRA": 399,
+    "CONTACTO": 50,
+    "MANUFACTURER_ID": 25,
+    "TAX_ID_PROGRAMS": 30,
+    "BROKER_EXPO": 6,
+    "BROKER_IMPO": 6,
+    "CLAVE_TRANSFER": 8,
+    "CLAVE_WEB": 40,
+    "RESPONSABLE": 80,
+    "POSICION": 30,
+    "INCOTERM": 19,
+    "VINCULACION": 1,
+    "TRANSFORMA_SUBMAQ": 1,
+}

backend/api/v1/modules/a76/layouts_csv/clients_and_providers/template_config.py

@@ -13,13 +13,27 @@ AH=COL_EXTRA (desfase).
 from typing import Dict, List, Any, Optional
 
 from ..common.cell_value import cell_to_str
+from ..common.inject_csv_labels_en import inject_en_labels_into_template_columns
+from ..common.template_locale import (
+    merge_download_header_into_lookup,
+    merge_labels_into_lookup,
+)
+from .common.csv_max_chars_for_headers import CLIENT_PROVIDER_CSV_COLUMN_MAX_CHARS
 
 TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     "client_providers": [
-        # Col A - Procedencia (E=Extranjero, N=Nacional)
-        {"canonical": "PROCEDENCIA", "aliases": ["TIPO PROCEDENCIA", "EXTranjero/Nacional", "E/N"]},
-        # Col B - Tipo Cliente (C/P/A)
-        {"canonical": "TIPO", "aliases": ["CLIENT_OR_PROVIDER", "TIPO ENTIDAD", "CLIENTE O PROVEEDOR"]},
+        # Col A - Procedencia (E=Extranjero, N=Nacional). Manual: select N/E (≤3). Ver CSV_HEADER_HINTS.md
+        {
+            "canonical": "PROCEDENCIA",
+            "aliases": ["TIPO PROCEDENCIA", "EXTranjero/Nacional", "E/N"],
+            "csv_hint": {"kind": "enum_codes", "codes": ["N", "E"]},
+        },
+        # Col B - Tipo Cliente (C/P/A). Manual: client/provider/both (≤3). Ver CSV_HEADER_HINTS.md
+        {
+            "canonical": "TIPO",
+            "aliases": ["CLIENT_OR_PROVIDER", "TIPO ENTIDAD", "CLIENTE O PROVEEDOR"],
+            "csv_hint": {"kind": "enum_codes", "codes": ["C", "P", "A"]},
+        },
         # Col C - Clave cliente/proveedor (máx 8 Clarion)
         {"canonical": "SHORT_NAME", "aliases": ["CLAVE", "CLAVE CORTA", "NOMBRE CORTO", "SIGLAS"]},
         # Col D - Nombre
@@ -92,6 +106,34 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     ],
 }
 
+inject_en_labels_into_template_columns(TEMPLATE_COLUMNS)
+
+
+def _inject_max_char_hints_client_providers(columns: Optional[List[Dict[str, Any]]]) -> None:
+    """Añade csv_hint max_chars desde CLIENT_PROVIDER_CSV_COLUMN_MAX_CHARS (paridad modelos/MAX_LEN)."""
+    if not columns:
+        return
+    for item in columns:
+        if not isinstance(item, dict):
+            continue
+        canon = item.get("canonical")
+        if not canon:
+            continue
+        n = CLIENT_PROVIDER_CSV_COLUMN_MAX_CHARS.get(str(canon).strip())
+        if n is None:
+            continue
+        mc: Dict[str, Any] = {"kind": "max_chars", "n": n}
+        ch = item.get("csv_hint")
+        if ch is None:
+            item["csv_hint"] = mc
+        elif isinstance(ch, dict):
+            item["csv_hint"] = [ch, mc]
+        elif isinstance(ch, list):
+            item["csv_hint"] = [*ch, mc]
+
+
+_inject_max_char_hints_client_providers(TEMPLATE_COLUMNS.get("client_providers"))
+
 
 def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
     """normalized_header -> canonical_name para plantilla client_providers."""
@@ -104,6 +146,8 @@ def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
         lookup[normalize_header_fn(canonical)] = canonical
         for alias in item.get("aliases") or []:
             lookup[normalize_header_fn(alias)] = canonical
+        merge_labels_into_lookup(item, canonical, normalize_header_fn, lookup)
+        merge_download_header_into_lookup(item, canonical, normalize_header_fn, lookup)
     return lookup
 
 

backend/api/v1/modules/a76/layouts_csv/common/csv_headers_en.py

@@ -0,0 +1,408 @@
+"""
+Traducciones EN de cabeceras CSV por nombre canónico (clave interna).
+Usado para plantillas descargadas locale=en y para aliases de carga.
+"""
+
+
+def _facturas_headers() -> dict[str, str]:
+    return {
+        # Encabezados / series / partidas — alineado con messages/en.json (pedimento, exchange rate, incrementables, discharge).
+        "PEDIMENTO": "Pedimento",
+        "REMESA": "Consignment",
+        "NUMERO FACTURA": "Invoice number",
+        "FECHA FACTURA": "Invoice date",
+        "TIPO DE CAMBIO": "Exchange rate",
+        "REGIMEN": "Regime",
+        "CLAVE PROVEEDOR": "Supplier code",
+        "CLAVE VENDIDO A": "Sold-to code",
+        "CLAVE ENVIADO A": "Ship-to code",
+        "AGENTE ADUANAL": "Customs broker",
+        "CLAVE TRANSPORTISTA": "Carrier code",
+        "NOMBRE CONDUCTOR": "Driver name",
+        "TIPO TRANSPORTE": "Transport type",
+        "CLAVE TRANSPORTE": "Transport code",
+        "NUMERO CAJA": "Trailer/box number",
+        "NUMERO TRANSPORTE": "Transport number",
+        "TIPO MONEDA": "Currency type",
+        "CLAVE MONEDA": "Currency code",
+        "FLETES": "Freight",
+        "VALOR SEGUROS": "Insurance value",
+        "SEGUROS": "Insurance",
+        "EMBALAJES": "Packaging",
+        "OTROS INCREMENTABLES": "Incrementables (other)",
+        "CLAVE INCOTERM": "Incoterm code",
+        "PRECINTO": "Seal",
+        "FECHA EMISION": "Issue date",
+        "TIPO PESO": "Weight type",
+        "E DOCUMENT": "E-document",
+        "NUM OPERACION": "Operation number",
+        "ADUANA DE CRUCE": "Customs office (crossing)",
+        "OBSERVACIONES E": "Export remarks",
+        "OBSERVACIONES I": "Import remarks",
+        "FACTURA ALTERNA": "Alternate invoice",
+        "NUM PROYECTO": "Project number",
+        "ORDEN COMPRA": "Purchase order",
+        "FACTURA EXPO REF": "Export invoice reference",
+        "MANIFIESTO": "Manifest",
+        "ENVIADO POR": "Shipped by",
+        # Partidas impo temp / compartidas
+        "NUMERO FACTURA EXPO": "Export invoice number",
+        "LINEA": "Line",
+        "LINEA EXPO": "Export line",
+        "LINEA IMPO": "Import line",
+        "LINEA FACTURA": "Invoice line",
+        "LINEA SERIE": "Serial line",
+        "TIPO DE IMPO": "Import type",
+        "FACTURA IMPO": "Import invoice",
+        "GENERA DESCARGA": "Generate discharge",
+        "CANTIDAD EXPORTADA/DESCARGAR": "Exported qty / qty to discharge",
+        "CLASE": "Class",
+        "CANTIDAD IMPORTADA": "Imported quantity",
+        "UNIDAD DE MEDIDA": "Unit of measure",
+        "COSTO UNITARIO": "Unit cost",
+        "PRECIO UNITARIO": "Unit price",
+        "VALOR COMERCIAL": "Commercial value",
+        "PESO NETO": "Net weight",
+        "PESO BRUTO": "Gross weight",
+        "CANTIDAD BULTOS": "Package count",
+        "CLAVE BULTOS": "Package code",
+        "PAIS ORIGEN": "Country of origin",
+        "FRACCION ARANCELARIA": "Tariff fraction",
+        "FRACCION": "Fraction",
+        "PREFERENCIA ARANCELARIA": "Tariff preference",
+        "SECTOR": "Sector",
+        "FRACCION AMERICANA": "US tariff fraction",
+        "ORDEN DE COMPRA": "Purchase order",
+        "DESCRIPCION ESPAÑOL": "Description (Spanish)",
+        "DESCRIPCION INGLES": "Description (English)",
+        "DESCRIPCION": "Description",
+        "MARCA": "Brand",
+        "MODELO": "Model",
+        "ES PARTIDA O SUBPARTIDA": "Line or sub-line",
+        "ES PARTIDA/SUBPARTIDA": "Line or sub-line",
+        "LINEA PRINCIPAL": "Main line",
+        "NUM. PARTE": "Part number",
+        "NUMPARTE": "Part number",
+        "SE PAGO IMPUESTO": "Tax paid (Y/N)",
+        "FORMA DE PAGO": "Payment method",
+        "METODO DE VALORACION": "Valuation method",
+        "DESCRIPCION EXTRA": "Extra description",
+        "INFORMACION ADICIONAL": "Additional information",
+        "AGREGAR/SUSTITUIR": "Add/replace",
+        "AGREGAR(A)/SUSTITUIR(S)": "Add/replace",
+        "TOTAL": "Total",
+        "NUMERO ENTRADA": "Entry number",
+        "LOTE": "Lot",
+        "ID TYPE": "ID type",
+        "SERIE": "Serial",
+        "SUB MODELO": "Sub-model",
+        "NUM PARTE": "Part number",
+        "NUMERO ID": "ID number",
+        "COL_EXTRA": "Extra column",
+    }
+
+
+def _series_merge() -> dict[str, str]:
+    """Series share columns; ensure overlap with facturas_headers."""
+    return {
+        "NUMERO FACTURA": "Invoice number",
+        "LINEA FACTURA": "Invoice line",
+        "LINEA SERIE": "Serial line",
+        "SERIE": "Serial",
+        "MODELO": "Model",
+        "NUM PARTE": "Part number",
+        "SUB MODELO": "Sub-model",
+        "NUMERO ID": "ID number",
+        "COL_EXTRA": "Extra column",
+    }
+
+
+def _customs_exchange_us_fractions_boms() -> dict[str, str]:
+    return {
+        # customs_brokers
+        "TIPO": "Type",
+        "CLAVE": "Code",
+        "LICENCIA": "License",
+        "NOMBRE": "Name",
+        "RFC": "Tax ID",
+        "DIRECCION": "Address",
+        "CODIGO POSTAL": "Postal code",
+        "CIUDAD": "City",
+        "ESTADO": "State",
+        "PAIS": "Country",
+        "TELEFONO": "Phone",
+        "FAX": "Fax",
+        "EMAIL": "Email",
+        "PERSONAL_ID": "Personal ID / CURP",
+        "COL_EXTRA": "Extra column",
+        "POSICION": "Position",
+        "EMPRESA": "Company",
+        "CONTACTO": "Contact",
+        # exchange_rates
+        "FECHA": "Date",
+        "VALOR": "Exchange rate",
+        "MONEDA_LOCAL": "Local currency",
+        "MONEDA_EXTRANJERA": "Foreign currency",
+        # us_tariff_fractions
+        "FRACCION_ARANCELARIA": "Tariff fraction",
+        "PREFIJO": "Prefix",
+        "UNIDAD_DE_MEDIDA": "Unit of measure",
+        "DESCRIPCION": "Description",
+        "TIPO_DE_ADVALOREM": "Ad valorem type",
+        "ADVALOREM_PCT": "Ad valorem %",
+        "ADVALOREM_DLLS": "Ad valorem (USD)",
+        # boms
+        "NUMPARTE_PADRE": "Parent part number",
+        "NUMPARTE_COMPONENTE": "Component part number",
+        "CANTIDAD": "Quantity",
+        "UNIMED": "Unit of measure",
+        "VERSION_BOM": "BOM version",
+        "VERSION_BILL": "Bill version",
+    }
+
+
+def _clients_providers_headers() -> dict[str, str]:
+    return {
+        "PROCEDENCIA": "Origin (foreign/domestic)",
+        "TIPO": "Entity type",
+        "SHORT_NAME": "Short code",
+        "NOMBRE": "Name",
+        "RFC": "Tax ID",
+        "DIRECCION": "Address",
+        "NUM_EXT": "Exterior number",
+        "CODIGO POSTAL": "Postal code",
+        "COLONIA": "District",
+        "CIUDAD": "City",
+        "ESTADO": "State",
+        "PAIS": "Country",
+        "TELEFONO": "Phone",
+        "FAX": "Fax",
+        "EMAIL": "Email",
+        "CURP": "CURP",
+        "TIPO_PROGRAMA_SECON": "SECON program type",
+        "NUM_PROGRAMA_SECON": "SECON program number",
+        "FECHA_AUT_SECON": "SECON authorization date",
+        "ES_PROSEC": "Is PROSEC",
+        "NUM_AUT_PROSEC": "PROSEC authorization number",
+        "VINCULACION": "Linkage",
+        "ES_EMPRESA_CERTIFICADA": "Certified company",
+        "REGISTRO_EMPRESA_CERT": "Certified company registry",
+        "INFORMACION_EXTRA": "Extra information",
+        "CONTACTO": "Contact",
+        "MANUFACTURER_ID": "Manufacturer ID",
+        "TAX_ID_PROGRAMS": "Tax ID programs",
+        "BROKER_EXPO": "Export broker",
+        "BROKER_IMPO": "Import broker",
+        "CLAVE_TRANSFER": "Transfer code",
+        "TRANSFORMA_SUBMAQ": "Transformer/sub-maquila",
+        "CLAVE_WEB": "Web code",
+        "COL_EXTRA": "Extra column",
+        "RESPONSABLE": "Responsible",
+        "POSICION": "Position",
+        "INCOTERM": "Incoterm",
+        "ACTIVO": "Active",
+    }
+
+
+def _pedimentos_headers() -> dict[str, str]:
+    return {
+        "AÑO": "Year",
+        "PATENTE": "Patent",
+        "NUMERO": "Number",
+        "PEDIMENTO": "Pedimento",
+        "TIPO_OPERACION": "Operation type",
+        "TIPO_PEDIMENTO": "Pediment type",
+        "CLAVE_PEDIMENTO": "Pediment code",
+        "REGIMEN": "Regime",
+        "FECHA_INICIO": "Start date",
+        "FECHA_FINAL": "End date",
+        "FECHA_PAGO": "Payment date",
+        "FECHA_ENTRADA_RECINTO": "Compound entry date",
+        "FECHA_EXTRACCION_RECINTO": "Compound exit date",
+        "FECHA_RECIBIDO": "Received date",
+        "FECHA_AUTORIZACION": "Authorization date",
+        "FECHA_CIERRE": "Closing date",
+        "FECHA_REVISION": "Review date",
+        "ADUANA_SECCION_CRUCE": "Customs office / crossing section",
+        "ACUSE_ELECTRONICO": "Electronic acknowledgment",
+        "INDIVIDUAL_CONSOLIDADO": "Individual/consolidated",
+        "MET_TRANSP_ENTRADA": "Inbound transport method",
+        "MET_TRANSP_ARRIVO": "Arrival transport method",
+        "MET_TRANSP_SALIDA": "Outbound transport method",
+        "IEPS": "IEPS",
+        "IEPS_2": "IEPS (2)",
+        "DTA": "DTA",
+        "DTA_2": "DTA (2)",
+        "CNT": "CNT",
+        "CNT_2": "CNT (2)",
+        "PREVALIDACION": "Pre-validation",
+        "MONTO_TIGIE": "TIGIE amount",
+        "PAGO_IMPUESTO": "Tax paid (Y/N)",
+        "ES_MIXTO": "Mixed (yes/no)",
+        "OBS_RECTIFICA": "Rectification remarks",
+        "OPCION_DESTINO": "Destination option",
+        "VALOR_IVA": "VAT value",
+        "VALOR_ME": "Foreign currency value",
+        "VALOR_ADUANAS": "Customs value",
+        "VALOR_USD": "USD value",
+        "VALOR_SEGUROS": "Insurance value",
+        "FLETE": "Freight",
+        "SEGUROS": "Insurance",
+        "EMBALAJES": "Packaging",
+        "OTROS_INCREMENTABLES": "Incrementables (other)",
+        "ESTATUS": "Status",
+        "PERSONA_REV": "Reviewer",
+        "OBSERVACIONES": "Remarks",
+        "TIPO_CAMBIO": "Exchange rate",
+        "REPRESENTANTE_AA": "Customs broker representative",
+        "CLIENTE_SHORT_NAME": "Client short code",
+        "CLAVE_DEST_ORIGEN": "Destination origin code",
+        "IDENTIFICADORES": "Identifiers",
+        "CUOTAS_COMPENSATORIAS": "Compensatory quotas",
+        "ERRORES": "Errors",
+        "MULTAS": "Fines",
+        "RECARGOS": "Surcharges",
+        "PRECIO_PAGADO": "Price paid",
+        "PESO_BRUTO": "Gross weight",
+        "IVA_DE_PREV": "VAT from pre-validation",
+        "IVA_2": "VAT (2)",
+        "IGI_2": "IGI (2)",
+        "FORMA_PAGO_IVA": "VAT payment method",
+        "FORMA_PAGO_IVA_2": "VAT payment method (2)",
+        "FORMA_PAGO_IGI": "IGI payment method",
+        "FORMA_PAGO_IGI_2": "IGI payment method (2)",
+        "FORMA_PAGO_IEPS_2": "IEPS payment method (2)",
+        "FORMA_PAGO_DTA": "DTA payment method",
+        "FORMA_PAGO_DTA_2": "DTA payment method (2)",
+        "FORMA_PAGO_CNT_2": "CNT payment method (2)",
+        "FORMA_PAGO_PREVAL": "Pre-validation payment method",
+        "FORMA_PAGO_PREVALIDACION_2": "Pre-validation payment method (2)",
+    }
+
+
+def _vehicles_drivers_trailers_transporters() -> dict[str, str]:
+    return {
+        # vehicles / transports template "vehicles"
+        "CLAVE": "Code",
+        "CODIGO DE ENTIDAD": "Entity code",
+        "TIPO TRANSPORTE": "Transport type",
+        "CLAVE TRANSPORTE": "Transport code",
+        "CLAVE ACE": "ACE code",
+        "PLACAS": "Plates",
+        "PRECINTO": "Seal",
+        "NUMERO DOT": "DOT number",
+        "TRANSPONDEDOR": "Transponder",
+        "VIN": "VIN",
+        "EMPRESA ASEGURADORA": "Insurance company",
+        "NUM. ASEGURADORA": "Insurance policy number",
+        "FECHA DE ASEGURADORA": "Insurance date",
+        "MONTO ASEGURADO": "Insured amount",
+        "COL_EXTRA": "Extra column",
+        # drivers
+        "TRANSPORTISTA": "Carrier",
+        "LINEA": "Line",
+        "CLAVE CONDUCTOR": "Driver code",
+        "NOMBRE(S)": "Given name(s)",
+        "APELLIDO PATERNO": "Last name (paternal)",
+        "SEXO": "Gender",
+        "FECHA NACIMIENTO": "Birth date",
+        "PAIS NACIMIENTO": "Birth country",
+        "LICENCIA": "License",
+        "FORMA IDENTIFICACION 1": "ID type 1",
+        "NUM. IDENTIFICACION 1": "ID number 1",
+        "FORMA IDENTIFICACION 2": "ID type 2",
+        "NUM. IDENTIFICACION 2": "ID number 2",
+        "IDENTIFICACION ACE": "ACE identification",
+        "PAIS": "Country",
+        "PAIS 2": "Country 2",
+        "ESTADO": "State",
+        "ESTADO 2": "State 2",
+        "PERMISO LINEA EXPRESS": "Express line permit",
+        "PERMISO MAT. PELIGROSO": "Hazmat permit",
+        "TRANSPORTA MAT. PELIGROSO?": "Transports hazardous material?",
+        # trailers
+        "NUMERO TRAILER": "Trailer number",
+        "TIPO TRAILER": "Trailer type",
+        "CODIGO ENTIDAD": "Entity code",
+        "CLAVE CONTENEDOR": "Container code",
+        # transporters
+        "CLAVE TRANSPORTISTA": "Carrier code",
+        "NOMBRE CORTO": "Short name",
+        "RESPONSABLE": "Responsible",
+        "RFC": "Tax ID",
+        "CALLES": "Street address",
+        "CODIGO CAAT": "CAAT code",
+        "CODIGO CARGADOR": "Loader code",
+        "CODIGO TRANS": "Transport code",
+        "TIPO INTERFASE TRANS": "Carrier interface type",
+        "SERVIDOR FTP": "FTP server",
+        "USUARIO FTP": "FTP user",
+        "CLAVE ACCESO FTP": "FTP password",
+        "DIRECTORIO FTP": "FTP directory",
+    }
+
+
+def _material_classes_headers() -> dict[str, str]:
+    # Coincide con classes/template_config.TEMPLATE_DOWNLOAD_HEADERS_EN (salvo columnas solo en template extendido).
+    return {
+        "CLASE": "CLASS CODE",
+        "DESCRIPCIONE": "DESCRIPTION (SPANISH)",
+        "DESCRIPCIONI": "DESCRIPTION (ENGLISH)",
+        "CLAVEMAT": "MATERIAL TYPE",
+        "UNIMED": "COMMERCIAL UOM",
+        "FRACCION": "MX FRACTION",
+        "FRACCIONAME": "US FRACTION",
+        "TASADEPRECIA": "DEPRECIATION RATE",
+        "CLAVESUB": "SUB KEY",
+        "REVFISICA": "PHYSICAL REVIEW (1/0)",
+        "FRACCIONEXENTAIVA": "PRODUCT/SERVICE CP CODE",
+    }
+
+
+def _parts_headers() -> dict[str, str]:
+    # Alineado con parts/template_config.TEMPLATE_DOWNLOAD_HEADERS_EN.
+    return {
+        "NUMPARTE": "PART NUMBER",
+        "NUMPARTECOM": "COMMERCIAL PART NUMBER",
+        "DESCRIPCIONE": "DESCRIPTION (SPANISH)",
+        "DESCRIPCIONI": "DESCRIPTION (ENGLISH)",
+        "CLASE": "CLASS",
+        "UNIMED": "COMMERCIAL UNIT OF MEASURE",
+        "COSTOUNIT": "UNIT COST",
+        "TIPOMONEDA": "COST CURRENCY TYPE",
+        "CLAVEMONEDA": "CURRENCY CODE",
+        "PESOUNIT": "UNIT WEIGHT",
+        "TIPOPESO": "WEIGHT TYPE",
+        "FRACCION": "FRACTION",
+        "FRACCIONAME": "US FRACTION",
+        "PAIS": "COUNTRY",
+        "PREFERENCIA": "PREFERENCE",
+        "SECTOR": "SECTOR",
+        "RUTAIMAGEN": "IMAGE PATH",
+        "FDAKEY": "FDA key",
+        "FCCKEY": "FCC key",
+        "LICENCIA": "License code",
+        "ECCN": "ECCN",
+        "EXPORTCODE": "Export code",
+        "EXCLUSION": "Exclusion",
+        "ACTIVO": "Active",
+    }
+
+
+def _merge_all() -> dict[str, str]:
+    merged: dict[str, str] = {}
+    for part in (
+        _facturas_headers(),
+        _series_merge(),
+        _customs_exchange_us_fractions_boms(),
+        _clients_providers_headers(),
+        _pedimentos_headers(),
+        _vehicles_drivers_trailers_transporters(),
+        _material_classes_headers(),
+        _parts_headers(),
+    ):
+        merged.update(part)
+    return merged
+
+
+HEADER_LABEL_EN: dict[str, str] = _merge_all()

backend/api/v1/modules/a76/layouts_csv/common/csv_hint_inject.py

@@ -0,0 +1,66 @@
+"""Helpers to merge csv_hint max_chars into column definitions after inject_en_labels."""
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+
+
+def inject_max_char_hints_into_columns(
+    columns: Optional[List[Dict[str, Any]]],
+    max_by_canonical: Dict[str, int],
+) -> None:
+    """Attach or append ``{\"kind\": \"max_chars\", \"n\": N}`` per canonical (aligned with validators/models)."""
+    if not columns or not max_by_canonical:
+        return
+    for item in columns:
+        if not isinstance(item, dict):
+            continue
+        canon = item.get("canonical")
+        if not canon:
+            continue
+        n = max_by_canonical.get(str(canon).strip())
+        if n is None:
+            continue
+        mc: Dict[str, Any] = {"kind": "max_chars", "n": n}
+        ch = item.get("csv_hint")
+        if ch is None:
+            item["csv_hint"] = mc
+        elif isinstance(ch, dict):
+            item["csv_hint"] = [ch, mc]
+        elif isinstance(ch, list):
+            item["csv_hint"] = [*ch, mc]
+
+
+def append_csv_hint(item: Dict[str, Any], extra: Dict[str, Any]) -> None:
+    """Añade un hint (digits, enum adicional, etc.) tras los existentes."""
+    ch = item.get("csv_hint")
+    if ch is None:
+        item["csv_hint"] = extra
+        return
+    if isinstance(ch, dict):
+        item["csv_hint"] = [ch, extra]
+        return
+    if isinstance(ch, list):
+        item["csv_hint"] = [*ch, extra]
+
+
+def merge_enum_hint_before_max(
+    item: Dict[str, Any],
+    enum_codes: List[str],
+) -> None:
+    """Prepend enum_codes hint if not already present (for columns with fixed ≤3 values)."""
+    if len(enum_codes) == 0 or len(enum_codes) > 3:
+        return
+    enum_hint: Dict[str, Any] = {"kind": "enum_codes", "codes": list(enum_codes)}
+    ch = item.get("csv_hint")
+    if ch is None:
+        item["csv_hint"] = enum_hint
+        return
+    if isinstance(ch, dict):
+        if ch.get("kind") == "enum_codes":
+            return
+        item["csv_hint"] = [enum_hint, ch]
+        return
+    if isinstance(ch, list):
+        if any(isinstance(x, dict) and x.get("kind") == "enum_codes" for x in ch):
+            return
+        item["csv_hint"] = [enum_hint, *ch]

backend/api/v1/modules/a76/layouts_csv/common/inject_csv_labels_en.py

@@ -0,0 +1,30 @@
+"""
+Inyecta labels.en en definiciones de columnas TEMPLATE_COLUMNS tras cargar el dict.
+"""
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+
+from .csv_headers_en import HEADER_LABEL_EN
+
+
+def inject_en_labels_into_columns(columns: Optional[List[Dict[str, Any]]]) -> None:
+    """Añade item['labels']['en'] cuando existe traducción para canonical."""
+    if not columns:
+        return
+    for item in columns:
+        if not isinstance(item, dict):
+            continue
+        c = item.get("canonical")
+        if not c:
+            continue
+        en = HEADER_LABEL_EN.get(str(c).strip())
+        if en:
+            lab = item.setdefault("labels", {})
+            lab["en"] = en
+
+
+def inject_en_labels_into_template_columns(template_columns: Dict[str, Optional[List[Dict[str, Any]]]]) -> None:
+    """Recorre todas las plantillas de un TEMPLATE_COLUMNS."""
+    for cols in template_columns.values():
+        inject_en_labels_into_columns(cols)

backend/api/v1/modules/a76/layouts_csv/common/template_locale.py

@@ -0,0 +1,160 @@
+"""
+Helpers for localized CSV template header labels (ES/EN) and alias registration.
+
+csv_hint (opcional en cada columna TEMPLATE_COLUMNS): un dict o lista de dicts (orden importa).
+- kind "enum_codes" + codes: lista corta (≤3) mostrada como "Base (a, b, c)" en descarga.
+- kind "digits" + n: sufijo numérico, ES "(n dígitos)" / EN "(n digits)".
+- kind "max_chars" + n: máximo caracteres (modelo / validadores CSV).
+- kind "literal" + es / en: sufijo libre entre paréntesis por idioma.
+
+cabeceras extendidas se registran en lookup de importación vía merge_download_header_into_lookup.
+"""
+from __future__ import annotations
+
+from typing import Any, Callable, Dict, Iterator, Optional
+
+CsvHintDict = Dict[str, Any]
+
+
+def normalize_locale(locale: Optional[str]) -> str:
+    if not locale:
+        return "es"
+    l = str(locale).lower().strip()
+    if l.startswith("en"):
+        return "en"
+    return "es"
+
+
+def display_header_for_locale(column: Dict[str, Any], locale: str) -> str:
+    """
+    Pick the CSV header cell for one column definition.
+    Falls back to canonical when labels are absent.
+    For ES: no usar labels.en antes que canonical — si solo existe labels.en (inyectado),
+    sin labels.es el CSV en español debe seguir usando el nombre canónico (Clarion), no el EN.
+    """
+    loc = normalize_locale(locale)
+    labels = column.get("labels")
+    if isinstance(labels, dict):
+        if loc == "en":
+            return str(labels.get("en") or labels.get("es") or column.get("canonical") or "")
+        return str(labels.get("es") or column.get("canonical") or labels.get("en") or "")
+    return str(column.get("canonical") or "")
+
+
+def _csv_hints_iter(column: Dict[str, Any]) -> Iterator[CsvHintDict]:
+    """Siempre es generador (yield from ()) para filas sin csv_hint."""
+    h = column.get("csv_hint")
+    if h is None:
+        yield from ()
+        return
+    if isinstance(h, list):
+        for item in h:
+            if isinstance(item, dict):
+                yield item
+    elif isinstance(h, dict):
+        yield h
+
+
+def _append_suffix(text: str, suffix_body: str) -> str:
+    inner = suffix_body.strip()
+    if not inner:
+        return text
+    return f"{text} ({inner})" if text else f"({inner})"
+
+
+def _max_chars_suffix(n: int, locale: str) -> str:
+    loc = normalize_locale(locale)
+    if loc == "en":
+        if n == 1:
+            return "max. 1 character"
+        return f"max. {n} characters"
+    if n == 1:
+        return "máx. 1 carácter"
+    return f"máx. {n} caracteres"
+
+
+def _apply_one_csv_hint(text: str, hint: CsvHintDict, locale: str) -> str:
+    """Aplica un sufijo a la cabecera acumulada (display o ya con hints previos)."""
+    loc = normalize_locale(locale)
+    kind = hint.get("kind")
+    if kind == "enum_codes":
+        codes = hint.get("codes")
+        if not isinstance(codes, list):
+            return text
+        if len(codes) == 0 or len(codes) > 3:
+            return text
+        inner = ", ".join(str(c) for c in codes)
+        return _append_suffix(text, inner)
+    if kind == "digits":
+        n = hint.get("n")
+        if n is None:
+            return text
+        try:
+            ni = int(n)
+        except (TypeError, ValueError):
+            return text
+        suf = f"{ni} digits" if loc == "en" else f"{ni} dígitos"
+        return _append_suffix(text, suf)
+    if kind == "max_chars":
+        n = hint.get("n")
+        if n is None:
+            return text
+        try:
+            ni = int(n)
+        except (TypeError, ValueError):
+            return text
+        if ni < 0:
+            return text
+        return _append_suffix(text, _max_chars_suffix(ni, loc))
+    if kind == "literal":
+        lit = hint.get("en") if loc == "en" else hint.get("es")
+        if not lit and isinstance(hint.get("es"), str):
+            lit = hint.get("es")
+        if not lit and isinstance(hint.get("en"), str):
+            lit = hint.get("en")
+        if lit:
+            return _append_suffix(text, str(lit))
+        return text
+    return text
+
+
+def download_header_cell(column: Dict[str, Any], locale: Optional[str] = None) -> str:
+    """
+    Texto final de la celda de cabecera en plantilla descargada (incl. sufijos de ayuda).
+    csv_hint puede ser un dict o lista de dicts (orden: p. ej. enum_codes luego max_chars).
+    enum_codes solo si len(codes) ≤ 3.
+    """
+    loc = normalize_locale(locale)
+    text = display_header_for_locale(column, loc)
+    for hint in _csv_hints_iter(column):
+        text = _apply_one_csv_hint(text, hint, loc)
+    return text
+
+
+def merge_download_header_into_lookup(
+    item: Dict[str, Any],
+    canonical: str,
+    normalize_header_fn: Callable[[str], str],
+    lookup: Dict[str, str],
+) -> None:
+    """Registra cabeceras de descarga ES/EN (con sufijos csv_hint) como alias del canonical."""
+    for loc in ("es", "en"):
+        cell = download_header_cell(item, loc)
+        if cell:
+            lookup[normalize_header_fn(cell)] = canonical
+
+
+def merge_labels_into_lookup(
+    item: Dict[str, Any],
+    canonical: str,
+    normalize_header_fn: Callable[[str], str],
+    lookup: Dict[str, str],
+) -> None:
+    """Register labels.es / labels.en as extra normalized aliases for canonical."""
+    labels = item.get("labels")
+    if not isinstance(labels, dict):
+        return
+    for key in ("es", "en"):
+        lab = labels.get(key)
+        if lab:
+            lookup[normalize_header_fn(str(lab))] = canonical

backend/api/v1/modules/a76/layouts_csv/customs_brokers/csv_max_chars_for_headers.py

@@ -0,0 +1,23 @@
+"""Máximos y enums para cabeceras CSV agentes aduanales (validadores / mappers)."""
+from typing import Dict
+
+from .common.mappers import MAX_LEN
+
+CUSTOMS_BROKERS_CSV_MAX_CHARS: Dict[str, int] = {
+    "CLAVE": MAX_LEN["broker_key"],
+    "LICENCIA": MAX_LEN["license"],
+    "NOMBRE": MAX_LEN["name"],
+    "RFC": MAX_LEN["tax_id"],
+    "DIRECCION": MAX_LEN["address"],
+    "CODIGO POSTAL": MAX_LEN["postal_code"],
+    "CIUDAD": MAX_LEN["city"],
+    "ESTADO": MAX_LEN["state"],
+    "PAIS": MAX_LEN["country"],
+    "TELEFONO": MAX_LEN["phone"],
+    "FAX": MAX_LEN["fax"],
+    "EMAIL": MAX_LEN["email"],
+    "PERSONAL_ID": MAX_LEN["personal_id"],
+    "POSICION": MAX_LEN["position"],
+    "EMPRESA": MAX_LEN["company"],
+    "CONTACTO": MAX_LEN["contact"],
+}

backend/api/v1/modules/a76/layouts_csv/customs_brokers/template_config.py

@@ -9,6 +9,10 @@ Solo se leen columnas definidas aquí; el resto se ignora.
 from typing import Dict, List, Any, Optional
 
 from ..common.cell_value import cell_to_str
+from ..common.csv_hint_inject import inject_max_char_hints_into_columns, merge_enum_hint_before_max
+from ..common.inject_csv_labels_en import inject_en_labels_into_template_columns
+from ..common.template_locale import merge_download_header_into_lookup, merge_labels_into_lookup
+from .csv_max_chars_for_headers import CUSTOMS_BROKERS_CSV_MAX_CHARS
 
 TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     "customs_brokers": [
@@ -49,6 +53,15 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     ],
 }
 
+inject_en_labels_into_template_columns(TEMPLATE_COLUMNS)
+
+inject_max_char_hints_into_columns(
+    TEMPLATE_COLUMNS.get("customs_brokers"), CUSTOMS_BROKERS_CSV_MAX_CHARS
+)
+for _col in TEMPLATE_COLUMNS.get("customs_brokers") or []:
+    if _col.get("canonical") == "TIPO":
+        merge_enum_hint_before_max(_col, ["MEX", "AME"])
+
 # Orden oficial de columnas (para CSV sin encabezado o detección)
 CUSTOMS_BROKERS_FIELDNAMES_ORDER = [
     "TIPO",
@@ -84,6 +97,8 @@ def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
         lookup[normalize_header_fn(canonical)] = canonical
         for alias in item.get("aliases") or []:
             lookup[normalize_header_fn(alias)] = canonical
+        merge_labels_into_lookup(item, canonical, normalize_header_fn, lookup)
+        merge_download_header_into_lookup(item, canonical, normalize_header_fn, lookup)
     for idx, name in enumerate(CUSTOMS_BROKERS_FIELDNAMES_ORDER):
         lookup[normalize_header_fn(f"_COL_{idx}_")] = name
     return lookup

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

@@ -1,4 +1 @@
-# common validators, mappers, fk_loader for drivers CSV import
-from .fk_loader import load_drivers_fk_sets
-
-__all__ = ["load_drivers_fk_sets"]
+"""Drivers CSV helpers (validators, FK loader). Import submodules explicitly to avoid heavy deps at package init."""

backend/api/v1/modules/a76/layouts_csv/drivers/csv_max_chars_for_headers.py

@@ -0,0 +1,28 @@
+"""Límites alineados a drivers/common/common_validators.MAX_LEN (canonical CSV)."""
+from typing import Dict
+
+from .common.common_validators import MAX_LEN as D_MAX
+
+DRIVERS_CSV_MAX_CHARS: Dict[str, int] = {
+    "TRANSPORTISTA": D_MAX["transporter_key"],
+    "LINEA": 30,
+    "CLAVE CONDUCTOR": D_MAX["driver_name"],
+    "LICENCIA": D_MAX["license_number"],
+    "PERMISO LINEA EXPRESS": D_MAX["express_line_id"],
+    "IDENTIFICACION ACE": D_MAX["ace_id"],
+    "FECHA NACIMIENTO": 30,
+    "SEXO": D_MAX["gender"],
+    "PAIS NACIMIENTO": D_MAX["birth_country"],
+    "TRANSPORTA MAT. PELIGROSO?": D_MAX["hazardous_material_auth"],
+    "PERMISO MAT. PELIGROSO": D_MAX["hazardous_material_state"],
+    "NOMBRE(S)": D_MAX["first_name"],
+    "APELLIDO PATERNO": D_MAX["last_name"],
+    "FORMA IDENTIFICACION 1": D_MAX["id_key1"],
+    "NUM. IDENTIFICACION 1": D_MAX["id_number1"],
+    "ESTADO": D_MAX["id_state1"],
+    "PAIS": D_MAX["id_country1"],
+    "FORMA IDENTIFICACION 2": D_MAX["id_key2"],
+    "NUM. IDENTIFICACION 2": D_MAX["id_number2"],
+    "ESTADO 2": D_MAX["id_state2"],
+    "PAIS 2": D_MAX["id_country2"],
+}

backend/api/v1/modules/a76/layouts_csv/drivers/template_config.py

@@ -5,6 +5,10 @@ Configuracion de plantilla CSV para Conductores (EstructuraCatConductor.xls).
 from typing import Dict, List, Any
 
 from ..common.cell_value import cell_to_str
+from ..common.csv_hint_inject import inject_max_char_hints_into_columns, merge_enum_hint_before_max
+from ..common.inject_csv_labels_en import inject_en_labels_into_template_columns
+from ..common.template_locale import merge_download_header_into_lookup, merge_labels_into_lookup
+from .csv_max_chars_for_headers import DRIVERS_CSV_MAX_CHARS
 
 TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     "drivers": [
@@ -33,6 +37,16 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     ],
 }
 
+inject_en_labels_into_template_columns(TEMPLATE_COLUMNS)
+
+inject_max_char_hints_into_columns(TEMPLATE_COLUMNS.get("drivers"), DRIVERS_CSV_MAX_CHARS)
+for _dc in TEMPLATE_COLUMNS.get("drivers") or []:
+    c = _dc.get("canonical")
+    if c == "SEXO":
+        merge_enum_hint_before_max(_dc, ["M", "F"])
+    elif c == "TRANSPORTA MAT. PELIGROSO?":
+        merge_enum_hint_before_max(_dc, ["SI", "NO"])
+
 
 def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
     cols = TEMPLATE_COLUMNS.get("drivers")
@@ -44,6 +58,8 @@ def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
         lookup[normalize_header_fn(canonical)] = canonical
         for alias in item.get("aliases") or []:
             lookup[normalize_header_fn(alias)] = canonical
+        merge_labels_into_lookup(item, canonical, normalize_header_fn, lookup)
+        merge_download_header_into_lookup(item, canonical, normalize_header_fn, lookup)
     return lookup
 
 

backend/api/v1/modules/a76/layouts_csv/exchange_rate/csv_max_chars_for_headers.py

@@ -0,0 +1,11 @@
+"""Límites cabecera plantilla tipos de cambio."""
+from typing import Dict
+
+from .common.common_validators import CURRENCY_MAX
+
+EXCHANGE_RATE_CSV_MAX_CHARS: Dict[str, int] = {
+    "FECHA": 30,
+    "VALOR": 24,
+    "MONEDA_LOCAL": CURRENCY_MAX,
+    "MONEDA_EXTRANJERA": CURRENCY_MAX,
+}

backend/api/v1/modules/a76/layouts_csv/exchange_rate/template_config.py

@@ -6,6 +6,10 @@ Solo se leen columnas definidas aquí; el resto se ignora.
 from typing import Dict, List, Any, Optional
 
 from ..common.cell_value import cell_to_str
+from ..common.csv_hint_inject import inject_max_char_hints_into_columns
+from ..common.inject_csv_labels_en import inject_en_labels_into_template_columns
+from ..common.template_locale import merge_download_header_into_lookup, merge_labels_into_lookup
+from .csv_max_chars_for_headers import EXCHANGE_RATE_CSV_MAX_CHARS
 
 TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     "exchange_rates": [
@@ -16,6 +20,12 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     ],
 }
 
+inject_en_labels_into_template_columns(TEMPLATE_COLUMNS)
+
+inject_max_char_hints_into_columns(
+    TEMPLATE_COLUMNS.get("exchange_rates"), EXCHANGE_RATE_CSV_MAX_CHARS
+)
+
 
 def build_normalized_lookup(normalize_header_fn, template_id: str = "exchange_rates") -> Dict[str, str]:
     """normalized_header -> canonical_name para plantilla exchange_rates."""
@@ -28,6 +38,8 @@ def build_normalized_lookup(normalize_header_fn, template_id: str = "exchange_ra
         lookup[normalize_header_fn(canonical)] = canonical
         for alias in item.get("aliases") or []:
             lookup[normalize_header_fn(alias)] = canonical
+        merge_labels_into_lookup(item, canonical, normalize_header_fn, lookup)
+        merge_download_header_into_lookup(item, canonical, normalize_header_fn, lookup)
     return lookup
 
 

backend/api/v1/modules/a76/layouts_csv/exportacion/template_config.py

@@ -6,6 +6,9 @@ Misma estructura que facturas exp_def_header / exp_def_details; módulo autocont
 from typing import Dict, List, Any, Optional
 
 from ..common.cell_value import cell_to_str
+from ..common.inject_csv_labels_en import inject_en_labels_into_template_columns
+from ..common.template_locale import merge_download_header_into_lookup, merge_labels_into_lookup
+from ..facturas.invoice_csv_column_hints import apply_invoice_csv_hints
 
 # Columnas para encabezado y partidas de exportación (EstructuraEncFacExpoCamReg / EstructuraParExpoCamReg)
 TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
@@ -101,6 +104,10 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     ],
 }
 
+inject_en_labels_into_template_columns(TEMPLATE_COLUMNS)
+
+apply_invoice_csv_hints(TEMPLATE_COLUMNS)
+
 
 def _resolve_template_columns(template_id: str) -> Optional[List[Dict[str, Any]]]:
     return TEMPLATE_COLUMNS.get(template_id)
@@ -117,6 +124,8 @@ def build_normalized_lookup(template_id: str, normalize_header_fn) -> Dict[str,
         lookup[normalize_header_fn(canonical)] = canonical
         for alias in item.get("aliases") or []:
             lookup[normalize_header_fn(alias)] = canonical
+        merge_labels_into_lookup(item, canonical, normalize_header_fn, lookup)
+        merge_download_header_into_lookup(item, canonical, normalize_header_fn, lookup)
     return lookup
 
 

backend/api/v1/modules/a76/layouts_csv/facturas/invoice_csv_column_hints.py

@@ -0,0 +1,116 @@
+"""
+Hints de cabecera para plantillas de factura (longitudes conservadoras + enums ≤3).
+
+Valores orientativos respecto a validación en tasks.py y modelos de factura; afinar si cambian ORM.
+"""
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+
+from ..common.csv_hint_inject import inject_max_char_hints_into_columns, merge_enum_hint_before_max
+
+# Unión de canónicos frecuentes en encabezados y partidas (multiplantilla)
+INVOICE_MAX_CHARS_BY_CANONICAL: Dict[str, int] = {
+    "NUMERO FACTURA": 100,
+    "FECHA FACTURA": 30,
+    "FECHA EMISION": 30,
+    "TIPO DE CAMBIO": 24,
+    "REGIMEN": 10,
+    "PEDIMENTO": 30,
+    "REMESA": 12,
+    "CLAVE PROVEEDOR": 40,
+    "CLAVE VENDIDO A": 40,
+    "CLAVE ENVIADO A": 40,
+    "AGENTE ADUANAL": 20,
+    "CLAVE TRANSPORTISTA": 30,
+    "NOMBRE CONDUCTOR": 80,
+    "TIPO TRANSPORTE": 10,
+    "CLAVE TRANSPORTE": 30,
+    "NUMERO CAJA": 20,
+    "NUMERO TRANSPORTE": 30,
+    "TIPO MONEDA": 10,
+    "CLAVE MONEDA": 10,
+    "FLETES": 24,
+    "VALOR SEGUROS": 24,
+    "SEGUROS": 24,
+    "EMBALAJES": 24,
+    "OTROS INCREMENTABLES": 24,
+    "CLAVE INCOTERM": 10,
+    "PRECINTO": 49,
+    "TIPO PESO": 10,
+    "E DOCUMENT": 5,
+    "NUM OPERACION": 30,
+    "ADUANA DE CRUCE": 80,
+    "OBSERVACIONES E": 500,
+    "OBSERVACIONES I": 500,
+    "FACTURA ALTERNA": 99,
+    "NUM PROYECTO": 14,
+    "ORDEN COMPRA": 50,
+    "FACTURA EXPO REF": 19,
+    "MANIFIESTO": 50,
+    "ENVIADO POR": 80,
+    "LINEA": 10,
+    "CLASE": 20,
+    "CANTIDAD IMPORTADA": 24,
+    "UNIDAD DE MEDIDA": 10,
+    "COSTO UNITARIO": 24,
+    "PESO NETO": 24,
+    "PESO BRUTO": 24,
+    "CANTIDAD BULTOS": 24,
+    "CLAVE BULTOS": 20,
+    "PAIS ORIGEN": 3,
+    "FRACCION ARANCELARIA": 20,
+    "PREFERENCIA ARANCELARIA": 10,
+    "SECTOR": 20,
+    "FRACCION AMERICANA": 20,
+    "ORDEN DE COMPRA": 50,
+    "DESCRIPCION ESPAÑOL": 256,
+    "DESCRIPCION INGLES": 256,
+    "MARCA": 40,
+    "MODELO": 40,
+    "NUM. PARTE": 70,
+    "SE PAGO IMPUESTO": 10,
+    "FORMA DE PAGO": 30,
+    "METODO DE VALORACION": 30,
+    "DESCRIPCION EXTRA": 256,
+    "INFORMACION ADICIONAL": 500,
+    "AGREGAR/SUSTITUIR": 10,
+    "TOTAL": 24,
+    "NUMERO ENTRADA": 30,
+    "LOTE": 40,
+    "ID TYPE": 20,
+    "NUMERO FACTURA EXPO": 100,
+    "LINEA EXPO": 10,
+    "TIPO DE IMPO": 10,
+    "FACTURA IMPO": 100,
+    "LINEA IMPO": 10,
+    "GENERA DESCARGA": 5,
+    "CANTIDAD EXPORTADA/DESCARGAR": 24,
+    "AGREGAR(A)/SUSTITUIR(S)": 12,
+    "ES PARTIDA O SUBPARTIDA": 10,
+    "ES PARTIDA/SUBPARTIDA": 10,
+    "LINEA FACTURA": 10,
+    "LINEA SERIE": 10,
+    "SERIE": 40,
+    "NUM PARTE": 70,
+    "SUB MODELO": 40,
+    "NUMERO ID": 40,
+}
+
+
+def apply_invoice_csv_hints(template_columns: Dict[str, Optional[List[Dict[str, Any]]]]) -> None:
+    """Inyecta max_chars y enums en todas las listas de columnas no nulas."""
+    for _tid, cols in template_columns.items():
+        if not cols:
+            continue
+        inject_max_char_hints_into_columns(cols, INVOICE_MAX_CHARS_BY_CANONICAL)
+        for item in cols:
+            c = item.get("canonical")
+            if c == "E DOCUMENT":
+                merge_enum_hint_before_max(item, ["S", "N"])
+            elif c in ("SE PAGO IMPUESTO",):
+                merge_enum_hint_before_max(item, ["SI", "NO"])
+            elif c == "GENERA DESCARGA":
+                merge_enum_hint_before_max(item, ["SI", "NO"])
+            elif c == "TIPO DE IMPO":
+                merge_enum_hint_before_max(item, ["I", "E"])

backend/api/v1/modules/a76/layouts_csv/facturas/template_config.py

@@ -7,6 +7,9 @@ Solo se escribe en BD lo que los modelos de facturas aceptan (respetando models)
 from typing import Dict, List, Any, Optional
 
 from ..common.cell_value import cell_to_str
+from ..common.inject_csv_labels_en import inject_en_labels_into_template_columns
+from ..common.template_locale import merge_download_header_into_lookup, merge_labels_into_lookup
+from .invoice_csv_column_hints import apply_invoice_csv_hints
 
 # Cada plantilla define sus columnas canónicas y alias (otros nombres que aceptamos en el CSV).
 # canonical = nombre estándar con el que trabajamos internamente; debe coincidir con lo que
@@ -266,6 +269,10 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     ],
 }
 
+inject_en_labels_into_template_columns(TEMPLATE_COLUMNS)
+
+apply_invoice_csv_hints(TEMPLATE_COLUMNS)
+
 
 def _resolve_template_columns(template_id: str) -> Optional[List[Dict[str, Any]]]:
     cols = TEMPLATE_COLUMNS.get(template_id)
@@ -300,6 +307,8 @@ def build_normalized_lookup(template_id: str, normalize_header_fn) -> Dict[str,
         lookup[normalize_header_fn(canonical)] = canonical
         for alias in item.get("aliases") or []:
             lookup[normalize_header_fn(alias)] = canonical
+        merge_labels_into_lookup(item, canonical, normalize_header_fn, lookup)
+        merge_download_header_into_lookup(item, canonical, normalize_header_fn, lookup)
     return lookup
 
 

backend/api/v1/modules/a76/layouts_csv/parts/csv_max_chars_for_headers.py

@@ -0,0 +1,29 @@
+"""Máximos alineados a parts/validators/common.py (canonical interno)."""
+from typing import Dict
+
+PART_NUMBERS_CSV_MAX_CHARS: Dict[str, int] = {
+    "NUMPARTE": 70,
+    "NUMPARTECOM": 70,
+    "DESCRIPCIONE": 256,
+    "DESCRIPCIONI": 256,
+    "CLASE": 20,
+    "UNIMED": 5,
+    "COSTOUNIT": 24,
+    "TIPOMONEDA": 3,
+    "CLAVEMONEDA": 3,
+    "PESOUNIT": 24,
+    "TIPOPESO": 6,
+    "FRACCION": 10,
+    "FRACCIONAME": 10,
+    "PAIS": 3,
+    "PREFERENCIA": 7,
+    "SECTOR": 5,
+    "RUTAIMAGEN": 255,
+    "FDAKEY": 30,
+    "FCCKEY": 30,
+    "LICENCIA": 30,
+    "ECCN": 20,
+    "EXPORTCODE": 20,
+    "EXCLUSION": 20,
+    "ACTIVO": 5,
+}

backend/api/v1/modules/a76/layouts_csv/parts/template_config.py

@@ -8,10 +8,19 @@ from typing import Dict, List, Any, Optional, Tuple
 
 from ..common.cell_value import cell_to_str
 from ..common import csv_reader as common_csv_reader
+from ..common.csv_hint_inject import inject_max_char_hints_into_columns
+from ..common.inject_csv_labels_en import inject_en_labels_into_template_columns
+from ..common.template_locale import (
+    download_header_cell,
+    merge_download_header_into_lookup,
+    merge_labels_into_lookup,
+    normalize_locale,
+)
+from .csv_max_chars_for_headers import PART_NUMBERS_CSV_MAX_CHARS
 
 
 # Valores que indican que la primera fila es cabecera (primera columna normalizada)
-FIRST_COLUMN_HEADER_VALUES = ("NUMERO DE PARTE", "NUMPARTE")
+FIRST_COLUMN_HEADER_VALUES = ("NUMERO DE PARTE", "NUMPARTE", "PART NUMBER")
 
 
 def detect_headers_or_data(
@@ -76,6 +85,27 @@ TEMPLATE_DOWNLOAD_HEADERS: List[str] = [
     "RUTA DE LA IMAGEN",
 ]
 
+# Misma longitud que TEMPLATE_DOWNLOAD_HEADERS; columnas I/J vacías.
+TEMPLATE_DOWNLOAD_HEADERS_EN: List[str] = [
+    "PART NUMBER",
+    "DESCRIPTION (SPANISH)",
+    "DESCRIPTION (ENGLISH)",
+    "CLASS",
+    "COMMERCIAL UNIT OF MEASURE",
+    "UNIT COST",
+    "COST CURRENCY TYPE",
+    "CURRENCY CODE",
+    "",
+    "",
+    "UNIT WEIGHT",
+    "WEIGHT TYPE",
+    "FRACTION",
+    "COUNTRY",
+    "PREFERENCE",
+    "SECTOR",
+    "IMAGE PATH",
+]
+
 # Para lectura cuando la primera fila es dato (sin cabecera): nombres únicos para columnas I y J
 TEMPLATE_FIELDNAMES_FOR_READING: List[str] = [
     "NUMERO DE PARTE",
@@ -126,6 +156,53 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     ],
 }
 
+inject_en_labels_into_template_columns(TEMPLATE_COLUMNS)
+
+inject_max_char_hints_into_columns(
+    TEMPLATE_COLUMNS.get("part_numbers"), PART_NUMBERS_CSV_MAX_CHARS
+)
+
+_PART_ORDER_DOWNLOAD = [
+    "NUMPARTE",
+    "DESCRIPCIONE",
+    "DESCRIPCIONI",
+    "CLASE",
+    "UNIMED",
+    "COSTOUNIT",
+    "TIPOMONEDA",
+    "CLAVEMONEDA",
+    None,
+    None,
+    "PESOUNIT",
+    "TIPOPESO",
+    "FRACCION",
+    "PAIS",
+    "PREFERENCIA",
+    "SECTOR",
+    "RUTAIMAGEN",
+]
+
+_PN_BY_CANON = {c["canonical"]: c for c in (TEMPLATE_COLUMNS.get("part_numbers") or [])}
+
+
+def _parts_download_column_defs() -> List[Dict[str, Any]]:
+    out: List[Dict[str, Any]] = []
+    for key in _PART_ORDER_DOWNLOAD:
+        if key is None:
+            out.append({"canonical": ""})
+        else:
+            base = _PN_BY_CANON.get(key)
+            out.append(dict(base) if base else {"canonical": key})
+    return out
+
+
+_PARTS_DOWNLOAD_DEFS = _parts_download_column_defs()
+
+
+def download_headers_for_locale(locale: str) -> List[str]:
+    loc = normalize_locale(locale)
+    return [download_header_cell(d, loc) for d in _PARTS_DOWNLOAD_DEFS]
+
 
 def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
     cols = TEMPLATE_COLUMNS.get("part_numbers")
@@ -137,6 +214,16 @@ def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
         lookup[normalize_header_fn(canonical)] = canonical
         for alias in item.get("aliases") or []:
             lookup[normalize_header_fn(alias)] = canonical
+        merge_labels_into_lookup(item, canonical, normalize_header_fn, lookup)
+        merge_download_header_into_lookup(item, canonical, normalize_header_fn, lookup)
+    # Plantilla EN: mismas posiciones que TEMPLATE_DOWNLOAD_HEADERS
+    for es_h, en_h in zip(TEMPLATE_DOWNLOAD_HEADERS, TEMPLATE_DOWNLOAD_HEADERS_EN):
+        if not es_h or not en_h:
+            continue
+        es_key = normalize_header_fn(es_h)
+        canon = lookup.get(es_key)
+        if canon:
+            lookup[normalize_header_fn(en_h)] = canon
     return lookup
 
 

backend/api/v1/modules/a76/layouts_csv/pedmientos/pedimentos_csv_hints.py

@@ -0,0 +1,58 @@
+"""Hints cabecera CSV pedimentos (dígitos col Clarion, enums acotados, longitudes típicas)."""
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+
+from ..common.csv_hint_inject import (
+    append_csv_hint,
+    inject_max_char_hints_into_columns,
+    merge_enum_hint_before_max,
+)
+
+# Cols 1–3: solo dígitos con longitud fija
+_PEDIMENTOS_DIGITS: Dict[str, int] = {
+    "AÑO": 2,
+    "PATENTE": 4,
+    "NUMERO": 7,
+}
+
+# Resto de campos con tope razonable (validadores/modelos variados)
+_PEDIMENTOS_MAX_CHARS: Dict[str, int] = {
+    "TIPO_OPERACION": 8,
+    "CLAVE_PEDIMENTO": 30,
+    "REGIMEN": 10,
+    "FECHA_INICIO": 30,
+    "FECHA_FINAL": 30,
+    "FECHA_PAGO": 30,
+    "ADUANA_SECCION_CRUCE": 120,
+    "ACUSE_ELECTRONICO": 80,
+    "INDIVIDUAL_CONSOLIDADO": 10,
+    "MET_TRANSP_ENTRADA": 30,
+    "MET_TRANSP_ARRIVO": 30,
+    "MET_TRANSP_SALIDA": 30,
+    "PEDIMENTO": 25,
+    "CLIENTE_SHORT_NAME": 30,
+    "TIPO_PEDIMENTO": 20,
+    "OBSERVACIONES": 500,
+}
+
+
+def apply_pedimentos_csv_hints(columns: Optional[List[Dict[str, Any]]]) -> None:
+    if not columns:
+        return
+    inject_max_char_hints_into_columns(columns, _PEDIMENTOS_MAX_CHARS)
+    for item in columns:
+        canon = item.get("canonical")
+        if not canon:
+            continue
+        nd = _PEDIMENTOS_DIGITS.get(str(canon).strip())
+        if nd is not None:
+            append_csv_hint(item, {"kind": "digits", "n": nd})
+        if canon == "TIPO_OPERACION":
+            merge_enum_hint_before_max(item, ["I", "E"])
+        elif canon == "PAGO_IMPUESTO":
+            merge_enum_hint_before_max(item, ["S", "N"])
+        elif canon == "ES_MIXTO":
+            merge_enum_hint_before_max(item, ["SI", "NO"])
+        elif canon == "INDIVIDUAL_CONSOLIDADO":
+            merge_enum_hint_before_max(item, ["IND", "CON"])

backend/api/v1/modules/a76/layouts_csv/pedmientos/template_config.py

@@ -10,6 +10,9 @@ from typing import Dict, List, Any, Optional, Tuple
 # Convierte valor de celda a str; si es lista (p. ej. CSV con columnas duplicadas), toma el primer elemento.
 # Re-exportado desde common para uso en validators; ver layouts_csv.common.cell_value.
 from ..common.cell_value import cell_to_str as _cell_to_str
+from ..common.inject_csv_labels_en import inject_en_labels_into_template_columns
+from ..common.template_locale import merge_download_header_into_lookup, merge_labels_into_lookup
+from .pedimentos_csv_hints import apply_pedimentos_csv_hints
 from ..common import csv_reader as common_csv_reader
 
 
@@ -116,6 +119,10 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     ],
 }
 
+inject_en_labels_into_template_columns(TEMPLATE_COLUMNS)
+
+apply_pedimentos_csv_hints(TEMPLATE_COLUMNS.get("pedimentos"))
+
 # Orden de columnas para CSV sin cabecera (primera fila = datos). Usado por detect_headers_or_data.
 PEDIMENTOS_TEMPLATE_ORDER: List[str] = [
     item["canonical"] for item in TEMPLATE_COLUMNS["pedimentos"]
@@ -190,6 +197,8 @@ def build_normalized_lookup(normalize_header_fn, template_id: str = "pedimentos"
         lookup[normalize_header_fn(canonical)] = canonical
         for alias in item.get("aliases") or []:
             lookup[normalize_header_fn(alias)] = canonical
+        merge_labels_into_lookup(item, canonical, normalize_header_fn, lookup)
+        merge_download_header_into_lookup(item, canonical, normalize_header_fn, lookup)
     return lookup
 
 

backend/api/v1/modules/a76/layouts_csv/trailers/csv_max_chars_for_headers.py

@@ -0,0 +1,16 @@
+"""Límites alineados a trailers/common/common_validators.MAX_LEN."""
+from typing import Dict
+
+from .common.common_validators import MAX_LEN as T_MAX
+
+TRAILERS_CSV_MAX_CHARS: Dict[str, int] = {
+    "NUMERO TRAILER": T_MAX["trailer_number"],
+    "CLAVE ACE": T_MAX["ace_trailer_number"],
+    "TIPO TRAILER": T_MAX["trailer_type_key"],
+    "PRECINTO": T_MAX["seal"],
+    "CODIGO ENTIDAD": T_MAX["entity_code"],
+    "PLACAS": T_MAX["plate_number"],
+    "ESTADO": T_MAX["state"],
+    "PAIS": T_MAX["country"],
+    "CLAVE CONTENEDOR": T_MAX["container_key"],
+}

backend/api/v1/modules/a76/layouts_csv/trailers/template_config.py

@@ -6,6 +6,10 @@ Mapeo: NUMERO TRAILER → trailer_number, CLAVE ACE → ace_trailer_number, etc.
 from typing import Dict, List, Any
 
 from ..common.cell_value import cell_to_str
+from ..common.csv_hint_inject import inject_max_char_hints_into_columns
+from ..common.inject_csv_labels_en import inject_en_labels_into_template_columns
+from ..common.template_locale import merge_download_header_into_lookup, merge_labels_into_lookup
+from .csv_max_chars_for_headers import TRAILERS_CSV_MAX_CHARS
 
 TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     "trailers": [
@@ -22,6 +26,10 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     ],
 }
 
+inject_en_labels_into_template_columns(TEMPLATE_COLUMNS)
+
+inject_max_char_hints_into_columns(TEMPLATE_COLUMNS.get("trailers"), TRAILERS_CSV_MAX_CHARS)
+
 
 def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
     cols = TEMPLATE_COLUMNS.get("trailers")
@@ -33,6 +41,8 @@ def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
         lookup[normalize_header_fn(canonical)] = canonical
         for alias in item.get("aliases") or []:
             lookup[normalize_header_fn(alias)] = canonical
+        merge_labels_into_lookup(item, canonical, normalize_header_fn, lookup)
+        merge_download_header_into_lookup(item, canonical, normalize_header_fn, lookup)
     return lookup
 
 

backend/api/v1/modules/a76/layouts_csv/transportistas/csv_max_chars_for_headers.py

@@ -0,0 +1,25 @@
+"""Límites alineados a transportistas/common/common_validators.MAX_LEN."""
+from typing import Dict
+
+from .common.common_validators import MAX_LEN as TR_MAX
+
+TRANSPORTISTAS_CSV_MAX_CHARS: Dict[str, int] = {
+    "CLAVE TRANSPORTISTA": TR_MAX["transporter_key"],
+    "NOMBRE": TR_MAX["name"],
+    "NOMBRE CORTO": TR_MAX["short_name"],
+    "RESPONSABLE": TR_MAX["responsible"],
+    "RFC": TR_MAX["rfc"],
+    "CALLES": TR_MAX["streets"],
+    "CODIGO POSTAL": TR_MAX["postal_code"],
+    "CIUDAD": TR_MAX["city"],
+    "ESTADO": TR_MAX["state"],
+    "PAIS": TR_MAX["country"],
+    "CODIGO CARGADOR": TR_MAX["loader_code"],
+    "CODIGO CAAT": TR_MAX["caat_code"],
+    "CODIGO TRANS": TR_MAX["transport_code"],
+    "TIPO INTERFASE TRANS": TR_MAX["transport_interface_type"],
+    "SERVIDOR FTP": TR_MAX["ftp_server"],
+    "USUARIO FTP": TR_MAX["ftp_user"],
+    "CLAVE ACCESO FTP": TR_MAX["ftp_password"],
+    "DIRECTORIO FTP": TR_MAX["ftp_directory"],
+}

backend/api/v1/modules/a76/layouts_csv/transportistas/template_config.py

@@ -6,6 +6,10 @@ Mapeo Clarion: Col A = CLAVE TRANSPORTISTA, B = NOMBRE, ... R = DIRECTORIO FTP,
 from typing import Dict, List, Any
 
 from ..common.cell_value import cell_to_str
+from ..common.csv_hint_inject import inject_max_char_hints_into_columns
+from ..common.inject_csv_labels_en import inject_en_labels_into_template_columns
+from ..common.template_locale import merge_download_header_into_lookup, merge_labels_into_lookup
+from .csv_max_chars_for_headers import TRANSPORTISTAS_CSV_MAX_CHARS
 
 TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     "transporters": [
@@ -31,6 +35,12 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     ],
 }
 
+inject_en_labels_into_template_columns(TEMPLATE_COLUMNS)
+
+inject_max_char_hints_into_columns(
+    TEMPLATE_COLUMNS.get("transporters"), TRANSPORTISTAS_CSV_MAX_CHARS
+)
+
 
 def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
     cols = TEMPLATE_COLUMNS.get("transporters")
@@ -42,6 +52,8 @@ def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
         lookup[normalize_header_fn(canonical)] = canonical
         for alias in item.get("aliases") or []:
             lookup[normalize_header_fn(alias)] = canonical
+        merge_labels_into_lookup(item, canonical, normalize_header_fn, lookup)
+        merge_download_header_into_lookup(item, canonical, normalize_header_fn, lookup)
     return lookup
 
 

backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/csv_max_chars_for_headers.py

@@ -0,0 +1,14 @@
+"""Fracción americana: common_validators CODE_MAX, etc."""
+from typing import Dict
+
+from .common.common_validators import CODE_MAX, PREFIX_MAX, UNIT_MAX, TYPE_MAX
+
+US_TARIFF_FRACTIONS_CSV_MAX_CHARS: Dict[str, int] = {
+    "FRACCION_ARANCELARIA": CODE_MAX,
+    "PREFIJO": PREFIX_MAX,
+    "UNIDAD_DE_MEDIDA": UNIT_MAX,
+    "DESCRIPCION": 500,
+    "TIPO_DE_ADVALOREM": TYPE_MAX,
+    "ADVALOREM_PCT": 24,
+    "ADVALOREM_DLLS": 24,
+}

backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/template_config.py

@@ -6,6 +6,10 @@ Solo se leen columnas definidas aquí; el resto se ignora.
 from typing import Dict, List, Any, Optional
 
 from ..common.cell_value import cell_to_str as _cell_to_str
+from ..common.csv_hint_inject import inject_max_char_hints_into_columns, merge_enum_hint_before_max
+from ..common.inject_csv_labels_en import inject_en_labels_into_template_columns
+from ..common.template_locale import merge_download_header_into_lookup, merge_labels_into_lookup
+from .csv_max_chars_for_headers import US_TARIFF_FRACTIONS_CSV_MAX_CHARS
 
 TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     "us_tariff_fractions": [
@@ -20,6 +24,15 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     ],
 }
 
+inject_en_labels_into_template_columns(TEMPLATE_COLUMNS)
+
+inject_max_char_hints_into_columns(
+    TEMPLATE_COLUMNS.get("us_tariff_fractions"), US_TARIFF_FRACTIONS_CSV_MAX_CHARS
+)
+for _uc in TEMPLATE_COLUMNS.get("us_tariff_fractions") or []:
+    if _uc.get("canonical") == "TIPO_DE_ADVALOREM":
+        merge_enum_hint_before_max(_uc, ["PO", "ME"])
+
 # Orden A..H para detectar desfase en 8ª columna (raw_row.values()[7])
 TEMPLATE_ORDER = [item["canonical"] for item in TEMPLATE_COLUMNS["us_tariff_fractions"]]
 DESFASE_COLUMN_INDEX = 7
@@ -36,6 +49,8 @@ def build_normalized_lookup(normalize_header_fn, template_id: str = "us_tariff_f
         lookup[normalize_header_fn(canonical)] = canonical
         for alias in item.get("aliases") or []:
             lookup[normalize_header_fn(alias)] = canonical
+        merge_labels_into_lookup(item, canonical, normalize_header_fn, lookup)
+        merge_download_header_into_lookup(item, canonical, normalize_header_fn, lookup)
     return lookup
 
 

backend/api/v1/modules/a76/layouts_csv/vehicles/csv_max_chars_for_headers.py

@@ -0,0 +1,24 @@
+"""Límites de cabecera alineados a vehicles/common/common_validators.MAX_LEN."""
+from typing import Dict
+
+from .common.common_validators import MAX_LEN as V_MAX
+
+VEHICLES_CSV_MAX_CHARS: Dict[str, int] = {
+    "CLAVE": V_MAX["vehicle_key"],
+    "CLAVE ACE": V_MAX["ace_vehicle_key"],
+    "CLAVE TRANSPORTE": V_MAX["transporter_key"],
+    "VIN": V_MAX["transport_identifier"],
+    "TIPO TRANSPORTE": V_MAX["transport_type"],
+    "CODIGO DE ENTIDAD": V_MAX["entity_code"],
+    "TRANSPONDEDOR": V_MAX["transponder_number"],
+    "NUMERO DOT": V_MAX["dot_number"],
+    "PLACAS": V_MAX["plate_number"],
+    "CIUDAD": V_MAX["city"],
+    "ESTADO": V_MAX["state"],
+    "PAIS": V_MAX["country"],
+    "PRECINTO": V_MAX["seal"],
+    "EMPRESA ASEGURADORA": V_MAX["insurance_company_name"],
+    "NUM. ASEGURADORA": V_MAX["insurance_number"],
+    "MONTO ASEGURADO": 24,
+    "FECHA DE ASEGURADORA": 30,
+}

backend/api/v1/modules/a76/layouts_csv/vehicles/template_config.py

@@ -6,6 +6,10 @@ Mapeo: CLAVE → vehicle_key, CLAVE ACE → ace_vehicle_key, etc.
 from typing import Dict, List, Any
 
 from ..common.cell_value import cell_to_str
+from ..common.csv_hint_inject import inject_max_char_hints_into_columns
+from ..common.inject_csv_labels_en import inject_en_labels_into_template_columns
+from ..common.template_locale import merge_download_header_into_lookup, merge_labels_into_lookup
+from .csv_max_chars_for_headers import VEHICLES_CSV_MAX_CHARS
 
 TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     "vehicles": [
@@ -30,6 +34,10 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     ],
 }
 
+inject_en_labels_into_template_columns(TEMPLATE_COLUMNS)
+
+inject_max_char_hints_into_columns(TEMPLATE_COLUMNS.get("vehicles"), VEHICLES_CSV_MAX_CHARS)
+
 
 def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
     cols = TEMPLATE_COLUMNS.get("vehicles")
@@ -41,6 +49,8 @@ def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
         lookup[normalize_header_fn(canonical)] = canonical
         for alias in item.get("aliases") or []:
             lookup[normalize_header_fn(alias)] = canonical
+        merge_labels_into_lookup(item, canonical, normalize_header_fn, lookup)
+        merge_download_header_into_lookup(item, canonical, normalize_header_fn, lookup)
     return lookup
 
 

backend/api/v1/modules/core/permissions/seed_v2.py

@@ -77,12 +77,9 @@ permissions_general = (
     gen_crud("cat_broker_concepts", "Conceptos de Agente Aduanal", "general_catalogs") +
     gen_crud("cat_classification", "Clasificación de Conceptos", "general_catalogs") +
     gen_crud("cat_identifiers", "Identificadores", "general_catalogs") +
-    gen_crud("cat_incoterms", "Incoterms (Catálogo)", "general_catalogs") +
     gen_crud("cat_inpc", "INPC", "general_catalogs") +
     gen_crud("cat_legends", "Leyendas Fijas", "general_catalogs") +
     gen_crud("cat_seals", "Sellos", "general_catalogs") +
-    gen_crud("cat_valuation", "Métodos de Valoración", "general_catalogs") +
-    gen_crud("cat_countries", "Países", "general_catalogs") +
     gen_crud("cat_ports", "Puertos", "general_catalogs") +
     gen_crud("cat_um_general", "UdM Generales", "general_catalogs") +
     gen_crud("cat_um_customs", "UdM Aduana MEX", "general_catalogs") +
@@ -92,17 +89,15 @@ permissions_general = (
     gen_crud("cat_unit_conversions", "Conversiones", "general_catalogs") +
     gen_crud("cat_equivalencies", "Equivalencias", "general_catalogs") +
     gen_crud("cat_exchange_rates", "Tipos de Cambio", "general_catalogs") +
-    gen_crud("cat_currency", "Tipos de Moneda", "general_catalogs") +
     gen_crud("cat_multi_currency_types", "Tipos de Moneda (Múltiples)", "general_catalogs") +
-    gen_crud("cat_inv_types", "Tipos de Factura", "general_catalogs") +
     gen_crud("cat_signatures", "Firmas Electrónicas", "general_catalogs") +
     gen_crud("cat_errors", "Catálogo de Errores", "general_catalogs") +
     gen_crud("cat_doda", "DODA", "general_catalogs") +
     gen_crud("cat_prevalidators", "Prevalidadores", "general_catalogs") +
     gen_crud("cat_notices", "Avisos Electrónicos", "general_catalogs") +
-    gen_crud("cat_crossing", "Avisos de Cruce", "general_catalogs") +
-    gen_crud("cat_warehouses", "Recintos", "general_catalogs") +
-    gen_crud("cat_sectors", "Sectores", "general_catalogs")
+    gen_crud("cat_sectors", "Sectores", "general_catalogs") +
+    gen_crud("cat_locations", "Ubicaciones", "general_catalogs") +
+    gen_crud("cat_depreciation_catalog", "Catálogo de Depreciación", "general_catalogs")
 )
 
 # ============================================================================