diff --git a/backend/api/v1/modules/a76/layouts_csv/classes/TEST_DATA_README.md b/backend/api/v1/modules/a76/layouts_csv/classes/TEST_DATA_README.md
deleted file mode 100644
index 29cf1111..00000000
--- a/backend/api/v1/modules/a76/layouts_csv/classes/TEST_DATA_README.md
+++ /dev/null
@@ -1,36 +0,0 @@
-# Datos de prueba – Import CSV Clases de Materiales
-
-## Archivo: `test_data_clases.csv`
-
-### Cómo usar
-
-1. **Catálogos necesarios**: El CSV usa códigos que deben existir en tu tenant/empresa:
-   - **TIPO DE MATERIAL**: p. ej. `MAT01` (catálogo Tipos de Activo Fijo / MaterialType).
-   - **U.M. COMERCIAL**: p. ej. `KG`, `MTR` (catálogo Unidades de Medida por tenant/company).
-   - **FRACCION ARANCELARIA**: mínimo 8 caracteres y que exista en Fracciones Arancelarias Sifr@ o Histórico (p. ej. `12345678`, `87654321`).
-   - **FRACCION AMERICANA**: que exista en catálogo Fracciones Americanas (p. ej. `1234567890123456`, `6543210987654321`).
-
-   Si no tienes esos códigos, crea al menos uno de cada catálogo o sustituye en el CSV por códigos reales de tu base.
-
-2. **Filas válidas (para probar carga correcta)**  
-   - Líneas 2 y 3: `CLASE01`, `CLASE02` — completas y correctas (ajusta tipos de material, U.M. y fracciones a tus catálogos).
-
-3. **Filas que disparan errores (para probar mensajes)**  
-   - **Línea 4**: Todas las celdas vacías → *"La columna de Clase esta vacio..."* (Col. A vacía).  
-   - **Línea 5**: `CLASE04` sin B,D,E,F → *"Existen campos vacios que son obligatorios..."* (validación completa).  
-   - **Línea 6**: Clase de más de 8 caracteres (`ABCDEFGHIJ`) → *"La Clase: ... supera la longitud de caracteres"*.  
-   - **Línea 7**: Tipo de material `INVALIDO` (no en catálogo) → *"(Col. D) El Tipo de Activo Fijo: INVALIDO no existe..."*.  
-   - **Línea 8**: U.M. `UMINE` (no en catálogo) → *"(Col. E) La Unidad de Medida Comercial: UMINE no existe..."*.  
-   - **Línea 9**: Fracción mexicana `1234` (< 8 caracteres) → *"La Fraccion 1234 no alcanza la longitud de 8 caracteres"*.  
-   - **Línea 10**: Fracción americana `INVALIDO` (no en catálogo) → *"(Col. G) La Fraccion Americana: INVALIDO no existe..."*.  
-   - **Línea 11**: Tasa de depreciación `150` (> 100) → *"(Col. H) La Tasa de Depreciación: 150 no puede ser mayor al 100 %"*.  
-   - **Línea 12**: Código producto CP `9999` (si existe catálogo CP y no está) → *"(Col. J) La clase: CLASE10 tiene asignado un código de producto inexistente"*.
-
-### CSV sin cabecera (opcional)
-
-Si quieres probar detección de “primera fila = datos”, usa un archivo cuya primera línea sea una fila de datos (no "CLAVE CLASE"...). El sistema usará `TEMPLATE_DOWNLOAD_HEADERS` como cabecera y la primera línea como dato.
-
-### Valores mínimos para una fila válida
-
-Con **Modo Actualizar** y clase ya existente, solo es obligatoria la Col. A (CLAVE CLASE).  
-Con **Modo Reemplazar** o clase nueva, son obligatorios: A, B, D, E, F (mínimo 8 caracteres y en catálogo), y el resto según reglas (H ≤ 100, J en catálogo CP si aplica).
diff --git a/backend/api/v1/modules/a76/layouts_csv/common/responses.py b/backend/api/v1/modules/a76/layouts_csv/common/responses.py
index f122c9cf..af33f9fb 100644
--- a/backend/api/v1/modules/a76/layouts_csv/common/responses.py
+++ b/backend/api/v1/modules/a76/layouts_csv/common/responses.py
@@ -9,16 +9,24 @@ def scan_result(
     processed_rows: int,
     error_count: int,
     errors_detail: List[Dict[str, Any]],
+    total_rows_in_file: Optional[int] = None,
+    message: Optional[str] = None,
 ) -> Dict[str, Any]:
-    """Respuesta de scan_file (waiting_confirmation)."""
-    return {
+    """Respuesta de scan_file (waiting_confirmation).
+    Si total_rows_in_file no se pasa, se usa processed_rows como total (comportamiento anterior).
+    """
+    total = total_rows_in_file if total_rows_in_file is not None else processed_rows
+    out: Dict[str, Any] = {
         "status": "waiting_confirmation",
         "job_id": job_id,
-        "total_rows": processed_rows,
+        "total_rows": total,
         "error_count": error_count,
         "valid_rows": processed_rows - error_count,
         "errors": errors_detail,
     }
+    if message:
+        out["message"] = message
+    return out
 
 
 def commit_result(
diff --git a/backend/api/v1/modules/a76/layouts_csv/pedmientos/common/common_validators.py b/backend/api/v1/modules/a76/layouts_csv/pedmientos/common/common_validators.py
index 64bb9980..3f9609ba 100644
--- a/backend/api/v1/modules/a76/layouts_csv/pedmientos/common/common_validators.py
+++ b/backend/api/v1/modules/a76/layouts_csv/pedmientos/common/common_validators.py
@@ -1,8 +1,40 @@
 """
 Validadores reutilizables para import CSV de pedimentos.
+Incluye parseo de fechas (paridad exchange_rate) para FECHA_INICIO, FECHA_FINAL, FECHA_PAGO.
 """
+from datetime import datetime, time
 from decimal import Decimal, InvalidOperation
-from typing import Dict, Any, Optional, Set
+from typing import Dict, Any, Optional, Set, List
+
+DATE_FORMATS: List[str] = ["%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y", "%d-%m-%Y", "%Y/%m/%d"]
+DATE_FORMAT_PREFERENCE_MAP: Dict[str, str] = {
+    "dd/mm/yyyy": "%d/%m/%Y",
+    "mm/dd/yyyy": "%m/%d/%Y",
+    "yyyy-mm-dd": "%Y-%m-%d",
+}
+
+
+def parse_date(
+    val: Optional[str], date_format_preference: Optional[str] = None
+) -> Optional[datetime]:
+    """Parsea fecha; si date_format_preference está definido, solo se acepta ese formato."""
+    if not val or not str(val).strip():
+        return None
+    raw = str(val).strip()
+    if date_format_preference and date_format_preference in DATE_FORMAT_PREFERENCE_MAP:
+        fmt = DATE_FORMAT_PREFERENCE_MAP[date_format_preference]
+        try:
+            parsed = datetime.strptime(raw, fmt)
+            return datetime.combine(parsed.date(), time.min)
+        except ValueError:
+            return None
+    for fmt in DATE_FORMATS:
+        try:
+            parsed = datetime.strptime(raw, fmt)
+            return datetime.combine(parsed.date(), time.min)
+        except ValueError:
+            continue
+    return None
 
 
 def check_required_max(row: Dict[str, Any], col: str, max_len: int, line_num: int) -> Optional[Dict[str, Any]]:
@@ -29,6 +61,42 @@ def check_int_in_set(
     return None
 
 
+def check_optional_int_in_set(
+    row: Dict[str, Any], col: str, valid_ids: Set[int], line_num: int
+) -> Optional[Dict[str, Any]]:
+    """Si la columna viene vacía no se valida; si trae valor debe ser entero y estar en valid_ids."""
+    val = (row.get(col) or "").strip()
+    if not val:
+        return None
+    try:
+        client_id = int(val)
+    except ValueError:
+        return {"line": line_num, "col": col, "msg": "Debe ser número entero"}
+    if valid_ids and client_id not in valid_ids:
+        return {"line": line_num, "col": col, "msg": "Cliente no existe en catálogo"}
+    return None
+
+
+def check_optional_short_name(
+    row: Dict[str, Any],
+    col: str,
+    short_name_to_id: Dict[str, int],
+    line_num: int,
+) -> Optional[Dict[str, Any]]:
+    """Si la columna viene vacía no se valida; si trae valor debe ser short_name existente en catálogo. Solo string, no ID."""
+    val = (row.get(col) or "").strip()
+    if not val:
+        return None
+    key = val.upper()
+    if key not in short_name_to_id:
+        return {
+            "line": line_num,
+            "col": col,
+            "msg": "Cliente no encontrado (short_name no existe en catálogo)",
+        }
+    return None
+
+
 def check_in_set(
     row: Dict[str, Any],
     col: str,
diff --git a/backend/api/v1/modules/a76/layouts_csv/pedmientos/common/fk_loader.py b/backend/api/v1/modules/a76/layouts_csv/pedmientos/common/fk_loader.py
index 411d289f..cb55fefe 100644
--- a/backend/api/v1/modules/a76/layouts_csv/pedmientos/common/fk_loader.py
+++ b/backend/api/v1/modules/a76/layouts_csv/pedmientos/common/fk_loader.py
@@ -1,25 +1,63 @@
 """
 Carga de conjuntos FK para validación de import CSV de pedimentos.
+Clarion: clave+regimen+tipo (CodePedimentoRegimen), aduana (CustomsSection), patente (CustomsBroker.license), existing_pedimento_keys.
+Legacy: valid_client_ids, valid_regimes, valid_pedimento_codes. Cliente por short_name: short_name_to_id.
 """
-from typing import Set, Tuple
+from typing import Dict, Set, Tuple
 
 from sqlalchemy.orm import Session
 
 
+def _pedimento_key(year: str, customs_office: str, license_val: str, pedimento_number: str) -> str:
+    """Clave única para identificar un pedimento (tenant/company se filtra en query)."""
+    return f"{year}|{customs_office}|{license_val}|{pedimento_number}"
+
+
 def load_pedimentos_fk_sets(
     session: Session, tenant_id: int, company_id: int
-) -> Tuple[Set[int], Set[str], Set[str]]:
+) -> Tuple[
+    Set[int],
+    Set[str],
+    Set[str],
+    Set[Tuple[str, str, str]],
+    Set[str],
+    Set[str],
+    Set[str],
+    Set[str],
+    Dict[str, int],
+]:
     """
-    Carga valid_client_ids (ClientProvider.id), valid_regimes (RegimenPedimento.code),
-    valid_pedimento_codes (PedimentoCode.code).
+    Carga todos los conjuntos FK para validación Clarion y legacy.
+    Returns:
+        valid_client_ids,
+        valid_regimes,
+        valid_pedimento_codes,
+        valid_clave_regimen_tipo,  # (pedimento_code, regimen_code, type_code)
+        valid_aduana_seccion,      # customs_code
+        existing_pedimento_keys,   # key strings para actualizar
+        valid_anexo22_claves,      # stub vacío hasta tener catálogo
+        valid_patentes,            # CustomsBroker.license (tenant/company)
+        short_name_to_id,          # short_name normalizado (upper) -> client id (primera aparición gana)
     """
     from api.v1.modules.a76.clients_and_providers.models import ClientProvider
-    from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento
+    from api.v1.modules.a76.customs_brokers.models import CustomsBroker
+    from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos
+    from api.v1.modules.public.reference_data.code_pedimento_regimens.models import (
+        CodePedimentoRegimen,
+    )
+    from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection
     from api.v1.modules.public.reference_data.pedimento_codes.models import PedimentoCode
+    from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento
 
     valid_client_ids: Set[int] = set()
     valid_regimes: Set[str] = set()
     valid_pedimento_codes: Set[str] = set()
+    valid_clave_regimen_tipo: Set[Tuple[str, str, str]] = set()
+    valid_aduana_seccion: Set[str] = set()
+    existing_pedimento_keys: Set[str] = set()
+    valid_anexo22_claves: Set[str] = set()
+    valid_patentes: Set[str] = set()
+    short_name_to_id: Dict[str, int] = {}
 
     for cp in (
         session.query(ClientProvider)
@@ -27,12 +65,85 @@ def load_pedimentos_fk_sets(
             ClientProvider.tenant_id == tenant_id,
             ClientProvider.company_id == company_id,
         )
+        .order_by(ClientProvider.id)
         .all()
     ):
         valid_client_ids.add(cp.id)
+        sn = (cp.short_name or "").strip()
+        if sn:
+            key = sn.upper()
+            if key not in short_name_to_id:
+                short_name_to_id[key] = cp.id
+
     for r in session.query(RegimenPedimento).all():
         valid_regimes.add(r.code)
+
     for pc in session.query(PedimentoCode).all():
         valid_pedimento_codes.add(pc.code)
 
-    return valid_client_ids, valid_regimes, valid_pedimento_codes
+    for cpr in session.query(CodePedimentoRegimen).all():
+        # type_code puede ser None; Clarion usa I/E. Normalizar a mayúsculas para coincidir con CSV.
+        t = (cpr.type_code or "").strip().upper()
+        if t:
+            valid_clave_regimen_tipo.add(
+                (
+                    (cpr.pedimento_code or "").strip().upper(),
+                    (cpr.regimen_code or "").strip().upper(),
+                    t,
+                )
+            )
+
+    for cs in session.query(CustomsSection).all():
+        valid_aduana_seccion.add(cs.customs_code.strip())
+
+    for cb in (
+        session.query(CustomsBroker)
+        .filter(
+            CustomsBroker.tenant_id == tenant_id,
+            CustomsBroker.company_id == company_id,
+        )
+        .all()
+    ):
+        if (cb.license or "").strip():
+            valid_patentes.add((cb.license or "").strip())
+
+    for p in (
+        session.query(
+            Pedimentos.year,
+            Pedimentos.customs_office,
+            Pedimentos.license,
+            Pedimentos.pedimento_number,
+        )
+        .filter(
+            Pedimentos.tenant_id == tenant_id,
+            Pedimentos.company_id == company_id,
+        )
+        .all()
+    ):
+        existing_pedimento_keys.add(
+            _pedimento_key(
+                (p.year or "").strip(),
+                (p.customs_office or "").strip(),
+                (p.license or "").strip(),
+                (p.pedimento_number or "").strip(),
+            )
+        )
+
+    return (
+        valid_client_ids,
+        valid_regimes,
+        valid_pedimento_codes,
+        valid_clave_regimen_tipo,
+        valid_aduana_seccion,
+        existing_pedimento_keys,
+        valid_anexo22_claves,
+        valid_patentes,
+        short_name_to_id,
+    )
+
+
+def pedimento_key_from_parsed(
+    year: str, customs_office: str, license_val: str, pedimento_number: str
+) -> str:
+    """Construye la clave para comparar con existing_pedimento_keys."""
+    return _pedimento_key(year, customs_office, license_val, pedimento_number)
diff --git a/backend/api/v1/modules/a76/layouts_csv/pedmientos/common/mappers.py b/backend/api/v1/modules/a76/layouts_csv/pedmientos/common/mappers.py
index 6eed3b87..3975f19a 100644
--- a/backend/api/v1/modules/a76/layouts_csv/pedmientos/common/mappers.py
+++ b/backend/api/v1/modules/a76/layouts_csv/pedmientos/common/mappers.py
@@ -1,9 +1,14 @@
 """
 Mapeo fila CSV → datos para PedimentosCreate.
+Soporta layout Clarion (PEDIMENTO, TIPO_OPERACION, CLAVE_PEDIMENTO, etc.) y legacy (AÑO, ADUANA, PATENTE, NUMERO).
 """
+from datetime import datetime
 from decimal import Decimal, InvalidOperation
 from typing import Dict, Any, Optional
 
+from ..template_config import is_clarion_layout, parse_pedimento_col_a
+from ..common.common_validators import parse_date
+
 
 def parse_decimal(val: Any) -> Optional[Decimal]:
     if val is None or (isinstance(val, str) and not val.strip()):
@@ -14,15 +19,102 @@ def parse_decimal(val: Any) -> Optional[Decimal]:
         return None
 
 
-def row_to_pedimento_data(row_norm: Dict[str, Any]) -> Dict[str, Any]:
-    """Build dict for PedimentosCreate from normalized CSV row."""
+def resolve_client_id_from_short_name(
+    valor: str, short_name_to_id: Dict[str, int]
+) -> Optional[int]:
+    """Resuelve short_name (string) a client_id. Vacío → None. Misma normalización que fk_loader (strip + upper)."""
+    if not valor or not str(valor).strip():
+        return None
+    key = str(valor).strip().upper()
+    return short_name_to_id.get(key)
+
+
+def _row_to_pedimento_data_clarion(
+    row_norm: Dict[str, Any],
+    short_name_to_id: Dict[str, int],
+    date_format_preference: Optional[str] = None,
+) -> Dict[str, Any]:
+    """Desde layout Clarion: Cols 1-3 (AÑO, PATENTE, NUMERO) o Col A (PEDIMENTO); Col H → customs_office; B,C,D,J y fechas. CLIENTE_SHORT_NAME opcional."""
+    año = (row_norm.get("AÑO") or "").strip()
+    patente = (row_norm.get("PATENTE") or "").strip()
+    numero = (row_norm.get("NUMERO") or "").strip()
+    if año and patente and numero:
+        year = año[:2]
+        license_val = patente[:4]
+        pedimento_number = numero[:7]
+    else:
+        ped = (row_norm.get("PEDIMENTO") or "").strip()
+        parsed = parse_pedimento_col_a(ped)
+        if not parsed:
+            raise ValueError("PEDIMENTO inválido o vacío")
+        year, license_val, pedimento_number = parsed
+    aduana = (row_norm.get("ADUANA_SECCION_CRUCE") or "").strip()[:3]
+    clave = (row_norm.get("CLAVE_PEDIMENTO") or "").strip()[:2]
+    regime = (row_norm.get("REGIMEN") or "").strip()[:3]
+    client_short_name = (row_norm.get("CLIENTE_SHORT_NAME") or "").strip()
+    client_id = resolve_client_id_from_short_name(client_short_name, short_name_to_id)
+
+    data: Dict[str, Any] = {
+        "year": year,
+        "customs_office": aduana,
+        "license": license_val,
+        "pedimento_number": pedimento_number,
+        "pedimento_code": clave,
+        "regime": regime,
+    }
+    if client_id is not None:
+        data["client_id"] = client_id
+
+    tipo = (row_norm.get("TIPO_OPERACION") or "").strip().upper()
+    if tipo == "I":
+        data["operation_type"] = "imp"
+    elif tipo == "E":
+        data["operation_type"] = "exp"
+
+    ind_con = (row_norm.get("INDIVIDUAL_CONSOLIDADO") or "").strip().upper() or "CON"
+    data["pedimento_type"] = "normal" if ind_con == "IND" else "consolidated"
+
+    status = (row_norm.get("ESTATUS") or "").strip()
+    if status:
+        data["status"] = status[:30]
+
+    data["usd_value"] = parse_decimal(row_norm.get("VALOR_USD"))
+    data["paid_price"] = parse_decimal(row_norm.get("PRECIO_PAGADO"))
+    data["gross_weight"] = parse_decimal(row_norm.get("PESO_BRUTO"))
+    data["exchange_rate"] = parse_decimal(row_norm.get("TIPO_CAMBIO"))
+    obs = (row_norm.get("OBSERVACIONES") or "").strip()
+    if obs:
+        data["observations"] = obs
+
+    # Fechas E, F, G → pedimento_dates
+    start_str = (row_norm.get("FECHA_INICIO") or "").strip()
+    end_str = (row_norm.get("FECHA_FINAL") or "").strip()
+    payment_str = (row_norm.get("FECHA_PAGO") or "").strip()
+    base = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
+    start_dt = parse_date(start_str, date_format_preference) if start_str else base
+    end_dt = parse_date(end_str, date_format_preference) if end_str else base
+    payment_dt = parse_date(payment_str, date_format_preference) if payment_str else base
+    if start_dt and end_dt:
+        data["pedimento_dates"] = {
+            "entry_date": start_dt,
+            "end_date": end_dt,
+            "start_date": start_dt,
+            "payment_date": payment_dt,
+        }
+    return data
+
+
+def _row_to_pedimento_data_legacy(
+    row_norm: Dict[str, Any], short_name_to_id: Dict[str, int]
+) -> Dict[str, Any]:
+    """Layout legacy: AÑO, ADUANA, PATENTE, NUMERO, CLIENTE_SHORT_NAME, CODIGO_PEDIMENTO/CLAVE_PEDIMENTO, REGIMEN."""
     year = (row_norm.get("AÑO") or "").strip()[:2]
-    customs_office = (row_norm.get("ADUANA") or "").strip()[:3]
+    customs_office = (row_norm.get("ADUANA_SECCION_CRUCE") or row_norm.get("ADUANA") or "").strip()[:3]
     license_val = (row_norm.get("PATENTE") or "").strip()[:4]
     pedimento_number = (row_norm.get("NUMERO") or "").strip()[:7]
-    client_id_str = (row_norm.get("CLIENTE_ID") or "").strip()
-    client_id = int(client_id_str) if client_id_str else None
-    pedimento_code = (row_norm.get("CODIGO_PEDIMENTO") or "").strip()[:2]
+    client_short_name = (row_norm.get("CLIENTE_SHORT_NAME") or "").strip()
+    client_id = resolve_client_id_from_short_name(client_short_name, short_name_to_id)
+    pedimento_code = (row_norm.get("CLAVE_PEDIMENTO") or row_norm.get("CODIGO_PEDIMENTO") or "").strip()[:2]
     regime = (row_norm.get("REGIMEN") or "").strip()[:3]
 
     data = {
@@ -38,22 +130,58 @@ def row_to_pedimento_data(row_norm: Dict[str, Any]) -> Dict[str, Any]:
     op = (row_norm.get("TIPO_OPERACION") or "").strip().lower()
     if op in ("imp", "exp"):
         data["operation_type"] = op
-
     ptype = (row_norm.get("TIPO_PEDIMENTO") or "").strip().lower()
     if ptype in ("normal", "consolidated", "complementary", "automobile"):
         data["pedimento_type"] = ptype
-
     status = (row_norm.get("ESTATUS") or "").strip()
     if status:
         data["status"] = status[:30]
-
     data["usd_value"] = parse_decimal(row_norm.get("VALOR_USD"))
     data["paid_price"] = parse_decimal(row_norm.get("PRECIO_PAGADO"))
     data["gross_weight"] = parse_decimal(row_norm.get("PESO_BRUTO"))
     data["exchange_rate"] = parse_decimal(row_norm.get("TIPO_CAMBIO"))
-
     obs = (row_norm.get("OBSERVACIONES") or "").strip()
     if obs:
         data["observations"] = obs
-
     return data
+
+
+def row_to_pedimento_data(
+    row_norm: Dict[str, Any],
+    short_name_to_id: Dict[str, int],
+    date_format_preference: Optional[str] = None,
+) -> Dict[str, Any]:
+    """
+    Build dict for PedimentosCreate from normalized CSV row.
+    Usa layout Clarion si existe PEDIMENTO; si no, layout legacy. Cliente por short_name → client_id.
+    """
+    if is_clarion_layout(row_norm):
+        return _row_to_pedimento_data_clarion(
+            row_norm, short_name_to_id, date_format_preference
+        )
+    return _row_to_pedimento_data_legacy(row_norm, short_name_to_id)
+
+
+def row_to_pedimento_data_merge_existing(
+    row_norm: Dict[str, Any],
+    existing: Dict[str, Any],
+    short_name_to_id: Dict[str, int],
+    date_format_preference: Optional[str] = None,
+) -> Dict[str, Any]:
+    """
+    Para modo actualizar: campos vacíos en la fila se rellenan con el pedimento existente.
+    """
+    if is_clarion_layout(row_norm):
+        new_data = _row_to_pedimento_data_clarion(
+            row_norm, short_name_to_id, date_format_preference
+        )
+    else:
+        new_data = _row_to_pedimento_data_legacy(row_norm, short_name_to_id)
+
+    for key, val in new_data.items():
+        if val is None or (isinstance(val, str) and not val.strip()):
+            if key in existing and existing[key] is not None:
+                new_data[key] = existing[key]
+    if not new_data.get("pedimento_dates") and existing.get("pedimento_dates"):
+        new_data["pedimento_dates"] = existing["pedimento_dates"]
+    return new_data
diff --git a/backend/api/v1/modules/a76/layouts_csv/pedmientos/routes.py b/backend/api/v1/modules/a76/layouts_csv/pedmientos/routes.py
index eea706fb..deb0b9d9 100644
--- a/backend/api/v1/modules/a76/layouts_csv/pedmientos/routes.py
+++ b/backend/api/v1/modules/a76/layouts_csv/pedmientos/routes.py
@@ -10,7 +10,7 @@ from uuid import uuid4
 
 from fastapi import APIRouter, File, HTTPException, Query, UploadFile, Depends
 from sqlalchemy.orm import Session
-from typing import Dict, Any
+from typing import Dict, Any, Optional
 
 from core.celery_app import celery_app
 from core.database import get_core_db
@@ -21,10 +21,10 @@ from .schemas import ImportJobResponse
 from .tasks import (
     scan_file,
     insert_valid_rows,
-    PED_IMPORT_FILE_PREFIX,
-    PED_IMPORT_META_PREFIX,
+    JOB_TYPE as PED_JOB_TYPE,
     PED_IMPORT_REDIS_TTL,
 )
+from ..common import storage as common_storage
 
 router = APIRouter()
 logger = logging.getLogger(__name__)
@@ -40,6 +40,8 @@ def _get_redis():
 async def upload_import_file(
     file: UploadFile = File(...),
     company_id: int = Query(..., description="Company ID"),
+    actualizar: bool = Query(False, description="Modo actualizar: validación parcial si el pedimento existe"),
+    dateFormat: Optional[str] = Query(None, description="Formato de fecha: dd/mm/yyyy, mm/dd/yyyy, yyyy-mm-dd"),
     db: Session = Depends(get_core_db),
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
@@ -58,22 +60,26 @@ async def upload_import_file(
     job_id = str(uuid4())
     contents = await file.read()
 
-    meta_data = {
+    file_key, meta_key, _ = common_storage.storage_keys(PED_JOB_TYPE, job_id)
+    meta_data: Dict[str, Any] = {
         "tenant_id": tenant_id,
         "company_id": company_id,
         "user_id": current_user.get("id"),
         "template_id": "pedimentos",
+        "actualizar": actualizar,
     }
+    if dateFormat:
+        meta_data["dateFormat"] = dateFormat
 
     try:
         r = _get_redis()
         r.set(
-            f"{PED_IMPORT_FILE_PREFIX}{job_id}",
+            file_key,
             base64.b64encode(contents),
             ex=PED_IMPORT_REDIS_TTL,
         )
         r.set(
-            f"{PED_IMPORT_META_PREFIX}{job_id}",
+            meta_key,
             json.dumps(meta_data).encode("utf-8"),
             ex=PED_IMPORT_REDIS_TTL,
         )
@@ -84,9 +90,11 @@ async def upload_import_file(
     try:
         upload_dir = layout_path("imports", "temp")
         os.makedirs(upload_dir, exist_ok=True)
-        with open(os.path.join(upload_dir, f"ped_{job_id}.csv"), "wb") as f:
+        csv_path = common_storage.file_path_for_job(PED_JOB_TYPE, job_id)
+        with open(csv_path, "wb") as f:
             f.write(contents)
-        with open(os.path.join(upload_dir, f"ped_{job_id}.meta.json"), "w") as f:
+        meta_path = csv_path.replace(".csv", ".meta.json")
+        with open(meta_path, "w") as f:
             json.dump(meta_data, f)
     except Exception as e:
         logger.warning(f"Pedimentos import: local file save failed: {e}")
diff --git a/backend/api/v1/modules/a76/layouts_csv/pedmientos/tasks.py b/backend/api/v1/modules/a76/layouts_csv/pedmientos/tasks.py
index 96da8f38..18ddb766 100644
--- a/backend/api/v1/modules/a76/layouts_csv/pedmientos/tasks.py
+++ b/backend/api/v1/modules/a76/layouts_csv/pedmientos/tasks.py
@@ -17,10 +17,15 @@ from ..common import normalize as common_normalize
 from ..common import meta as common_meta
 from ..common import responses as common_responses
 from ..common import csv_reader as common_csv_reader
-from .template_config import row_from_template
+from .template_config import (
+    row_from_template,
+    is_clarion_layout,
+    parse_pedimento_col_a,
+    detect_headers_or_data,
+)
 from .validators import validate_row_pedimento
-from .common.fk_loader import load_pedimentos_fk_sets
-from .common.mappers import row_to_pedimento_data
+from .common.mappers import row_to_pedimento_data, row_to_pedimento_data_merge_existing
+from .common.fk_loader import load_pedimentos_fk_sets, pedimento_key_from_parsed
 
 logger = logging.getLogger(__name__)
 
@@ -42,12 +47,19 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str,
     file_path = common_storage.ensure_file_from_redis(JOB_TYPE, job_id, "Pedimentos import")
     if not file_path:
         return {"status": "failed", "error": "Archivo no encontrado (expirado o no subido). Sube de nuevo."}
+    if os.path.getsize(file_path) == 0:
+        return {"status": "failed", "error": "El archivo está vacío. Verifica que el CSV tenga contenido."}
     common_storage.ensure_meta_from_redis(JOB_TYPE, job_id, file_path, "Pedimentos import")
 
     error_path = common_storage.error_path_for_job(JOB_TYPE, job_id)
 
+    fieldnames, has_header = detect_headers_or_data(
+        file_path,
+        common_normalize.normalize_header,
+        parse_pedimento_col_a,
+    )
     try:
-        total_rows = common_csv_reader.count_csv_rows(file_path)
+        total_rows = common_csv_reader.count_csv_rows(file_path, has_header=has_header)
     except Exception as e:
         return {"status": "failed", "error": str(e)}
 
@@ -56,11 +68,23 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str,
     except ValueError as e:
         return {"status": "failed", "error": str(e)}
 
+    meta = common_meta.load_meta(file_path) or {}
+    actualizar = meta.get("actualizar", False)
+    date_format_preference = meta.get("dateFormat") or meta.get("date_format")
+
     try:
         with CoreSessionLocal() as session:
-            valid_client_ids, valid_regimes, valid_pedimento_codes = load_pedimentos_fk_sets(
-                session, tenant_id, company_id
-            )
+            (
+                valid_client_ids,
+                valid_regimes,
+                valid_pedimento_codes,
+                valid_clave_regimen_tipo,
+                valid_aduana_seccion,
+                existing_pedimento_keys,
+                valid_anexo22_claves,
+                valid_patentes,
+                short_name_to_id,
+            ) = load_pedimentos_fk_sets(session, tenant_id, company_id)
     except Exception as e:
         logger.error("Pedimentos import: failed to load FK sets: %s", e)
         return {"status": "failed", "error": "No se pudo cargar catálogos"}
@@ -72,13 +96,25 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str,
 
     try:
         with open(error_path, "w", encoding="utf-8") as f_err:
-            for i, row in common_csv_reader.iter_csv_rows(file_path):
+            for i, row in common_csv_reader.iter_csv_rows(file_path, fieldnames=fieldnames):
                 if progress_callback and i % 500 == 0:
                     progress_callback(i, total_rows, error_count)
 
                 row_norm = _norm_row(row)
                 err = validate_row_pedimento(
-                    row_norm, i, valid_client_ids, valid_regimes, valid_pedimento_codes
+                    row_norm,
+                    i,
+                    short_name_to_id,
+                    valid_regimes,
+                    valid_pedimento_codes,
+                    valid_clave_regimen_tipo,
+                    valid_aduana_seccion,
+                    existing_pedimento_keys,
+                    valid_anexo22_claves,
+                    valid_patentes,
+                    actualizar=actualizar,
+                    raw_row=row,
+                    date_format_preference=date_format_preference,
                 )
                 if err:
                     error_count += 1
@@ -118,8 +154,7 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
         if not os.path.exists(alt_path):
             return {"status": "failed", "error": "Archivo no encontrado (expirado). Sube y confirma de nuevo."}
         file_path = alt_path
-    else:
-        common_storage.ensure_meta_from_redis(JOB_TYPE, job_id, file_path, "Pedimentos import")
+    common_storage.ensure_meta_from_redis(JOB_TYPE, job_id, file_path, "Pedimentos import")
 
     error_path = common_storage.error_path_for_job(JOB_TYPE, job_id)
     error_lines = common_storage.get_error_lines(JOB_TYPE, job_id, error_path)
@@ -129,35 +164,90 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
     except ValueError as e:
         return {"status": "failed", "error": str(e)}
 
+    meta = common_meta.load_meta(file_path) or {}
+    actualizar = meta.get("actualizar", False)
+    date_format_preference = meta.get("dateFormat") or meta.get("date_format")
+
     try:
         with CoreSessionLocal() as session:
-            valid_client_ids, valid_regimes, valid_pedimento_codes = load_pedimentos_fk_sets(
-                session, tenant_id, company_id
-            )
+            (
+                valid_client_ids,
+                valid_regimes,
+                valid_pedimento_codes,
+                valid_clave_regimen_tipo,
+                valid_aduana_seccion,
+                existing_pedimento_keys,
+                valid_anexo22_claves,
+                valid_patentes,
+                short_name_to_id,
+            ) = load_pedimentos_fk_sets(session, tenant_id, company_id)
     except Exception as e:
         logger.error("Pedimentos import: failed to load FK sets: %s", e)
         return {"status": "failed", "error": "No se pudo cargar catálogos"}
 
-    from api.v1.modules.a76.pedmientos.dtos.pedimentos import PedimentosCreate
+    from api.v1.modules.a76.pedmientos.dtos.pedimentos import (
+        PedimentosCreate,
+        PedimentosUpdate,
+    )
     from api.v1.modules.a76.pedmientos.dtos.pedimento_dates import PedimentoDatesCreate
     from api.v1.modules.a76.pedmientos.services.pedimentos import PedimentosService
+    from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos
 
     inserted_count = 0
+    updated_count = 0
     skipped_invalid = 0
     skipped_missing_fk = 0
     skipped_duplicate = 0
     skipped_details: List[Dict[str, Any]] = []
     meta_path = common_meta.get_meta_path(file_path)
 
+    fieldnames_commit, _ = detect_headers_or_data(
+        file_path,
+        common_normalize.normalize_header,
+        parse_pedimento_col_a,
+    )
+
+    def _key_from_row(r: Dict[str, Any]) -> Optional[str]:
+        if is_clarion_layout(r):
+            año = (r.get("AÑO") or "").strip()
+            patente = (r.get("PATENTE") or "").strip()
+            numero = (r.get("NUMERO") or "").strip()
+            if año and patente and numero:
+                adu = (r.get("ADUANA_SECCION_CRUCE") or "").strip()[:3]
+                return pedimento_key_from_parsed(año[:2], adu, patente[:4], numero[:7])
+            parsed = parse_pedimento_col_a((r.get("PEDIMENTO") or "").strip())
+            if not parsed:
+                return None
+            y, lic, num = parsed
+            adu = (r.get("ADUANA_SECCION_CRUCE") or "").strip()[:3]
+            return pedimento_key_from_parsed(y, adu, lic, num)
+        y = (r.get("AÑO") or "").strip()[:2]
+        adu = (r.get("ADUANA_SECCION_CRUCE") or r.get("ADUANA") or "").strip()[:3]
+        lic = (r.get("PATENTE") or "").strip()[:4]
+        num = (r.get("NUMERO") or "").strip()[:7]
+        return pedimento_key_from_parsed(y, adu, lic, num)
+
     try:
         with CoreSessionLocal() as session:
-            for i, row in common_csv_reader.iter_csv_rows(file_path):
+            for i, row in common_csv_reader.iter_csv_rows(file_path, fieldnames=fieldnames_commit):
                 if i in error_lines:
                     continue
 
                 row_norm = _norm_row(row)
                 err = validate_row_pedimento(
-                    row_norm, i, valid_client_ids, valid_regimes, valid_pedimento_codes
+                    row_norm,
+                    i,
+                    short_name_to_id,
+                    valid_regimes,
+                    valid_pedimento_codes,
+                    valid_clave_regimen_tipo,
+                    valid_aduana_seccion,
+                    existing_pedimento_keys,
+                    valid_anexo22_claves,
+                    valid_patentes,
+                    actualizar=actualizar,
+                    raw_row=row,
+                    date_format_preference=date_format_preference,
                 )
                 if err:
                     skipped_invalid += 1
@@ -167,16 +257,95 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
                     })
                     continue
 
+                key = _key_from_row(row_norm)
                 try:
-                    data = row_to_pedimento_data(row_norm)
-                    if "pedimento_dates" not in data or data.get("pedimento_dates") is None:
-                        data["pedimento_dates"] = PedimentoDatesCreate(
-                            entry_date=datetime.now(),
-                            end_date=datetime.now(),
+                    # Si el pedimento ya existe: actualizar (merge) o reemplazar (crear). Si no existe: crear.
+                    if key and key in existing_pedimento_keys:
+                        existing = (
+                            session.query(Pedimentos)
+                            .filter(
+                                Pedimentos.tenant_id == tenant_id,
+                                Pedimentos.company_id == company_id,
+                                Pedimentos.year == key.split("|")[0],
+                                Pedimentos.customs_office == key.split("|")[1],
+                                Pedimentos.license == key.split("|")[2],
+                                Pedimentos.pedimento_number == key.split("|")[3],
+                            )
+                            .first()
                         )
-                    create_data = PedimentosCreate(**data)
-                    PedimentosService.create(session, create_data, tenant_id, company_id)
-                    inserted_count += 1
+                        if existing:
+                            if actualizar:
+                                # Modo actualizar: merge con existente
+                                existing_dict = {
+                                    "year": existing.year,
+                                    "customs_office": existing.customs_office,
+                                    "license": existing.license,
+                                    "pedimento_number": existing.pedimento_number,
+                                    "client_id": existing.client_id,
+                                    "pedimento_code": existing.pedimento_code,
+                                    "regime": existing.regime,
+                                    "operation_type": existing.operation_type,
+                                    "pedimento_type": existing.pedimento_type,
+                                    "status": existing.status,
+                                    "usd_value": existing.usd_value,
+                                    "paid_price": existing.paid_price,
+                                    "gross_weight": existing.gross_weight,
+                                    "exchange_rate": existing.exchange_rate,
+                                    "observations": existing.observations,
+                                }
+                                if existing.pedimento_dates:
+                                    existing_dict["pedimento_dates"] = {
+                                        "entry_date": existing.pedimento_dates.entry_date,
+                                        "end_date": existing.pedimento_dates.end_date,
+                                        "start_date": getattr(
+                                            existing.pedimento_dates, "start_date", existing.pedimento_dates.entry_date
+                                        ),
+                                        "payment_date": getattr(
+                                            existing.pedimento_dates, "payment_date", existing.pedimento_dates.entry_date
+                                        ),
+                                    }
+                                data = row_to_pedimento_data_merge_existing(
+                                    row_norm,
+                                    existing_dict,
+                                    short_name_to_id,
+                                    date_format_preference,
+                                )
+                            else:
+                                # Modo crear: reemplazar con datos del CSV
+                                data = row_to_pedimento_data(
+                                    row_norm, short_name_to_id, date_format_preference
+                                )
+                                if "pedimento_dates" not in data or data.get("pedimento_dates") is None:
+                                    data["pedimento_dates"] = PedimentoDatesCreate(
+                                        entry_date=datetime.now(),
+                                        end_date=datetime.now(),
+                                    )
+                            update_data = {k: v for k, v in data.items() if k != "pedimento_dates"}
+                            if data.get("pedimento_dates"):
+                                update_data["pedimento_dates"] = PedimentoDatesCreate(
+                                    **data["pedimento_dates"]
+                                )
+                            update_schema = PedimentosUpdate(**update_data)
+                            PedimentosService.update(
+                                session, existing.id, tenant_id, update_schema, company_id
+                            )
+                            updated_count += 1
+                        else:
+                            skipped_invalid += 1
+                            skipped_details.append({"line": i, "reason": "Pedimento no encontrado"})
+                    else:
+                        # No existe: crear (ambos modos)
+                        data = row_to_pedimento_data(
+                            row_norm, short_name_to_id, date_format_preference
+                        )
+                        if "pedimento_dates" not in data or data.get("pedimento_dates") is None:
+                            data["pedimento_dates"] = PedimentoDatesCreate(
+                                entry_date=datetime.now(),
+                                end_date=datetime.now(),
+                            )
+                        create_data = PedimentosCreate(**data)
+                        PedimentosService.create(session, create_data, tenant_id, company_id)
+                        inserted_count += 1
                 except ValueError as ve:
                     if "Ya existe" in str(ve) or "duplicate" in str(ve).lower():
                         skipped_duplicate += 1
@@ -201,7 +370,12 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
     )
 
     total_skipped = skipped_invalid + skipped_missing_fk + skipped_duplicate
-    if inserted_count == 0 and total_skipped > 0:
+    if inserted_count == 0 and updated_count == 0 and total_skipped > 0:
+        reasons = "; ".join(
+            f"Línea {d.get('line', '?')}: {d.get('reason', '')}" for d in skipped_details[:5]
+        )
+        if len(skipped_details) > 5:
+            reasons += f" (+{len(skipped_details) - 5} más)"
         return {
             "status": "warning",
             "inserted": 0,
@@ -210,9 +384,9 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
             "skipped_missing_fk": skipped_missing_fk,
             "skipped_duplicate": skipped_duplicate,
             "skipped_details": skipped_details,
-            "message": f"No se insertaron registros. {total_skipped} rechazados.",
+            "message": f"No se insertaron registros. {total_skipped} rechazados. Motivos: {reasons}",
         }
-    if inserted_count == 0:
+    if inserted_count == 0 and updated_count == 0:
         return {
             "status": "failed",
             "error": "No hay registros válidos en el archivo CSV",
@@ -226,7 +400,7 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
     return {
         "status": "finished",
         "inserted": inserted_count,
-        "updated": 0,
+        "updated": updated_count,
         "skipped_invalid": skipped_invalid,
         "skipped_missing_fk": skipped_missing_fk,
         "skipped_duplicate": skipped_duplicate,
diff --git a/backend/api/v1/modules/a76/layouts_csv/pedmientos/template_config.py b/backend/api/v1/modules/a76/layouts_csv/pedmientos/template_config.py
index 06fea167..ffea72b9 100644
--- a/backend/api/v1/modules/a76/layouts_csv/pedmientos/template_config.py
+++ b/backend/api/v1/modules/a76/layouts_csv/pedmientos/template_config.py
@@ -1,22 +1,97 @@
 """
 Configuración de plantilla CSV para Pedimentos (EstructuraCatPedimentos.xls).
-Solo se leen columnas definidas aquí; el resto se ignora.
+Layout Clarion: Cols 1-3 = AÑO (2), PATENTE (4), NUMERO (7); luego TIPO, CLAVE PEDIMENTO, REGIMEN,
+fechas, ADUANA Y SECCION CRUCE, etc. Compatibilidad: PEDIMENTO (##-####-#######) y layout legacy.
 """
+import csv
+import io
+from typing import Dict, List, Any, Optional, Tuple
 
-from typing import Dict, List, Any, Optional
+# Longitudes para validación (sin afectar modelos)
+AÑO_LEN = 2
+PATENTE_LEN = 4
+NUMERO_LEN = 7
 
+# Columnas Clarion (canónicos) + aliases exactos de la plantilla y legacy
 TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     "pedimentos": [
+        # Cols 1-3: AÑO, PATENTE, NUMERO (reemplazan Col A única PEDIMENTO)
         {"canonical": "AÑO", "aliases": ["YEAR", "ANIO"]},
-        {"canonical": "ADUANA", "aliases": ["CUSTOMS_OFFICE", "CUSTOMS OFFICE"]},
         {"canonical": "PATENTE", "aliases": ["LICENCIA", "LICENSE", "LIC"]},
-        {"canonical": "NUMERO", "aliases": ["PEDIMENTO_NUMBER", "PEDIMENTO NUMBER", "NUMERO PEDIMENTO"]},
-        {"canonical": "CLIENTE_ID", "aliases": ["CLIENT_ID", "CLIENTE", "ID CLIENTE"]},
-        {"canonical": "TIPO_OPERACION", "aliases": ["OPERATION_TYPE", "OPERACION"]},
-        {"canonical": "TIPO_PEDIMENTO", "aliases": ["PEDIMENTO_TYPE", "TIPO"]},
-        {"canonical": "CODIGO_PEDIMENTO", "aliases": ["PEDIMENTO_CODE", "CODIGO", "CLAVE PEDIMENTO"]},
-        {"canonical": "REGIMEN", "aliases": ["REGIME"]},
-        {"canonical": "ESTATUS", "aliases": ["STATUS", "ESTADO"]},
+        {"canonical": "NUMERO", "aliases": ["PEDIMENTO_NUMBER", "PEDIMENTO NUMBER"]},
+        # Col B
+        {"canonical": "TIPO_OPERACION", "aliases": ["TIPO MOV(I=Importación,E=Exportación)", "TIPO MOV(I=Impotación,E=Expotación)", "TIPO MOV", "TIPO", "OPERATION_TYPE", "OPERACION"]},
+        # Col C
+        {"canonical": "CLAVE_PEDIMENTO", "aliases": ["CLAVE PEDIMENTO", "CODIGO_PEDIMENTO", "PEDIMENTO_CODE", "CODIGO"]},
+        # Col D
+        {"canonical": "REGIMEN", "aliases": ["REGIME", "CLAVE RÉGIMEN"]},
+        # Col E, F, G
+        {"canonical": "FECHA_INICIO", "aliases": ["FECHA INICIO", "FECHA INICIAL"]},
+        {"canonical": "FECHA_FINAL", "aliases": ["FECHA FINAL", "FECHA FIN"]},
+        {"canonical": "FECHA_PAGO", "aliases": ["FECHA DE PAGO", "FECHA PAGO"]},
+        # Col H
+        {"canonical": "ADUANA_SECCION_CRUCE", "aliases": ["ADUANA Y SECCION DE CRUCE", "ADUANA Y SECCION CRUCE", "ADUANA", "CUSTOMS_OFFICE", "CUSTOMS OFFICE"]},
+        # Col I
+        {"canonical": "ACUSE_ELECTRONICO", "aliases": ["ACUSE ELECTRONICO", "ACUSE ELECTRÓNICO"]},
+        # Col J
+        {"canonical": "INDIVIDUAL_CONSOLIDADO", "aliases": ["INDIVIDUAL o CONSOLIDADO (IND,CON)", "INDIVIDUAL o CONSOLIDADO", "IND/CON", "INDIVIDUAL CONSOLIDADO"]},
+        # Col K, L, M
+        {"canonical": "MET_TRANSP_ENTRADA", "aliases": ["MET TRANS ENTRADA", "METODO TRANSP ENTRADA"]},
+        {"canonical": "MET_TRANSP_ARRIVO", "aliases": ["MET TRANS ARRIVO", "METODO TRANSP ARRIVO"]},
+        {"canonical": "MET_TRANSP_SALIDA", "aliases": ["MET TRANS SALIDA", "METODO TRANSP SALIDA"]},
+        # Col N (desfase) y resto de columnas de la plantilla
+        {"canonical": "IEPS", "aliases": []},
+        {"canonical": "DTA", "aliases": []},
+        {"canonical": "CNT", "aliases": []},
+        {"canonical": "PREVALIDACION", "aliases": []},
+        {"canonical": "MONTO_TIGIE", "aliases": ["MONTO TIGIE"]},
+        {"canonical": "PAGO_IMPUESTO", "aliases": ["PAGO IMPUESTO? (S/N)"]},
+        {"canonical": "ES_MIXTO", "aliases": ["ES MIXTO (SI/NO)"]},
+        {"canonical": "OBS_RECTIFICA", "aliases": ["OBS RECTIFICA"]},
+        {"canonical": "OPCION_DESTINO", "aliases": ["OPCION DESTINO(Interior del Pais/Región Fronteriza/Franja Fronteriza)", "OPCION DESTINO"]},
+        {"canonical": "VALOR_IVA", "aliases": ["VALOR IVA"]},
+        {"canonical": "VALOR_ME", "aliases": ["VALOR ME"]},
+        {"canonical": "VALOR_ADUANAS", "aliases": ["VALOR ADUANAS"]},
+        {"canonical": "FLETE", "aliases": []},
+        {"canonical": "VALOR_SEGUROS", "aliases": ["VALOR SEGUROS"]},
+        {"canonical": "SEGUROS", "aliases": []},
+        {"canonical": "EMBALAJES", "aliases": []},
+        {"canonical": "OTROS_INCREMENTABLES", "aliases": ["OTROS INCREMENTABLES"]},
+        {"canonical": "ESTATUS", "aliases": ["ESTATUS (ABIERTO/CERRADO)", "STATUS", "ESTADO"]},
+        {"canonical": "PERSONA_REV", "aliases": ["PERSONA REV"]},
+        {"canonical": "FECHA_CIERRE", "aliases": ["FECHA CIERRE"]},
+        {"canonical": "FECHA_REVISION", "aliases": ["FECHA REVISION"]},
+        {"canonical": "FECHA_AUTORIZACION", "aliases": ["FECHA AUTORIZACION"]},
+        {"canonical": "FECHA_RECIBIDO", "aliases": ["FECHA RECIBIDO"]},
+        {"canonical": "REPRESENTANTE_AA", "aliases": ["REPRESENTANTE AA"]},
+        {"canonical": "CLAVE_DEST_ORIGEN", "aliases": ["CLAVE DEST ORIGEN"]},
+        {"canonical": "FECHA_ENTRADA_RECINTO", "aliases": ["FECHA ENTRADA RECINTO"]},
+        {"canonical": "FECHA_EXTRACCION_RECINTO", "aliases": ["FECHA EXTRACCION RECINTO"]},
+        {"canonical": "ERRORES", "aliases": []},
+        {"canonical": "FORMA_PAGO_DTA", "aliases": ["FORMA PAGO DTA"]},
+        {"canonical": "FORMA_PAGO_IGI", "aliases": ["FORMA PAGO IGI"]},
+        {"canonical": "FORMA_PAGO_PREVAL", "aliases": ["FORMA PAGO PREVAL"]},
+        {"canonical": "FORMA_PAGO_IVA", "aliases": ["FORMA PAGO IVA"]},
+        {"canonical": "RECARGOS", "aliases": []},
+        {"canonical": "MULTAS", "aliases": []},
+        {"canonical": "IVA_DE_PREV", "aliases": ["IVA DE PREV"]},
+        {"canonical": "CUOTAS_COMPENSATORIAS", "aliases": ["CUOTAS CONPENSATORIAS"]},
+        {"canonical": "IDENTIFICADORES", "aliases": []},
+        {"canonical": "IEPS_2", "aliases": ["IEPS 2"]},
+        {"canonical": "FORMA_PAGO_IEPS_2", "aliases": ["FORMA DE PAGO IEPS 2"]},
+        {"canonical": "DTA_2", "aliases": ["DTA 2"]},
+        {"canonical": "FORMA_PAGO_DTA_2", "aliases": ["FORMA DE PAGO DTA 2"]},
+        {"canonical": "IVA_2", "aliases": ["IVA 2"]},
+        {"canonical": "FORMA_PAGO_IVA_2", "aliases": ["FORMA DE PAGO IVA 2"]},
+        {"canonical": "IGI_2", "aliases": ["IGI 2"]},
+        {"canonical": "FORMA_PAGO_IGI_2", "aliases": ["FORMA DE PAGO IGI 2"]},
+        {"canonical": "FORMA_PAGO_PREVALIDACION_2", "aliases": ["FORMA DE PAGO PREVALIDACION 2"]},
+        {"canonical": "CNT_2", "aliases": ["CNT 2"]},
+        {"canonical": "FORMA_PAGO_CNT_2", "aliases": ["FORMA DE PAGO CNT 2"]},
+        # Legacy / compatibilidad (PEDIMENTO una columna ##-####-#######, y otros)
+        {"canonical": "PEDIMENTO", "aliases": ["NUMERO DE PEDIMENTO (##-####-######)", "NUMERO DE PEDIMENTO", "NUMERO PEDIMENTO", "PED"]},
+        {"canonical": "CLIENTE_SHORT_NAME", "aliases": ["CLIENTE", "SHORT_NAME", "CLIENTE_ID", "CLIENT_ID", "ID CLIENTE"]},
+        {"canonical": "TIPO_PEDIMENTO", "aliases": ["PEDIMENTO_TYPE", "TIPO PED"]},
         {"canonical": "VALOR_USD", "aliases": ["USD_VALUE", "VALOR USD", "USD"]},
         {"canonical": "PRECIO_PAGADO", "aliases": ["PAID_PRICE", "PRECIO PAGADO"]},
         {"canonical": "PESO_BRUTO", "aliases": ["GROSS_WEIGHT", "PESO BRUTO", "PESO"]},
@@ -25,6 +100,61 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     ],
 }
 
+# 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"]
+]
+
+
+def _first_row_looks_like_three_cols_data(first_row: List[str]) -> bool:
+    """True si las primeras 3 celdas son AÑO (2 dígitos), PATENTE (4), NUMERO (7)."""
+    if not first_row or len(first_row) < 3:
+        return False
+    c0 = (first_row[0] or "").strip()
+    c1 = (first_row[1] or "").strip()
+    c2 = (first_row[2] or "").strip()
+    return (
+        len(c0) <= AÑO_LEN and c0.isdigit()
+        and len(c1) <= PATENTE_LEN and c1.isdigit()
+        and len(c2) <= NUMERO_LEN and c2.isdigit()
+    )
+
+
+def detect_headers_or_data(
+    file_path: str,
+    normalize_header_fn,
+    parse_pedimento_fn,
+    encoding: str = "utf-8-sig",
+) -> Tuple[Optional[List[str]], bool]:
+    """
+    Lee la primera línea del CSV y decide si es cabecera o dato.
+    - Si las primeras 3 celdas son 2, 4 y 7 dígitos (AÑO, PATENTE, NUMERO) -> has_header=False.
+    - Si la primera celda tiene formato ##-####-####### (compatibilidad) -> has_header=False.
+    - Si no -> has_header=True. Devuelve (fieldnames, has_header).
+    """
+    try:
+        with open(file_path, "r", encoding=encoding) as f:
+            sample = f.read(2048)
+    except Exception:
+        return None, True
+    lines = sample.splitlines()
+    if not lines:
+        return None, True
+    try:
+        dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+    except Exception:
+        dialect = csv.excel
+    reader = csv.reader(io.StringIO(lines[0]), dialect=dialect)
+    first_row = next(reader, None)
+    if not first_row:
+        return None, True
+    if _first_row_looks_like_three_cols_data(first_row):
+        return list(PEDIMENTOS_TEMPLATE_ORDER), False
+    first_cell = (first_row[0] or "").strip()
+    if parse_pedimento_fn(first_cell) is not None:
+        return list(PEDIMENTOS_TEMPLATE_ORDER), False
+    return None, True
+
 
 def build_normalized_lookup(normalize_header_fn, template_id: str = "pedimentos") -> Dict[str, str]:
     """normalized_header -> canonical_name para plantilla pedimentos."""
@@ -51,3 +181,39 @@ def row_from_template(row: Dict[str, Any], normalize_header_fn, template_id: str
         if key_norm in lookup:
             out[lookup[key_norm]] = value
     return out
+
+
+# Formato Clarion Col A: ##-####-####### (15 chars, guiones en posiciones 3 y 8, 1-based)
+PEDIMENTO_LEN = 15
+PEDIMENTO_DASH_POSITIONS = (2, 7)  # 0-based: pos 2 y 7 deben ser '-'
+
+
+def parse_pedimento_col_a(pedimento_str: str) -> Optional[Tuple[str, str, str]]:
+    """
+    Parsea Col A (PEDIMENTO) formato ##-####-#######.
+    Retorna (year_2, license_4, pedimento_number_7) o None si formato inválido.
+    """
+    if not pedimento_str or not isinstance(pedimento_str, str):
+        return None
+    s = (pedimento_str or "").strip()
+    if len(s) != PEDIMENTO_LEN:
+        return None
+    if s[PEDIMENTO_DASH_POSITIONS[0]] != "-" or s[PEDIMENTO_DASH_POSITIONS[1]] != "-":
+        return None
+    year = s[0:2]
+    license_val = s[3:7]
+    pedimento_number = s[8:15]
+    if not year.isdigit() or not license_val.isdigit() or not pedimento_number.isdigit():
+        return None
+    return (year, license_val, pedimento_number)
+
+
+def is_clarion_layout(row_norm: Dict[str, Any]) -> bool:
+    """True si la fila tiene las 3 columnas AÑO, PATENTE, NUMERO con valor, o PEDIMENTO con valor (compatibilidad)."""
+    año = (row_norm.get("AÑO") or "").strip()
+    patente = (row_norm.get("PATENTE") or "").strip()
+    numero = (row_norm.get("NUMERO") or "").strip()
+    if año and patente and numero:
+        return True
+    ped = (row_norm.get("PEDIMENTO") or "").strip()
+    return bool(ped)
diff --git a/backend/api/v1/modules/a76/layouts_csv/pedmientos/validators/common.py b/backend/api/v1/modules/a76/layouts_csv/pedmientos/validators/common.py
index 2c3e381a..e1311509 100644
--- a/backend/api/v1/modules/a76/layouts_csv/pedmientos/validators/common.py
+++ b/backend/api/v1/modules/a76/layouts_csv/pedmientos/validators/common.py
@@ -1,59 +1,439 @@
 """
-Validaciones comunes de fila para import CSV de pedimentos (por tipo).
+Validaciones comunes de fila para import CSV de pedimentos.
+Paridad Clarion: desfase (Col N), Cols 1-3 (AÑO, PATENTE, NUMERO) u Col A PEDIMENTO, obligatorios B–H (VALIDA_TODA),
+VALIDACIONES_PEDIMENTO (Tipo I/E, Clave+Régimen+Tipo, Aduana, Patente, IND/CON, Método transporte).
 """
-from typing import Dict, Any, Optional, Set
+from typing import Dict, Any, Optional, Set, Tuple
 
+from ..template_config import (
+    AÑO_LEN,
+    PATENTE_LEN,
+    NUMERO_LEN,
+    PEDIMENTO_LEN,
+    PEDIMENTO_DASH_POSITIONS,
+    parse_pedimento_col_a,
+)
 from ..common.common_validators import (
     check_required_max,
-    check_int_in_set,
     check_in_set,
+    check_int_in_set,
+    check_optional_int_in_set,
+    check_optional_short_name,
     check_optional_max_length,
     check_optional_decimal,
     check_optional_in_set,
-    PEDIMENTO_CODE_MAX,
-    REGIMEN_MAX,
+    parse_date,
+    DATE_FORMAT_PREFERENCE_MAP,
 )
 
+# --- Desfase
+MSG_DESFASE = "Advertencia: Podría existir un desfase en esta línea."
+MSG_DESFASE_SOLUCION = "Revisar esta línea del archivo CSV y verificar cada campo esté en la posición correcta."
 
-def validate_row_pedimento_required(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
-    for col, max_len in [
-        ("AÑO", 2),
-        ("ADUANA", 3),
-        ("PATENTE", 4),
-        ("NUMERO", 7),
-    ]:
-        err = check_required_max(row, col, max_len, line_num)
+# --- Col A ---
+MSG_PEDIMENTO_VACIO = (
+    "Error: (Col. A) La columna de Pedimento está vacío y no se pueden hacer las validaciones. "
+    "Capturar en la Columna A un Pedimento nuevo o uno ya existente al cual desee actualizar campos"
+)
+MSG_PEDIMENTO_LONGITUD = (
+    "Error: (Col. A) La columna de Pedimento no cumple con la longitud o sintaxis correcta Ej. (XX-XXXX-XXXXXXX)"
+)
+MSG_PEDIMENTO_FORMATO = "Error: (Col. A) El Formato del Pedimento: {val} es incorrecto."
+MSG_PEDIMENTO_FORMATO_SOLUCION = "Capturar en la columna A el campo Pedimento con este formato ##-####-#######."
+
+# --- Cols 1-3 (AÑO, PATENTE, NUMERO) ---
+MSG_ANIO_PATENTE_NUMERO = "Error: (Cols 1-3) AÑO, PATENTE y NUMERO son obligatorios (AÑO 2 dígitos, PATENTE 4, NUMERO 7)."
+MSG_PATENTE_NO_EXISTE = "Error: (Patente) La patente {val} no está dada de alta en el catálogo de agentes aduanales."
+
+# --- Obligatorios B–H ---
+MSG_OBLIGATORIOS = "Existen campos vacios que son obligatorios, es la {campos}."
+MSG_OBLIGATORIOS_SOLUCION = "Revisar la línea del archivo y capturar los campos con la información correcta."
+
+# --- Tipo B ---
+MSG_TIPO_INVALIDO = "Error: (Col. B) El Tipo de Operación: {val} no es valido."
+MSG_TIPO_SOLUCION = "Capturar una opción valida, que seria I para Importación, E para Exportación."
+
+# --- Combinación B+C+D ---
+MSG_COMBINACION = "Error, (Col. B, C y D) La combinación de una operación de {b} con clave de pedimento {c} y régimen {d} no es posible."
+MSG_COMBINACION_SOLUCION = "Las posibles combinaciones son {comb}."
+
+# --- Aduana H ---
+MSG_ADUANA_NO_EXISTE = "Error: (Col. H) La aduana y sección: {val} no existe."
+MSG_ADUANA_SOLUCION = "Revisar que sea correcta esta aduana y seccion de cruce, de ser así actualice los catálogos fijos."
+
+# --- Col J IND/CON ---
+MSG_IND_CON_INVALIDO = "Error: (Col. J) La opción de Individual/Consolidado: {val} no es valido."
+MSG_IND_CON_SOLUCION = "Capturar una opción valida: IND para Individual, CON para Consolidado: o dejar el campo vacio y automaticamente se asigna Consolidado."
+
+# --- Transporte K/L/M ---
+MSG_TRANSP_NO_EXISTE = "Error: (Col. {col}) El Metodo de Transporte: {val} no existe en el Catálogo del Anexo 22 Apendice 3."
+MSG_TRANSP_SOLUCION = "Revisar si esta asignado Correctamente."
+
+# --- Fechas ---
+FECHA_MAX_LEN = 10
+
+
+def validate_row_desfase_pedimento(
+    raw_row: Dict[str, Any], line_num: int
+) -> Optional[Dict[str, Any]]:
+    """Si la fila tiene al menos 16 columnas y la 16ª (Col N, IEPS/desfase) tiene valor, error de desfase. Orden: AÑO,PATENTE,NUMERO,TIPO,...,IEPS en índice 15."""
+    values_ordered = list(raw_row.values()) if raw_row else []
+    desfase_idx = 15  # IEPS en PEDIMENTOS_TEMPLATE_ORDER (tras AÑO,PATENTE,NUMERO + 12 columnas más)
+    if len(values_ordered) >= (desfase_idx + 1) and (values_ordered[desfase_idx] or "").strip():
+        return {
+            "line": line_num,
+            "col": "",
+            "msg": f"{MSG_DESFASE} {MSG_DESFASE_SOLUCION}",
+        }
+    return None
+
+
+def validate_row_pedimento_required_col_a(
+    row: Dict[str, Any], line_num: int
+) -> Optional[Dict[str, Any]]:
+    """Col A (PEDIMENTO) obligatorio y longitud 15."""
+    val = (row.get("PEDIMENTO") or "").strip()
+    if not val:
+        return {"line": line_num, "col": "PEDIMENTO", "msg": MSG_PEDIMENTO_VACIO}
+    if len(val) != PEDIMENTO_LEN:
+        return {
+            "line": line_num,
+            "col": "PEDIMENTO",
+            "msg": f"{MSG_PEDIMENTO_LONGITUD} {MSG_PEDIMENTO_FORMATO_SOLUCION}",
+        }
+    return None
+
+
+def validate_row_pedimento_format(
+    row: Dict[str, Any], line_num: int
+) -> Optional[Dict[str, Any]]:
+    """Formato ##-####-#######: guiones en posiciones 3 y 8 (1-based). Solo aplica cuando la fila trae PEDIMENTO."""
+    val = (row.get("PEDIMENTO") or "").strip()
+    if not val or len(val) != PEDIMENTO_LEN:
+        return None
+    if val[PEDIMENTO_DASH_POSITIONS[0]] != "-" or val[PEDIMENTO_DASH_POSITIONS[1]] != "-":
+        return {
+            "line": line_num,
+            "col": "PEDIMENTO",
+            "msg": f"{MSG_PEDIMENTO_FORMATO.format(val=val)} {MSG_PEDIMENTO_FORMATO_SOLUCION}",
+        }
+    return None
+
+
+def validate_row_pedimento_required_three_cols(
+    row: Dict[str, Any], line_num: int
+) -> Optional[Dict[str, Any]]:
+    """Cols 1-3: AÑO (2 dígitos), PATENTE (4), NUMERO (7) obligatorios y numéricos."""
+    cols_missing = []
+    año = (row.get("AÑO") or "").strip()
+    patente = (row.get("PATENTE") or "").strip()
+    numero = (row.get("NUMERO") or "").strip()
+    if not año:
+        cols_missing.append("Col.1) AÑO")
+    elif len(año) > AÑO_LEN or not año.isdigit():
+        return {"line": line_num, "col": "AÑO", "msg": "AÑO debe ser 2 dígitos numéricos."}
+    if not patente:
+        cols_missing.append("Col.2) PATENTE")
+    elif len(patente) > PATENTE_LEN or not patente.isdigit():
+        return {"line": line_num, "col": "PATENTE", "msg": "PATENTE debe ser 4 dígitos numéricos."}
+    if not numero:
+        cols_missing.append("Col.3) NUMERO")
+    elif len(numero) > NUMERO_LEN or not numero.isdigit():
+        return {"line": line_num, "col": "NUMERO", "msg": "NUMERO debe ser 7 dígitos numéricos."}
+    if cols_missing:
+        campos = ", ".join(cols_missing)
+        return {
+            "line": line_num,
+            "col": "",
+            "msg": f"{MSG_ANIO_PATENTE_NUMERO} Faltan: {campos}.",
+        }
+    return None
+
+
+def validate_row_patente(
+    row: Dict[str, Any],
+    line_num: int,
+    valid_patentes: Set[str],
+) -> Optional[Dict[str, Any]]:
+    """Si la fila trae PATENTE o PEDIMENTO (parseado), valida que la patente esté en el catálogo de agentes aduanales."""
+    patente_val = (row.get("PATENTE") or "").strip()
+    if not patente_val:
+        ped = (row.get("PEDIMENTO") or "").strip()
+        parsed = parse_pedimento_col_a(ped) if ped else None
+        if parsed:
+            _, patente_val, _ = parsed
+    if not patente_val:
+        return None
+    if valid_patentes and patente_val not in valid_patentes:
+        return {
+            "line": line_num,
+            "col": "PATENTE",
+            "msg": MSG_PATENTE_NO_EXISTE.format(val=patente_val),
+        }
+    return None
+
+
+def validate_row_pedimento_required_full(
+    row: Dict[str, Any], line_num: int
+) -> Optional[Dict[str, Any]]:
+    """Obligatorios cuando no es actualizar: TIPO (B), CLAVE_PEDIMENTO (C), REGIMEN (D), FECHA_INICIO (E), FECHA_FINAL (F), FECHA_PAGO (G), ADUANA_SECCION_CRUCE (H)."""
+    cols_missing = []
+    col_labels = [
+        ("TIPO_OPERACION", "Col.B) Tipo Operación"),
+        ("CLAVE_PEDIMENTO", "Col.C) Clave Pedimento"),
+        ("REGIMEN", "Col.D) Clave Régimen"),
+        ("FECHA_INICIO", "Col.E) Fecha Inicio"),
+        ("FECHA_FINAL", "Col.F) Fecha Final"),
+        ("FECHA_PAGO", "Col.G) Fecha de Pago"),
+        ("ADUANA_SECCION_CRUCE", "Col.H) Aduana y Sección de Cruce"),
+    ]
+    for key, label in col_labels:
+        if not (row.get(key) or "").strip():
+            cols_missing.append(label)
+    if not cols_missing:
+        return None
+    campos = ", ".join(cols_missing)
+    return {
+        "line": line_num,
+        "col": "",
+        "msg": f"{MSG_OBLIGATORIOS.format(campos=campos)} {MSG_OBLIGATORIOS_SOLUCION}",
+    }
+
+
+def validate_row_fecha_pedimento(
+    row: Dict[str, Any],
+    col: str,
+    line_num: int,
+    date_format_preference: Optional[str] = None,
+) -> Optional[Dict[str, Any]]:
+    """Valida que el campo de fecha sea válido (longitud ≤10, parseable)."""
+    val = (row.get(col) or "").strip()
+    if not val:
+        return None
+    if len(val) > FECHA_MAX_LEN:
+        return {
+            "line": line_num,
+            "col": col,
+            "msg": f"Error: ({col}) La fecha supera la longitud de caracteres.",
+        }
+    if parse_date(val, date_format_preference) is None:
+        format_label = (
+            DATE_FORMAT_PREFERENCE_MAP.get(date_format_preference, "##/##/####")
+            if date_format_preference
+            else "##/##/####"
+        )
+        return {
+            "line": line_num,
+            "col": col,
+            "msg": f"Error: ({col}) La fecha no coincide con el formato ({format_label}).",
+        }
+    return None
+
+
+def validate_row_tipo_operacion(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
+    """Col B: si tiene valor debe ser I o E."""
+    val = (row.get("TIPO_OPERACION") or "").strip().upper()
+    if not val:
+        return None
+    if val not in ("I", "E"):
+        return {
+            "line": line_num,
+            "col": "TIPO_OPERACION",
+            "msg": f"{MSG_TIPO_INVALIDO.format(val=row.get('TIPO_OPERACION'))} {MSG_TIPO_SOLUCION}",
+        }
+    return None
+
+
+def validate_row_clave_regimen_tipo(
+    row: Dict[str, Any],
+    line_num: int,
+    valid_clave_regimen_tipo: Set[Tuple[str, str, str]],
+) -> Optional[Dict[str, Any]]:
+    """Combinación B+C+D debe existir en CodePedimentoRegimen."""
+    b = (row.get("TIPO_OPERACION") or "").strip().upper()
+    c = (row.get("CLAVE_PEDIMENTO") or "").strip().upper()
+    d = (row.get("REGIMEN") or "").strip().upper()
+    if not b or not c or not d:
+        return None
+    key = (c, d, b)
+    if valid_clave_regimen_tipo and key not in valid_clave_regimen_tipo:
+        # Opcional: listar combinaciones válidas para esa clave
+        comb = ", ".join(
+            f"operación {t} con clave {cp} y régimen {r}"
+            for cp, r, t in valid_clave_regimen_tipo
+            if cp == c
+        ) or "ninguna para esta clave"
+        return {
+            "line": line_num,
+            "col": "TIPO_OPERACION",
+            "msg": f"{MSG_COMBINACION.format(b=b, c=c, d=d)} {MSG_COMBINACION_SOLUCION.format(comb=comb)}",
+        }
+    return None
+
+
+def validate_row_aduana(
+    row: Dict[str, Any],
+    line_num: int,
+    valid_aduana_seccion: Set[str],
+) -> Optional[Dict[str, Any]]:
+    """Col H: si tiene valor debe existir en CustomsSection."""
+    val = (row.get("ADUANA_SECCION_CRUCE") or "").strip()
+    if not val:
+        return None
+    # customs_code puede ser 3 caracteres; si Col H trae más, tomar primeros 3
+    code = val[:3] if len(val) >= 3 else val
+    if valid_aduana_seccion and code not in valid_aduana_seccion:
+        return {
+            "line": line_num,
+            "col": "ADUANA_SECCION_CRUCE",
+            "msg": f"{MSG_ADUANA_NO_EXISTE.format(val=val)} {MSG_ADUANA_SOLUCION}",
+        }
+    return None
+
+
+def validate_row_ind_con(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
+    """Col J: si tiene valor debe ser IND o CON."""
+    val = (row.get("INDIVIDUAL_CONSOLIDADO") or "").strip().upper()
+    if not val:
+        return None
+    if val not in ("IND", "CON"):
+        return {
+            "line": line_num,
+            "col": "INDIVIDUAL_CONSOLIDADO",
+            "msg": f"{MSG_IND_CON_INVALIDO.format(val=val)} {MSG_IND_CON_SOLUCION}",
+        }
+    return None
+
+
+def validate_row_transporte(
+    row: Dict[str, Any],
+    line_num: int,
+    col_key: str,
+    col_label: str,
+    valid_anexo22_claves: Set[str],
+) -> Optional[Dict[str, Any]]:
+    """Valida una columna de método de transporte (K, L o M) contra Anexo 22. Si el conjunto está vacío, no se valida."""
+    val = (row.get(col_key) or "").strip().upper()
+    if not val:
+        return None
+    if valid_anexo22_claves and val not in valid_anexo22_claves:
+        return {
+            "line": line_num,
+            "col": col_key,
+            "msg": f"{MSG_TRANSP_NO_EXISTE.format(col=col_label, val=val)} {MSG_TRANSP_SOLUCION}",
+        }
+    return None
+
+
+def validaciones_pedimento(
+    row: Dict[str, Any],
+    line_num: int,
+    valid_clave_regimen_tipo: Set[Tuple[str, str, str]],
+    valid_aduana_seccion: Set[str],
+    valid_anexo22_claves: Set[str],
+    valid_patentes: Set[str],
+    short_name_to_id: Optional[Dict[str, int]] = None,
+    date_format_preference: Optional[str] = None,
+) -> Optional[Dict[str, Any]]:
+    """Orquestador: formato PEDIMENTO si aplica, tipo B, combinación B+C+D, aduana H, patente, IND/CON J, transporte K/L/M, fechas E/F/G. CLIENTE_SHORT_NAME opcional."""
+    if (row.get("PEDIMENTO") or "").strip():
+        err = validate_row_pedimento_format(row, line_num)
+        if err:
+            return err
+    # CLIENTE_SHORT_NAME opcional: si viene valor se valida que exista en catálogo; si viene vacío no se exige
+    if short_name_to_id is not None:
+        err = check_optional_short_name(row, "CLIENTE_SHORT_NAME", short_name_to_id, line_num)
+        if err:
+            return err
+    err = validate_row_patente(row, line_num, valid_patentes)
+    if err:
+        return err
+    err = validate_row_tipo_operacion(row, line_num)
+    if err:
+        return err
+    err = validate_row_clave_regimen_tipo(row, line_num, valid_clave_regimen_tipo)
+    if err:
+        return err
+    err = validate_row_aduana(row, line_num, valid_aduana_seccion)
+    if err:
+        return err
+    err = validate_row_ind_con(row, line_num)
+    if err:
+        return err
+    err = validate_row_transporte(
+        row, line_num, "MET_TRANSP_ENTRADA", "K", valid_anexo22_claves
+    )
+    if err:
+        return err
+    err = validate_row_transporte(
+        row, line_num, "MET_TRANSP_ARRIVO", "L", valid_anexo22_claves
+    )
+    if err:
+        return err
+    err = validate_row_transporte(
+        row, line_num, "MET_TRANSP_SALIDA", "M", valid_anexo22_claves
+    )
+    if err:
+        return err
+    for col in ("FECHA_INICIO", "FECHA_FINAL", "FECHA_PAGO"):
+        err = validate_row_fecha_pedimento(
+            row, col, line_num, date_format_preference
+        )
         if err:
             return err
     return None
 
 
-def validate_row_pedimento_fk(
+# --- Legacy (layout sin PEDIMENTO único): mantener para compatibilidad ---
+def validate_row_pedimento_required_legacy(
+    row: Dict[str, Any], line_num: int
+) -> Optional[Dict[str, Any]]:
+    """Requeridos para layout legacy: AÑO, ADUANA (ADUANA_SECCION_CRUCE), PATENTE, NUMERO."""
+    for col, max_len in [
+        ("AÑO", 2),
+        ("ADUANA_SECCION_CRUCE", 3),
+        ("PATENTE", 4),
+        ("NUMERO", 7),
+    ]:
+        val = (row.get(col) or "").strip()
+        if not val:
+            return {"line": line_num, "col": col, "msg": "Requerido"}
+        if len(val) > max_len:
+            return {"line": line_num, "col": col, "msg": f"Máximo {max_len} caracteres"}
+    return None
+
+
+def validate_row_pedimento_fk_legacy(
     row: Dict[str, Any],
     line_num: int,
-    valid_client_ids: Set[int],
+    short_name_to_id: Dict[str, int],
     valid_regimes: Set[str],
     valid_pedimento_codes: Set[str],
 ) -> Optional[Dict[str, Any]]:
-    err = check_int_in_set(row, "CLIENTE_ID", valid_client_ids, line_num)
+    """FK para layout legacy (CLAVE_PEDIMENTO o CODIGO_PEDIMENTO, REGIMEN). CLIENTE_SHORT_NAME opcional."""
+    err = check_optional_short_name(row, "CLIENTE_SHORT_NAME", short_name_to_id, line_num)
     if err:
         return err
+    # Template puede dar CLAVE_PEDIMENTO o CODIGO_PEDIMENTO según alias
+    code = (row.get("CLAVE_PEDIMENTO") or row.get("CODIGO_PEDIMENTO") or "").strip()
+    if not code:
+        return {"line": line_num, "col": "CLAVE_PEDIMENTO", "msg": "Requerido"}
+    if len(code) > 2:
+        return {"line": line_num, "col": "CLAVE_PEDIMENTO", "msg": "Máximo 2 caracteres"}
+    if valid_pedimento_codes and code not in valid_pedimento_codes:
+        return {"line": line_num, "col": "CLAVE_PEDIMENTO", "msg": "No existe en código pedimento"}
     err = check_in_set(
-        row, "CODIGO_PEDIMENTO", valid_pedimento_codes, line_num,
-        max_len=PEDIMENTO_CODE_MAX, catalog_name="código pedimento",
-    )
-    if err:
-        return err
-    err = check_in_set(
-        row, "REGIMEN", valid_regimes, line_num,
-        max_len=REGIMEN_MAX, catalog_name="régimen",
+        row, "REGIMEN", valid_regimes, line_num, max_len=3, catalog_name="régimen"
     )
     if err:
         return err
     return None
 
 
-def validate_row_pedimento_optionals(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
+def validate_row_pedimento_optionals_legacy(
+    row: Dict[str, Any], line_num: int
+) -> Optional[Dict[str, Any]]:
+    """Opcionales para layout legacy."""
     err = check_optional_max_length(row, "ESTATUS", 30, line_num)
     if err:
         return err
@@ -62,16 +442,16 @@ def validate_row_pedimento_optionals(row: Dict[str, Any], line_num: int) -> Opti
         if err:
             return err
     err = check_optional_in_set(
-        row, "TIPO_OPERACION", {"imp", "exp"}, line_num,
-        "Debe ser imp o exp",
+        row, "TIPO_OPERACION", {"imp", "exp"}, line_num, "Debe ser imp o exp"
     )
     if err:
         return err
     err = check_optional_in_set(
-        row, "TIPO_PEDIMENTO",
+        row,
+        "TIPO_PEDIMENTO",
         {"normal", "consolidated", "complementary", "automobile"},
         line_num,
-        "Tipo no válido (normal, consolidated, complementary, automobile)",
+        "Tipo no válido",
     )
     if err:
         return err
diff --git a/backend/api/v1/modules/a76/layouts_csv/pedmientos/validators/create.py b/backend/api/v1/modules/a76/layouts_csv/pedmientos/validators/create.py
index 4d6149a0..107daf66 100644
--- a/backend/api/v1/modules/a76/layouts_csv/pedmientos/validators/create.py
+++ b/backend/api/v1/modules/a76/layouts_csv/pedmientos/validators/create.py
@@ -1,37 +1,126 @@
 """
 Punto de entrada de validación para import de una fila pedimento.
-Recibe conjuntos FK (valid_client_ids, valid_regimes, valid_pedimento_codes) para validar referencias.
+Flujo Clarion:
+  - Modo actualizar: crear si no existe, actualizar si existe. Si existe → VALIDA_PARCIAL; si no existe → VALIDA_TODA.
+  - Modo crear: crear siempre; si existe reemplazar. Siempre VALIDA_TODA.
+Legacy: si la fila no tiene PEDIMENTO (Col A Clarion), se usa validación por AÑO/ADUANA/PATENTE/NUMERO.
 """
-from typing import Dict, Any, Optional, Set
+from typing import Dict, Any, Optional, Set, Tuple
 
+from ..template_config import is_clarion_layout, parse_pedimento_col_a
+from ..common.fk_loader import pedimento_key_from_parsed
 from .common import (
-    validate_row_pedimento_required,
-    validate_row_pedimento_fk,
-    validate_row_pedimento_optionals,
+    validate_row_desfase_pedimento,
+    validate_row_pedimento_required_col_a,
+    validate_row_pedimento_format,
+    validate_row_pedimento_required_three_cols,
+    validate_row_pedimento_required_full,
+    validaciones_pedimento,
+    validate_row_pedimento_required_legacy,
+    validate_row_pedimento_fk_legacy,
+    validate_row_pedimento_optionals_legacy,
 )
 
 
+def _clarion_pedimento_key(row: Dict[str, Any]) -> Optional[str]:
+    """Construye la clave (year|customs_office|license|pedimento_number) desde fila Clarion (3 cols o PEDIMENTO)."""
+    año = (row.get("AÑO") or "").strip()
+    patente = (row.get("PATENTE") or "").strip()
+    numero = (row.get("NUMERO") or "").strip()
+    if año and patente and numero:
+        aduana = (row.get("ADUANA_SECCION_CRUCE") or "").strip()[:3]
+        return pedimento_key_from_parsed(año[:2], aduana, patente[:4], numero[:7])
+    parsed = parse_pedimento_col_a((row.get("PEDIMENTO") or "").strip())
+    if not parsed:
+        return None
+    year, license_val, pedimento_number = parsed
+    aduana = (row.get("ADUANA_SECCION_CRUCE") or "").strip()[:3]
+    return pedimento_key_from_parsed(year, aduana, license_val, pedimento_number)
+
+
 def validate_row_pedimento(
     row: Dict[str, Any],
     line_num: int,
-    valid_client_ids: Set[int],
+    short_name_to_id: Dict[str, int],
     valid_regimes: Set[str],
     valid_pedimento_codes: Set[str],
+    valid_clave_regimen_tipo: Set[Tuple[str, str, str]],
+    valid_aduana_seccion: Set[str],
+    existing_pedimento_keys: Set[str],
+    valid_anexo22_claves: Set[str],
+    valid_patentes: Set[str],
+    actualizar: bool = False,
+    raw_row: Optional[Dict[str, Any]] = None,
+    date_format_preference: Optional[str] = None,
 ) -> Optional[Dict[str, Any]]:
     """
     Valida una fila de CSV de pedimentos.
-    Requeridos: AÑO, ADUANA, PATENTE, NUMERO, CLIENTE_ID, CODIGO_PEDIMENTO, REGIMEN (y FKs en catálogos).
-    Opcionales: ESTATUS, decimales, TIPO_OPERACION, TIPO_PEDIMENTO.
+    - Si raw_row está presente, se valida desfase (Col N) primero.
+    - Layout Clarion (PEDIMENTO con valor): flujo VALIDA_TODA / VALIDA_PARCIAL según actualizar y existing_pedimento_keys.
+    - Layout legacy (sin PEDIMENTO): requeridos AÑO, ADUANA, PATENTE, NUMERO, CODIGO_PEDIMENTO, REGIMEN; CLIENTE_SHORT_NAME opcional.
     """
-    err = validate_row_pedimento_required(row, line_num)
-    if err:
-        return err
-    err = validate_row_pedimento_fk(
-        row, line_num, valid_client_ids, valid_regimes, valid_pedimento_codes
+    if raw_row is not None:
+        err = validate_row_desfase_pedimento(raw_row, line_num)
+        if err:
+            return err
+
+    if not is_clarion_layout(row):
+        err = validate_row_pedimento_required_legacy(row, line_num)
+        if err:
+            return err
+        err = validate_row_pedimento_fk_legacy(
+            row, line_num, short_name_to_id, valid_regimes, valid_pedimento_codes
+        )
+        if err:
+            return err
+        return validate_row_pedimento_optionals_legacy(row, line_num)
+
+    # Layout Clarion: validar Cols 1-3 (AÑO, PATENTE, NUMERO) o Col A (PEDIMENTO) si compatibilidad
+    has_three_cols = (
+        (row.get("AÑO") or "").strip()
+        and (row.get("PATENTE") or "").strip()
+        and (row.get("NUMERO") or "").strip()
     )
+    if has_three_cols:
+        err = validate_row_pedimento_required_three_cols(row, line_num)
+    else:
+        err = validate_row_pedimento_required_col_a(row, line_num)
+        if not err:
+            err = validate_row_pedimento_format(row, line_num)
     if err:
         return err
-    err = validate_row_pedimento_optionals(row, line_num)
+
+    key = _clarion_pedimento_key(row)
+    # Actualizar: si existe → validación parcial; si no existe → validación completa (crear).
+    # Crear: siempre validación completa (crear siempre; si existe reemplazar).
+    use_partial = (
+        actualizar
+        and key is not None
+        and key in existing_pedimento_keys
+    )
+
+    if use_partial:
+        return validaciones_pedimento(
+            row,
+            line_num,
+            valid_clave_regimen_tipo,
+            valid_aduana_seccion,
+            valid_anexo22_claves,
+            valid_patentes,
+            short_name_to_id=short_name_to_id,
+            date_format_preference=date_format_preference,
+        )
+
+    err = validate_row_pedimento_required_full(row, line_num)
     if err:
         return err
-    return None
+    return validaciones_pedimento(
+        row,
+        line_num,
+        valid_clave_regimen_tipo,
+        valid_aduana_seccion,
+        valid_anexo22_claves,
+        valid_patentes,
+        short_name_to_id=short_name_to_id,
+        date_format_preference=date_format_preference,
+    )
diff --git a/backend/api/v1/modules/a76/pedmientos/dtos/pedimentos.py b/backend/api/v1/modules/a76/pedmientos/dtos/pedimentos.py
index 4767f77d..ffde30df 100644
--- a/backend/api/v1/modules/a76/pedmientos/dtos/pedimentos.py
+++ b/backend/api/v1/modules/a76/pedmientos/dtos/pedimentos.py
@@ -107,7 +107,7 @@ class PedimentosCreate(PedimentosBase):
     customs_office: str = Field(..., max_length=3, description="Customs office")
     license: str = Field(..., max_length=4, description="License")
     pedimento_number: str = Field(..., max_length=7, description="Pedimento number")
-    client_id: int = Field(..., description="Client ID")
+    client_id: Optional[int] = Field(None, description="Client ID (opcional)")
     # operation_type, pedimento_type, status son opcionales - se pueden llenar después
     pedimento_code: str = Field(..., max_length=2, description="Pedimento key")
     regime: str = Field(..., max_length=3, description="Regime")
diff --git a/backend/api/v1/modules/a76/pedmientos/models/pedimentos.py b/backend/api/v1/modules/a76/pedmientos/models/pedimentos.py
index f97600dc..fe104ac7 100644
--- a/backend/api/v1/modules/a76/pedmientos/models/pedimentos.py
+++ b/backend/api/v1/modules/a76/pedmientos/models/pedimentos.py
@@ -128,7 +128,9 @@ class Pedimentos(Base, TenantScopedMixin, TimestampMixin):
     customs_office: Mapped[str] = mapped_column(String(3))
     license: Mapped[str] = mapped_column(String(4))
     pedimento_number: Mapped[str] = mapped_column(String(7))
-    client_id: Mapped[int] = mapped_column(Integer)
+    client_id: Mapped[Optional[int]] = mapped_column(
+        Integer, nullable=True
+    )  # Opcional: no requerido para el pedimento
     operation_type: Mapped[OperationType] = mapped_column(String(3))
     pedimento_type: Mapped[PedimentoType] = mapped_column(String(20))
     pedimento_code: Mapped[str] = mapped_column(String(2))
diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts
index f966c5db..358f8e01 100644
--- a/frontend/src/lib/api.ts
+++ b/frontend/src/lib/api.ts
@@ -476,11 +476,18 @@ export const api = {
 
 	// CSV import for Pedimentos (pedimentos/imports)
 	pedimentosImports: {
-		upload: (file: File, companyId: number) => {
+		upload: (
+			file: File,
+			companyId: number,
+			params?: { actualizar?: boolean; dateFormat?: string }
+		) => {
 			const formData = new FormData();
 			formData.append('file', file);
+			const search = new URLSearchParams({ company_id: String(companyId) });
+			if (params?.actualizar !== undefined) search.set('actualizar', String(!!params.actualizar));
+			if (params?.dateFormat != null) search.set('dateFormat', params.dateFormat);
 			return fetchApi(
-				`/v1/a76/pedimentos/imports/upload?company_id=${companyId}`,
+				`/v1/a76/pedimentos/imports/upload?${search.toString()}`,
 				{ method: 'POST', body: formData }
 			);
 		},
diff --git a/frontend/src/lib/components/dashboard/pedimentos/edit/general-tab-form.svelte b/frontend/src/lib/components/dashboard/pedimentos/edit/general-tab-form.svelte
index aa8d3f8e..88b31a85 100644
--- a/frontend/src/lib/components/dashboard/pedimentos/edit/general-tab-form.svelte
+++ b/frontend/src/lib/components/dashboard/pedimentos/edit/general-tab-form.svelte
@@ -665,9 +665,9 @@
 			</div>
 
 			<div class="grid grid-cols-1 gap-4 md:grid-cols-2">
-				<!-- ID del Cliente -->
+				<!-- ID del Cliente (opcional) -->
 				<div class="space-y-2">
-					<Label for="client_id">Cliente <span class="text-red-500">*</span></Label>
+					<Label for="client_id">Cliente</Label>
 					<Select.Root
 						type="single"
 						value={String(formData.client_id ?? '')}
diff --git a/frontend/src/routes/dashboard/csv-upload/+page.svelte b/frontend/src/routes/dashboard/csv-upload/+page.svelte
index ffd8fdfb..bc17f489 100644
--- a/frontend/src/routes/dashboard/csv-upload/+page.svelte
+++ b/frontend/src/routes/dashboard/csv-upload/+page.svelte
@@ -221,7 +221,14 @@
 
 		if (usePedimentosImport) {
 			try {
-				const res = await api.pedimentosImports.upload(file, companyId);
+				const catalogosSettings = allSettings['catalogos'] || {};
+				const actualizar = catalogosSettings['mode'] === 'update';
+				const dateFormat =
+					globalSettings['dateFormat'] ?? catalogosSettings['dateFormat'] ?? globalCsvParams.find((p) => p.name === 'dateFormat')?.defaultValue ?? 'dd/mm/yyyy';
+				const res = await api.pedimentosImports.upload(file, companyId, {
+					actualizar,
+					dateFormat
+				});
 				if (res.data?.job_id) {
 					currentJobId = res.data.job_id;
 					pollStatus();
@@ -516,11 +523,12 @@
 				const skippedFk = res.data?.skipped_missing_fk || 0;
 				const skippedDup = res.data?.skipped_duplicate || 0;
 				const totalSkipped = skippedInvalid + skippedFk + skippedDup;
+				const backendMessage = res.data?.message;
 
 				if (inserted === 0) {
-					toast.error(`No se insertaron registros. ${totalSkipped} fueron rechazados.`);
+					toast.error(backendMessage || `No se insertaron registros. ${totalSkipped} fueron rechazados.`);
 				} else {
-					toast.warning(`Solo se insertaron ${inserted} de ${inserted + totalSkipped} registros.`);
+					toast.warning(backendMessage || `Solo se insertaron ${inserted} de ${inserted + totalSkipped} registros.`);
 				}
 				isUploading = false;
 			} else if (res.data?.status === 'finished') {
diff --git a/frontend/src/routes/dashboard/pedimentos/edit/[id]/+page.svelte b/frontend/src/routes/dashboard/pedimentos/edit/[id]/+page.svelte
index ccd4acc7..cf4c3a14 100644
--- a/frontend/src/routes/dashboard/pedimentos/edit/[id]/+page.svelte
+++ b/frontend/src/routes/dashboard/pedimentos/edit/[id]/+page.svelte
@@ -408,14 +408,13 @@
 				}
 			}
 
-			// Validar campos requeridos para creación
+			// Validar campos requeridos para creación (client_id es opcional)
 			if (data.isCreate && generalFormData) {
-				const requiredFields = {
+				const requiredFields: Record<string, string> = {
 					year: 'Año',
 					customs_office: 'Aduana',
 					license: 'Patente',
 					pedimento_number: 'Número de Pedimento',
-					client_id: 'ID del Cliente',
 					pedimento_code: 'Clave',
 					regime: 'Régimen'
 				};