feature/validaciones-clarion-cliente-proveedor

.gitignore

@@ -30,6 +30,7 @@ wheels/
 backend/.env
 frontend/.env
 backend/SCRIPTS/
+.cursor/
 
 # IDEs
 .vscode/

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

@@ -1,5 +1,7 @@
 """
 Validadores reutilizables para import CSV de clientes y proveedores.
+Paridad Clarion: procedencia E/N, tipo C/P/A, clave máx 8, SECON, Prosec, Vinculación,
+Es Empresa Certificada, Transformador/SubMaq, desfase.
 """
 from typing import Dict, Any, Optional
 
@@ -8,8 +10,15 @@ from api.v1.modules.a76.clients_and_providers.models import ClientOrProviderEnum
 RFC_MAX = 30
 NAME_MAX = 256
 SHORT_NAME_MAX = 10
+# Clarion: Col C máx 8 caracteres
+SHORT_NAME_MAX_CLARION = 8
 CURP_MAX = 19
 
+# Valores permitidos Col Q (Tipo programa SECON)
+TIPO_PROGRAMA_SECON_VALIDOS = frozenset(
+    {"IMMEX", "Maquila", "Pitex", "Ecex", "RECIME", "Pronex", "Ninguno"}
+)
+
 
 def check_required_max(row: Dict[str, Any], col: str, max_len: int, line_num: int) -> Optional[Dict[str, Any]]:
     val = (row.get(col) or "").strip()
@@ -32,12 +41,21 @@ def check_max_length(row: Dict[str, Any], col: str, max_len: int, line_num: int)
 def parse_client_or_provider(val: Optional[str]) -> Optional[ClientOrProviderEnum]:
     if not val or not str(val).strip():
         return None
-    v = str(val).strip().lower()
-    if v in ("client", "cliente", "c"):
+    s = str(val).strip()
+    v = s.lower()
+    # Una sola letra: C, P, A o B (Clarion: A=Ambos; B también usado como Ambos)
+    if len(v) == 1:
+        if v == "c":
+            return ClientOrProviderEnum.CLIENT
+        if v == "p":
+            return ClientOrProviderEnum.PROVIDER
+        if v in ("a", "b"):
+            return ClientOrProviderEnum.BOTH
+    if v in ("client", "cliente"):
         return ClientOrProviderEnum.CLIENT
-    if v in ("provider", "proveedor", "p"):
+    if v in ("provider", "proveedor"):
         return ClientOrProviderEnum.PROVIDER
-    if v in ("both", "ambos", "b", "cliente y proveedor"):
+    if v in ("both", "ambos"):
         return ClientOrProviderEnum.BOTH
     return None
 
@@ -50,11 +68,211 @@ def check_tipo_client_provider(row: Dict[str, Any], line_num: int) -> Optional[D
         return {
             "line": line_num,
             "col": "TIPO",
-            "msg": "Valor no válido. Use Cliente, Proveedor o Ambos.",
+            "msg": "Valor no válido. Use Cliente (C), Proveedor (P), Ambos (A) o dejar vacío.",
+            "solution": "Capturar una opción valida: C para Cliente, P para Proveedor, A para Ambos o dejar el campo vacio.",
         }
     return None
 
 
+def check_procedencia(
+    row: Dict[str, Any], line_num: int
+) -> Optional[Dict[str, Any]]:
+    """Col A: E o N. Si PAIS (Col L) tiene valor: N → MEX, E → no MEX."""
+    val = (row.get("PROCEDENCIA") or "").strip().upper()
+    if not val:
+        return None
+    if val not in ("E", "N"):
+        return {
+            "line": line_num,
+            "col": "PROCEDENCIA",
+            "msg": f"Error: (Col. A) El tipo de cliente: {val} es incorrecto.",
+            "solution": "Capturar en la columna A el Tipo de cliente correcto: E para Extranjero y N para Nacional.",
+        }
+    pais = (row.get("PAIS") or "").strip().upper()
+    if not pais:
+        return None
+    if val == "N" and pais != "MEX":
+        return {
+            "line": line_num,
+            "col": "PAIS",
+            "msg": f"Error: (Col. L) El tipo de cliente es Nacional y tiene el país {pais}, es incorrecto.",
+            "solution": "Capturar en la columna L el país con clave MEX, ya que es un cliente Nacional.",
+        }
+    if val == "E" and pais == "MEX":
+        return {
+            "line": line_num,
+            "col": "PAIS",
+            "msg": f"Error: (Col. L) El tipo de cliente es Extranjero y tiene el país MEX, es incorrecto.",
+            "solution": "Capturar en la columna L un país con clave diferente de MEX, ya que es un cliente Extranjero.",
+        }
+    return None
+
+
+def check_short_name_max_clarion(
+    row: Dict[str, Any], line_num: int, max_len: int = SHORT_NAME_MAX_CLARION
+) -> Optional[Dict[str, Any]]:
+    """Col C: Clave cliente/proveedor máx 8 caracteres (Clarion)."""
+    val = (row.get("SHORT_NAME") or "").strip()
+    if not val:
+        return None
+    if len(val) > max_len:
+        return {
+            "line": line_num,
+            "col": "SHORT_NAME",
+            "msg": f"Error: (Col. C) La Clave de Cliente/Proveedor supera la longitud de caracteres (máx {max_len}).",
+            "solution": f"Capturar en la columna C una Clave de Cliente/Proveedor de {max_len} caracteres como máximo.",
+        }
+    return None
+
+
+def check_tipo_programa_secon(
+    row: Dict[str, Any], line_num: int
+) -> Optional[Dict[str, Any]]:
+    """Col Q: IMMEX, Maquila, Pitex, Ecex, RECIME, Pronex, Ninguno. Si no Ninguno → R y S obligatorios; si Ninguno → R y S vacíos."""
+    q = (row.get("TIPO_PROGRAMA_SECON") or "").strip()
+    r = (row.get("NUM_PROGRAMA_SECON") or "").strip()
+    s = (row.get("FECHA_AUT_SECON") or "").strip()
+    if q and q not in TIPO_PROGRAMA_SECON_VALIDOS:
+        return {
+            "line": line_num,
+            "col": "TIPO_PROGRAMA_SECON",
+            "msg": f"Error: (Col. Q) El Tipo de Programa SECON: {q} es incorrecto.",
+            "solution": "Capturar los Tipos de Programa correctos: IMMEX, Maquila, Pitex, Ecex, RECIME, Pronex o Ninguno.",
+        }
+    if not q or q == "Ninguno":
+        if r:
+            return {
+                "line": line_num,
+                "col": "NUM_PROGRAMA_SECON",
+                "msg": "Error: (Col. R) El Tipo de Programa es Ninguno y está capturado el número de programa.",
+                "solution": "Borrar la información en la columna R del archivo o asignar un Programa en la columna Q.",
+            }
+        if s:
+            return {
+                "line": line_num,
+                "col": "FECHA_AUT_SECON",
+                "msg": "Error: (Col. S) El Tipo de Programa es Ninguno y está capturada la Fecha de autorización.",
+                "solution": "Borrar la información en la columna S del archivo o asignar un Programa en la columna Q.",
+            }
+        return None
+    # No es Ninguno: R y S obligatorios
+    if not r:
+        return {
+            "line": line_num,
+            "col": "NUM_PROGRAMA_SECON",
+            "msg": f"Error: (Col. R) El Tipo de Programa es: {q} y no está capturado el número de programa.",
+            "solution": "Capturarlo en la columna R del archivo.",
+        }
+    if not s:
+        return {
+            "line": line_num,
+            "col": "FECHA_AUT_SECON",
+            "msg": f"Error: (Col. S) El Tipo de Programa es: {q} y no está capturada la Fecha de autorización.",
+            "solution": "Capturarla en la columna S del archivo.",
+        }
+    return None
+
+
+def check_es_prosec_num_aut(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
+    """Col T (SI/NO): si SI → Col U obligatoria; si no SI → Col U vacía."""
+    t = (row.get("ES_PROSEC") or "").strip().upper()
+    u = (row.get("NUM_AUT_PROSEC") or "").strip()
+    if t and t not in ("SI", "NO"):
+        return {
+            "line": line_num,
+            "col": "ES_PROSEC",
+            "msg": f"Error: (Col. T) La opción de si Es Prosec? {t} no es valida.",
+            "solution": "Capturar una opción valida: SI, NO, o dejar el campo vacio (se asigna NO).",
+        }
+    if t == "SI" and not u:
+        return {
+            "line": line_num,
+            "col": "NUM_AUT_PROSEC",
+            "msg": "Error: (Col. U) Es Prosec? es SI y no está capturado el número de permiso.",
+            "solution": "Capturar en la columna U el número de Permiso PROSEC o cambiar la opcion a NO en la columna T.",
+        }
+    if t != "SI" and u:
+        return {
+            "line": line_num,
+            "col": "NUM_AUT_PROSEC",
+            "msg": "Error: (Col. U) Es Prosec? no es SI y está capturado el número de permiso.",
+            "solution": "Borrar la información de la columna U o cambiar la opcion a SI en la columna T.",
+        }
+    return None
+
+
+def check_vinculacion(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
+    """Col V: solo 0, 1 o 2 (o vacío → 0)."""
+    val = (row.get("VINCULACION") or "").strip()
+    if not val:
+        return None
+    if val not in ("0", "1", "2"):
+        return {
+            "line": line_num,
+            "col": "VINCULACION",
+            "msg": f"Error: (Col. V) La opción de Vinculación: {val} no es valida.",
+            "solution": "Capturar una opción valida: 0, 1, 2 o dejar el campo vacio (se asigna 0).",
+        }
+    return None
+
+
+def check_es_empresa_certificada_registro(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
+    """Col W (SI/NO): si SI → Col X obligatoria; si no SI → Col X vacía."""
+    w = (row.get("ES_EMPRESA_CERTIFICADA") or "").strip().upper()
+    x = (row.get("REGISTRO_EMPRESA_CERT") or "").strip()
+    if w and w not in ("SI", "NO"):
+        return {
+            "line": line_num,
+            "col": "ES_EMPRESA_CERTIFICADA",
+            "msg": f"Error: (Col. W) La opción de si Es Empresa Certificada? {w} no es valida.",
+            "solution": "Capturar una opción valida: SI, NO, o dejar el campo vacio (se asigna NO).",
+        }
+    if w == "SI" and not x:
+        return {
+            "line": line_num,
+            "col": "REGISTRO_EMPRESA_CERT",
+            "msg": "Error: (Col. X) Es Empresa Certificada? es SI y no está capturado el número de empresa certificada.",
+            "solution": "Capturar en la columna X el número de Empresa Certificada o cambiar la opción a NO en la columna W.",
+        }
+    if w != "SI" and x:
+        return {
+            "line": line_num,
+            "col": "REGISTRO_EMPRESA_CERT",
+            "msg": "Error: (Col. X) Es Empresa Certificada? no es SI y está capturado el número de empresa certificada.",
+            "solution": "Borrar la información de la columna X o cambiar la opción a SI en la columna W.",
+        }
+    return None
+
+
+def check_transformador_submaq(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
+    """Col AF: primera letra T, S o N (Transformador, SubMaquila, Ninguno) o vacío → Ninguno."""
+    val = (row.get("TRANSFORMA_SUBMAQ") or "").strip().upper()
+    if not val:
+        return None
+    first = val[:1] if val else ""
+    if first not in ("T", "S", "N"):
+        return {
+            "line": line_num,
+            "col": "TRANSFORMA_SUBMAQ",
+            "msg": f"Error: (Col. AF) La opción (SUBMAQUILA/TRANSFORMADOR/NINGUNO): {val} no es valida.",
+            "solution": "Capturar una opción valida: Transformador, SubMaquila, Ninguno o dejar el campo vacio (se asigna Ninguno).",
+        }
+    return None
+
+
+def check_desfase(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
+    """Si COL_EXTRA (Col AH) tiene valor → advertencia de desfase."""
+    val = (row.get("COL_EXTRA") or "").strip()
+    if not val:
+        return None
+    return {
+        "line": line_num,
+        "col": "COL_EXTRA",
+        "msg": "Advertencia: Podría existir un desfase en esta línea.",
+        "solution": "Revisar esta línea del archivo CSV y verificar cada campo esté en la posición correcta.",
+    }
+
+
 def parse_active(val: Optional[str]) -> bool:
     if not val or not str(val).strip():
         return True

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

@@ -1,6 +1,16 @@
 """
-Mapeo fila CSV → datos para ClientProvider y opcional ClientProviderAddress.
+Mapeo fila CSV → datos para ClientProvider, ClientProviderAddress y ClientProviderPrograms.
+Paridad Clarion: columnas A–AG a modelos.
+
+El CSV se alimenta en base a las tablas/modelos:
+- cp_data: atributos de ClientProvider (clients_and_providers). Claves = nombres de columna del modelo.
+- address_data: atributos de ClientProviderAddress (clients_and_providers_address). Se asignan en tasks.
+- programs_data: atributos de ClientProviderPrograms (clients_and_providers_programs). Se asignan en tasks.
+
+Las validaciones (validators) aplican reglas de negocio Clarion y respetan longitudes máximas de los modelos.
 """
+from datetime import datetime
+from decimal import Decimal
 from typing import Dict, Any, Optional, Tuple
 
 from api.v1.modules.a76.clients_and_providers.models import ClientOrProviderEnum
@@ -26,6 +36,21 @@ MAX_LEN = {
     "state": 30,
     "country": 3,
     "contact": 50,
+    "extra_information": 399,
+    "web_key": 40,
+    "program": 7,
+    "program_number": 40,
+    "prosec_authorization": 20,
+    "secon_authorization": 20,
+    "manufacturer_id": 25,
+    "tax_id_programs": 30,
+    "broker": 6,
+    "import_broker": 6,
+    "transfer_key": 8,
+    "certified_company_registry": 40,
+    "neighborhood": 40,
+    "exterior_number": 20,
+    "fax": 30,
 }
 
 
@@ -40,48 +65,136 @@ def _str_or_none(val: Any, max_len: Optional[int] = None) -> Optional[str]:
     return s
 
 
+def _parse_date_to_yyyymmdd(val: Any) -> Optional[int]:
+    """Convierte fecha dd/mm/yyyy o similar a entero YYYYMMDD para secon_auth_date."""
+    if not val or not str(val).strip():
+        return None
+    s = str(val).strip()
+    for fmt in ("%d/%m/%Y", "%Y-%m-%d", "%d-%m-%Y", "%Y/%m/%d"):
+        try:
+            d = datetime.strptime(s[:10], fmt)
+            return d.year * 10000 + d.month * 100 + d.day
+        except ValueError:
+            continue
+    return None
+
+
 def row_to_client_provider_data(
     row_norm: Dict[str, Any], tenant_id: int, company_id: int
-) -> Tuple[Dict[str, Any], Optional[Dict[str, Any]]]:
+) -> Tuple[Dict[str, Any], Optional[Dict[str, Any]], Optional[Dict[str, Any]]]:
     """
-    Mapea fila normalizada a datos para ClientProvider y opcional ClientProviderAddress.
-    Devuelve (cp_data, address_data_or_none). address_data es para crear después del flush (necesita client_id).
+    Mapea fila normalizada a datos para ClientProvider, ClientProviderAddress y ClientProviderPrograms.
+    Devuelve (cp_data, address_data_or_none, programs_data_or_none).
+    Para compatibilidad con Clarion: se requiere RFC o SHORT_NAME para considerar la fila válida.
     """
     rfc = _str_or_none(row_norm.get("RFC"), MAX_LEN["rfc"])
-    if not rfc:
-        return ({}, None)
+    short_name = _str_or_none(row_norm.get("SHORT_NAME"), MAX_LEN["short_name"])
+    if not rfc and not short_name:
+        return ({}, None, None)
 
     client_or_provider = parse_client_or_provider(row_norm.get("TIPO")) or ClientOrProviderEnum.BOTH
+    procedencia = _str_or_none(row_norm.get("PROCEDENCIA"), 1)
+    if procedencia:
+        procedencia = procedencia.upper()[:1]
+
+    # Vinculación 0/1/2 → string
+    vinc = (row_norm.get("VINCULACION") or "").strip()
+    linking = None
+    if vinc in ("0", "1", "2"):
+        linking = vinc
+
+    # Transformador/SubMaquila: primera letra T/S/N
+    trans = (row_norm.get("TRANSFORMA_SUBMAQ") or "").strip().upper()[:1]
+    transform_subassembly = trans if trans in ("T", "S", "N") else None
 
     cp_data = {
         "tenant_id": tenant_id,
         "company_id": company_id,
-        "rfc": rfc,
+        "rfc": rfc or None,
         "name": _str_or_none(row_norm.get("NOMBRE"), MAX_LEN["name"]),
-        "short_name": _str_or_none(row_norm.get("SHORT_NAME"), MAX_LEN["short_name"]),
+        "short_name": short_name,
         "curp": _str_or_none(row_norm.get("CURP"), MAX_LEN["curp"]),
         "client_or_provider": client_or_provider,
+        "type_nat_foreign": procedencia,
+        "linking": linking,
+        "transform_subassembly": transform_subassembly,
+        "extra_information": _str_or_none(row_norm.get("INFORMACION_EXTRA"), MAX_LEN["extra_information"]),
+        "web_key": _str_or_none(row_norm.get("CLAVE_WEB"), MAX_LEN["web_key"]),
         "responsible": _str_or_none(row_norm.get("RESPONSABLE"), MAX_LEN["responsible"]),
         "position": _str_or_none(row_norm.get("POSICION"), MAX_LEN["position"]),
         "incoterm": _str_or_none(row_norm.get("INCOTERM"), MAX_LEN["incoterm"]),
         "is_active": parse_active(row_norm.get("ACTIVO")),
+        "is_national_provider": True if procedencia == "N" else (False if procedencia == "E" else None),
     }
 
+    # Address
     email = _str_or_none(row_norm.get("EMAIL"), MAX_LEN["email"])
     phone = _str_or_none(row_norm.get("TELEFONO"), MAX_LEN["phone"])
     address_str = _str_or_none(row_norm.get("DIRECCION"), MAX_LEN["address"])
-    if email or phone or address_str:
+    address_data = None
+    if email or phone or address_str or _str_or_none(row_norm.get("CODIGO POSTAL")) or _str_or_none(row_norm.get("COLONIA")) or _str_or_none(row_norm.get("NUM_EXT")):
         address_data = {
             "tenant_id": tenant_id,
             "company_id": company_id,
             "streets": address_str,
+            "exterior_number": _str_or_none(row_norm.get("NUM_EXT"), MAX_LEN["exterior_number"]),
+            "neighborhood": _str_or_none(row_norm.get("COLONIA"), MAX_LEN["neighborhood"]),
             "postal_code": _str_or_none(row_norm.get("CODIGO POSTAL"), MAX_LEN["postal_code"]),
             "city": _str_or_none(row_norm.get("CIUDAD"), MAX_LEN["city"]),
             "state": _str_or_none(row_norm.get("ESTADO"), MAX_LEN["state"]),
             "country": _str_or_none(row_norm.get("PAIS"), MAX_LEN["country"]),
             "phone": phone,
+            "fax_number": _str_or_none(row_norm.get("FAX"), MAX_LEN["fax"]),
             "email": email,
             "contact": _str_or_none(row_norm.get("CONTACTO"), MAX_LEN["contact"]),
         }
-        return (cp_data, address_data)
-    return (cp_data, None)
+
+    # Programs (SECON, PROSEC, empresa certificada, etc.)
+    tipo_prog = _str_or_none(row_norm.get("TIPO_PROGRAMA_SECON"), MAX_LEN["program"])
+    if tipo_prog and tipo_prog.lower() == "ninguno":
+        tipo_prog = None
+    num_prog = _str_or_none(row_norm.get("NUM_PROGRAMA_SECON"), MAX_LEN["program_number"])
+    fecha_secon = _parse_date_to_yyyymmdd(row_norm.get("FECHA_AUT_SECON"))
+    es_prosec = (row_norm.get("ES_PROSEC") or "").strip().upper()
+    prosec_val = "1" if es_prosec == "SI" else ("0" if es_prosec else None)
+    num_aut_prosec = _str_or_none(row_norm.get("NUM_AUT_PROSEC"), MAX_LEN["prosec_authorization"])
+    es_cert = (row_norm.get("ES_EMPRESA_CERTIFICADA") or "").strip().upper()
+    is_certified = "S" if es_cert == "SI" else ("N" if es_cert else None)
+    reg_cert = _str_or_none(row_norm.get("REGISTRO_EMPRESA_CERT"), MAX_LEN["certified_company_registry"])
+    vinc_prop = row_norm.get("VINCULACION")
+    applied_proportion = None
+    if vinc_prop is not None and str(vinc_prop).strip() in ("0", "1", "2"):
+        try:
+            applied_proportion = Decimal(str(vinc_prop).strip())
+        except Exception:
+            pass
+
+    programs_data = None
+    if (
+        tipo_prog or num_prog or fecha_secon is not None or prosec_val or num_aut_prosec
+        or is_certified or reg_cert
+        or _str_or_none(row_norm.get("MANUFACTURER_ID"))
+        or _str_or_none(row_norm.get("TAX_ID_PROGRAMS"))
+        or _str_or_none(row_norm.get("BROKER_EXPO"))
+        or _str_or_none(row_norm.get("BROKER_IMPO"))
+        or _str_or_none(row_norm.get("CLAVE_TRANSFER"))
+        or applied_proportion is not None
+    ):
+        programs_data = {
+            "program": tipo_prog[:7] if tipo_prog else None,
+            "program_number": num_prog,
+            "secon_authorization": num_prog,
+            "secon_auth_date": fecha_secon,
+            "prosec": prosec_val,
+            "prosec_authorization": num_aut_prosec,
+            "is_certified_company": is_certified,
+            "certified_company_registry": reg_cert,
+            "manufacturer_id": _str_or_none(row_norm.get("MANUFACTURER_ID"), MAX_LEN["manufacturer_id"]),
+            "tax_id": _str_or_none(row_norm.get("TAX_ID_PROGRAMS"), MAX_LEN["tax_id_programs"]),
+            "broker": _str_or_none(row_norm.get("BROKER_EXPO"), MAX_LEN["broker"]),
+            "import_broker": _str_or_none(row_norm.get("BROKER_IMPO"), MAX_LEN["import_broker"]),
+            "transfer_key": _str_or_none(row_norm.get("CLAVE_TRANSFER"), MAX_LEN["transfer_key"]),
+            "applied_proportion": applied_proportion,
+        }
+
+    return (cp_data, address_data, programs_data)

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

@@ -6,7 +6,7 @@ Usa layouts_csv.common (storage, normalize, meta, responses, csv_reader).
 import json
 import logging
 import os
-from typing import Dict, Any, Optional, List
+from typing import Dict, Any, Optional, List, Set
 
 from core.celery_app import celery_app
 from core.database import CoreSessionLocal
@@ -49,6 +49,26 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str,
     except ValueError as e:
         return {"status": "failed", "error": str(e)}
 
+    meta = common_meta.load_meta(file_path) or {}
+    actualizar = meta.get("actualizar", False)
+    existing_short_names: Set[str] = set()
+    if actualizar:
+        try:
+            with CoreSessionLocal() as session:
+                from api.v1.modules.a76.clients_and_providers.models import ClientProvider
+                for cp in (
+                    session.query(ClientProvider)
+                    .filter(
+                        ClientProvider.tenant_id == tenant_id,
+                        ClientProvider.company_id == company_id,
+                    )
+                    .all()
+                ):
+                    if (cp.short_name or "").strip():
+                        existing_short_names.add((cp.short_name or "").strip())
+        except Exception as e:
+            logger.warning("CP import: could not load existing short_names for ACT: %s", e)
+
     error_count = 0
     processed_rows = 0
     errors_detail: List[Dict[str, Any]] = []
@@ -61,7 +81,11 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str,
                     progress_callback(i, total_rows, error_count)
 
                 row_norm = row_from_template(row, common_normalize.normalize_header)
-                err = validate_row_client_provider(row_norm, i)
+                err = validate_row_client_provider(
+                    row_norm, i,
+                    actualizar=actualizar,
+                    existing_short_names=existing_short_names if actualizar else None,
+                )
                 if err:
                     error_count += 1
                     error_lines_list.append(err["line"])
@@ -111,12 +135,34 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
     except ValueError as e:
         return {"status": "failed", "error": str(e)}
 
+    meta = common_meta.load_meta(file_path) or {}
+    actualizar = meta.get("actualizar", False)
+    existing_short_names: Set[str] = set()
+    if actualizar:
+        from api.v1.modules.a76.clients_and_providers.models import ClientProvider
+        try:
+            with CoreSessionLocal() as session:
+                for cp in (
+                    session.query(ClientProvider)
+                    .filter(
+                        ClientProvider.tenant_id == tenant_id,
+                        ClientProvider.company_id == company_id,
+                    )
+                    .all()
+                ):
+                    if (cp.short_name or "").strip():
+                        existing_short_names.add((cp.short_name or "").strip())
+        except Exception as e:
+            logger.warning("CP commit: could not load existing short_names for ACT: %s", e)
+
     from api.v1.modules.a76.clients_and_providers.models import (
         ClientProvider,
         ClientProviderAddress,
+        ClientProviderPrograms,
     )
 
     inserted_count = 0
+    updated_count = 0
     skipped_invalid = 0
     skipped_details: List[Dict[str, Any]] = []
     meta_path = common_meta.get_meta_path(file_path)
@@ -124,6 +170,7 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
     try:
         with CoreSessionLocal() as session:
             existing_by_rfc: Dict[str, ClientProvider] = {}
+            existing_by_short_name: Dict[str, ClientProvider] = {}
             for cp in (
                 session.query(ClientProvider)
                 .filter(
@@ -132,15 +179,23 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
                 )
                 .all()
             ):
-                key = (cp.rfc or "").strip()
-                existing_by_rfc[key] = cp
+                rfc_key = (cp.rfc or "").strip()
+                if rfc_key:
+                    existing_by_rfc[rfc_key] = cp
+                sn_key = (cp.short_name or "").strip()
+                if sn_key:
+                    existing_by_short_name[sn_key] = cp
 
             for i, row in common_csv_reader.iter_csv_rows(file_path):
                 if i in error_lines:
                     continue
 
                 row_norm = row_from_template(row, common_normalize.normalize_header)
-                err = validate_row_client_provider(row_norm, i)
+                err = validate_row_client_provider(
+                    row_norm, i,
+                    actualizar=actualizar,
+                    existing_short_names=existing_short_names if actualizar else None,
+                )
                 if err:
                     skipped_invalid += 1
                     skipped_details.append({
@@ -149,29 +204,45 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
                     })
                     continue
 
-                cp_data, address_data = row_to_client_provider_data(row_norm, tenant_id, company_id)
-                if not cp_data or not cp_data.get("rfc"):
+                cp_data, address_data, programs_data = row_to_client_provider_data(row_norm, tenant_id, company_id)
+                if not cp_data:
                     skipped_invalid += 1
-                    skipped_details.append({"line": i, "reason": "RFC requerido"})
+                    skipped_details.append({"line": i, "reason": "Fila sin RFC ni Clave"})
                     continue
-
-                rfc = cp_data["rfc"]
-                existing = existing_by_rfc.get(rfc)
-                if existing:
+                if actualizar:
+                    short_name_key = (cp_data.get("short_name") or "").strip()
+                    if not short_name_key:
+                        skipped_invalid += 1
+                        skipped_details.append({"line": i, "reason": "Clave (SHORT_NAME) requerida en modo Actualizar"})
+                        continue
+                    existing = existing_by_short_name.get(short_name_key)
+                    if not existing:
+                        skipped_invalid += 1
+                        skipped_details.append({"line": i, "reason": "Clave no existe en catálogo"})
+                        continue
+                    # Merge: fill from existing when csv value is empty
                     for k, v in cp_data.items():
-                        if k not in ("tenant_id", "company_id", "rfc"):
+                        if k in ("tenant_id", "company_id"):
+                            continue
+                        if v is None or (isinstance(v, str) and not v.strip()):
+                            existing_val = getattr(existing, k, None)
+                            if existing_val is not None:
+                                cp_data[k] = existing_val
+                    for k, v in cp_data.items():
+                        if k not in ("tenant_id", "company_id"):
                             setattr(existing, k, v)
                     session.add(existing)
-                    inserted_count += 1
-                else:
-                    new_cp = ClientProvider(**cp_data)
-                    session.add(new_cp)
-                    session.flush()
-                    existing_by_rfc[rfc] = new_cp
-                    inserted_count += 1
-                    if address_data:
+                    updated_count += 1
+                    # Update address if present
+                    if address_data and existing.address:
+                        addr = existing.address
+                        for k, v in address_data.items():
+                            if k not in ("tenant_id", "company_id") and v is not None:
+                                setattr(addr, k, v)
+                        session.add(addr)
+                    elif address_data:
                         addr = ClientProviderAddress(
-                            client_id=new_cp.id,
+                            client_id=existing.id,
                             tenant_id=tenant_id,
                             company_id=company_id,
                             streets=address_data.get("streets"),
@@ -182,8 +253,118 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
                             phone=address_data.get("phone"),
                             email=address_data.get("email"),
                             contact=address_data.get("contact"),
+                            exterior_number=address_data.get("exterior_number"),
+                            neighborhood=address_data.get("neighborhood"),
+                            fax_number=address_data.get("fax_number"),
                         )
                         session.add(addr)
+                    # Update or create programs
+                    if programs_data:
+                        prog = session.query(ClientProviderPrograms).filter(
+                            ClientProviderPrograms.client_id == existing.id,
+                        ).first()
+                        if prog:
+                            for k, v in programs_data.items():
+                                if v is not None:
+                                    setattr(prog, k, v)
+                            session.add(prog)
+                        else:
+                            prog = ClientProviderPrograms(
+                                client_id=existing.id,
+                                tenant_id=tenant_id,
+                                company_id=company_id,
+                                **{k: v for k, v in programs_data.items() if v is not None},
+                            )
+                            session.add(prog)
+                else:
+                    # Alta / Reemplazar: key por RFC
+                    if not cp_data.get("rfc"):
+                        skipped_invalid += 1
+                        skipped_details.append({"line": i, "reason": "RFC requerido"})
+                        continue
+                    rfc = cp_data["rfc"]
+                    existing = existing_by_rfc.get(rfc)
+                    if existing:
+                        for k, v in cp_data.items():
+                            if k not in ("tenant_id", "company_id", "rfc"):
+                                setattr(existing, k, v)
+                        session.add(existing)
+                        updated_count += 1
+                        if address_data and existing.address:
+                            addr = existing.address
+                            for k, v in address_data.items():
+                                if k not in ("tenant_id", "company_id") and v is not None:
+                                    setattr(addr, k, v)
+                            session.add(addr)
+                        elif address_data:
+                            addr = ClientProviderAddress(
+                                client_id=existing.id,
+                                tenant_id=tenant_id,
+                                company_id=company_id,
+                                streets=address_data.get("streets"),
+                                postal_code=address_data.get("postal_code"),
+                                city=address_data.get("city"),
+                                state=address_data.get("state"),
+                                country=address_data.get("country"),
+                                phone=address_data.get("phone"),
+                                email=address_data.get("email"),
+                                contact=address_data.get("contact"),
+                                exterior_number=address_data.get("exterior_number"),
+                                neighborhood=address_data.get("neighborhood"),
+                                fax_number=address_data.get("fax_number"),
+                            )
+                            session.add(addr)
+                        if programs_data:
+                            prog = session.query(ClientProviderPrograms).filter(
+                                ClientProviderPrograms.client_id == existing.id,
+                            ).first()
+                            if prog:
+                                for k, v in programs_data.items():
+                                    if v is not None:
+                                        setattr(prog, k, v)
+                                session.add(prog)
+                            else:
+                                prog = ClientProviderPrograms(
+                                    client_id=existing.id,
+                                    tenant_id=tenant_id,
+                                    company_id=company_id,
+                                    **{k: v for k, v in programs_data.items() if v is not None},
+                                )
+                                session.add(prog)
+                    else:
+                        new_cp = ClientProvider(**cp_data)
+                        session.add(new_cp)
+                        session.flush()
+                        existing_by_rfc[rfc] = new_cp
+                        if (new_cp.short_name or "").strip():
+                            existing_by_short_name[(new_cp.short_name or "").strip()] = new_cp
+                        inserted_count += 1
+                        if address_data:
+                            addr = ClientProviderAddress(
+                                client_id=new_cp.id,
+                                tenant_id=tenant_id,
+                                company_id=company_id,
+                                streets=address_data.get("streets"),
+                                postal_code=address_data.get("postal_code"),
+                                city=address_data.get("city"),
+                                state=address_data.get("state"),
+                                country=address_data.get("country"),
+                                phone=address_data.get("phone"),
+                                email=address_data.get("email"),
+                                contact=address_data.get("contact"),
+                                exterior_number=address_data.get("exterior_number"),
+                                neighborhood=address_data.get("neighborhood"),
+                                fax_number=address_data.get("fax_number"),
+                            )
+                            session.add(addr)
+                        if programs_data:
+                            prog = ClientProviderPrograms(
+                                client_id=new_cp.id,
+                                tenant_id=tenant_id,
+                                company_id=company_id,
+                                **{k: v for k, v in programs_data.items() if v is not None},
+                            )
+                            session.add(prog)
 
             try:
                 session.commit()
@@ -203,7 +384,7 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
         meta_path=meta_path,
     )
 
-    if inserted_count == 0 and skipped_invalid > 0:
+    if inserted_count == 0 and updated_count == 0 and skipped_invalid > 0:
         return {
             "status": "warning",
             "inserted": 0,
@@ -212,9 +393,9 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
             "skipped_duplicate": 0,
             "skipped_missing_fk": 0,
             "skipped_details": skipped_details,
-            "message": f"No se insertaron registros. {skipped_invalid} rechazados.",
+            "message": f"No se insertaron ni actualizaron registros. {skipped_invalid} rechazados.",
         }
-    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",
@@ -228,7 +409,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_duplicate": 0,
         "skipped_missing_fk": 0,

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

@@ -1,26 +1,88 @@
 """
 Configuración de plantilla CSV para Clientes y Proveedores (EstructuraCatClienteProv.xls).
+Layout Clarion: Col A = PROCEDENCIA (E/N), B = TIPO (C/P/A), C = CLAVE (máx 8), D = NOMBRE, E = RFC, F–AG.
 Solo se leen columnas definidas aquí; el resto se ignora.
-Definir cabeceras según la primera fila del XLS oficial (frontend/static/csv/EstructuraCatClienteProv.xls).
+Correspondencia Clarion → canonical: A=PROCEDENCIA, B=TIPO, C=SHORT_NAME, D=NOMBRE, E=RFC, F=CALLES(DIRECCION),
+G=NUM_EXT, H=CODIGO POSTAL, I=COLONIA, J=CIUDAD, K=ESTADO, L=PAIS, M=TELEFONO, N=FAX, O=EMAIL, P=CURP,
+Q=TIPO_PROGRAMA_SECON, R=NUM_PROGRAMA_SECON, S=FECHA_AUT_SECON, T=ES_PROSEC, U=NUM_AUT_PROSEC, V=VINCULACION,
+W=ES_EMPRESA_CERTIFICADA, X=REGISTRO_EMPRESA_CERT, Y=INFORMACION_EXTRA, Z=CONTACTO, AA=MANUFACTURER_ID,
+AB=TAX_ID_PROGRAMS, AC=BROKER_EXPO, AD=BROKER_IMPO, AE=CLAVE_TRANSFER, AF=TRANSFORMA_SUBMAQ, AG=CLAVE_WEB,
+AH=COL_EXTRA (desfase).
 """
 
 from typing import Dict, List, Any, Optional
 
 TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     "client_providers": [
-        {"canonical": "NOMBRE", "aliases": ["RAZON SOCIAL", "NAME", "RAZON SOCIAL O NOMBRE"]},
-        {"canonical": "RFC", "aliases": ["TAX_ID", "TAXID", "IDENTIFICADOR FISCAL", "IDENTIFICACION FISCAL"]},
+        # 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"]},
-        {"canonical": "EMAIL", "aliases": ["CORREO", "E-MAIL", "CORREO ELECTRONICO"]},
+        # Col C - Clave cliente/proveedor (máx 8 Clarion)
         {"canonical": "SHORT_NAME", "aliases": ["CLAVE", "CLAVE CORTA", "NOMBRE CORTO", "SIGLAS"]},
-        {"canonical": "CURP", "aliases": []},
-        {"canonical": "TELEFONO", "aliases": ["PHONE", "TEL", "TELEFONO CONTACTO"]},
-        {"canonical": "DIRECCION", "aliases": ["DOMICILIO", "DIRECCION FISCAL", "CALLE"]},
+        # Col D - Nombre
+        {"canonical": "NOMBRE", "aliases": ["RAZON SOCIAL", "NAME", "RAZON SOCIAL O NOMBRE"]},
+        # Col E - RFC
+        {"canonical": "RFC", "aliases": ["TAX_ID", "TAXID", "IDENTIFICADOR FISCAL", "IDENTIFICACION FISCAL"]},
+        # Col F - Calles
+        {"canonical": "DIRECCION", "aliases": ["CALLES", "DOMICILIO", "DIRECCION FISCAL", "CALLE"]},
+        # Col G - Número exterior
+        {"canonical": "NUM_EXT", "aliases": ["NUM EXTERIOR", "NUMERO EXTERIOR", "NO EXT"]},
+        # Col H - Código postal
         {"canonical": "CODIGO POSTAL", "aliases": ["CODIGOPOSTAL", "CP", "C.P."]},
+        # Col I - Colonia
+        {"canonical": "COLONIA", "aliases": []},
+        # Col J - Ciudad
         {"canonical": "CIUDAD", "aliases": ["MUNICIPIO"]},
+        # Col K - Estado
         {"canonical": "ESTADO", "aliases": []},
-        {"canonical": "PAIS", "aliases": ["COUNTRY"]},
+        # Col L - País M3
+        {"canonical": "PAIS", "aliases": ["COUNTRY", "PAIS M3"]},
+        # Col M - Teléfono
+        {"canonical": "TELEFONO", "aliases": ["PHONE", "TEL", "TELEFONO CONTACTO"]},
+        # Col N - Fax
+        {"canonical": "FAX", "aliases": ["NUMERO DE FAX", "FAX NUMBER"]},
+        # Col O - Email
+        {"canonical": "EMAIL", "aliases": ["CORREO", "E-MAIL", "CORREO ELECTRONICO"]},
+        # Col P - CURP
+        {"canonical": "CURP", "aliases": []},
+        # Col Q - Tipo programa SECON
+        {"canonical": "TIPO_PROGRAMA_SECON", "aliases": ["PROGRAMA", "TIPO PROGRAMA SECON", "TIPO PROGRAMA"]},
+        # Col R - Número programa SECON
+        {"canonical": "NUM_PROGRAMA_SECON", "aliases": ["NUM PROGRAMA SECON", "NUMERO PROGRAMA"]},
+        # Col S - Fecha autorización SECON
+        {"canonical": "FECHA_AUT_SECON", "aliases": ["FECHA AUT SECON", "FECHA AUTORIZACION SECON"]},
+        # Col T - Es Prosec?
+        {"canonical": "ES_PROSEC", "aliases": ["ES PROSEC", "PROSEC"]},
+        # Col U - Número autorización PROSEC
+        {"canonical": "NUM_AUT_PROSEC", "aliases": ["NUM AUT PROSEC", "NUMERO AUTORIZACION PROSEC"]},
+        # Col V - Vinculación (0/1/2)
+        {"canonical": "VINCULACION", "aliases": []},
+        # Col W - Es Empresa Certificada?
+        {"canonical": "ES_EMPRESA_CERTIFICADA", "aliases": ["ES EMPRESA CERTIFICADA", "EMPRESA CERTIFICADA"]},
+        # Col X - Registro empresa certificada
+        {"canonical": "REGISTRO_EMPRESA_CERT", "aliases": ["REGISTRO EMPRESA CERTIFICADA", "REGISTRO EMP CERT"]},
+        # Col Y - Información extra
+        {"canonical": "INFORMACION_EXTRA", "aliases": ["INFORMACION EXTRA", "INFO EXTRA"]},
+        # Col Z - Contacto
         {"canonical": "CONTACTO", "aliases": ["CONTACT", "PERSONA CONTACTO"]},
+        # Col AA - Manufacturer ID
+        {"canonical": "MANUFACTURER_ID", "aliases": ["MANUFACTURERID", "MANUFACTURER ID"]},
+        # Col AB - Tax ID (programas)
+        {"canonical": "TAX_ID_PROGRAMS", "aliases": ["TAX ID", "TAXID PROGRAMS"]},
+        # Col AC - Broker exportación
+        {"canonical": "BROKER_EXPO", "aliases": ["BROKER EXPO", "BROKER EXPORTACION"]},
+        # Col AD - Broker importación
+        {"canonical": "BROKER_IMPO", "aliases": ["BROKER IMPO", "BROKER IMPORTACION"]},
+        # Col AE - Clave transfer
+        {"canonical": "CLAVE_TRANSFER", "aliases": ["CLAVE TRANSFER", "TRANSFER KEY"]},
+        # Col AF - Transformador/SubMaquila/Ninguno (T/S/N)
+        {"canonical": "TRANSFORMA_SUBMAQ", "aliases": ["TRANSFORMADOR SUBMAQUILA", "TRASFORMA SUBMAQ"]},
+        # Col AG - Clave interface web
+        {"canonical": "CLAVE_WEB", "aliases": ["CLAVE WEB", "CLAVE INTERFACE WEB", "WEB KEY"]},
+        # Col AH - Desfase (si tiene valor → advertencia)
+        {"canonical": "COL_EXTRA", "aliases": ["DESFASE", "COLUMNA EXTRA"]},
+        # Legacy / otros
         {"canonical": "RESPONSABLE", "aliases": ["RESPONSABLE AREA"]},
         {"canonical": "POSICION", "aliases": ["CARGO", "PUESTO"]},
         {"canonical": "INCOTERM", "aliases": []},

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

@@ -0,0 +1,143 @@
+"""
+Tests para validaciones CSV de clientes y proveedores (paridad Clarion).
+"""
+import pytest
+
+from api.v1.modules.a76.layouts_csv.clients_and_providers.validators.common import (
+    validate_row_client_provider,
+    validate_row_desfase,
+    validate_row_clave_vacia,
+    validaciones_cliente_o_prov,
+    valida_toda_cliente_o_prov,
+    valida_parcial_cliente_o_prov,
+)
+from api.v1.modules.a76.layouts_csv.clients_and_providers.common.common_validators import (
+    check_procedencia,
+    check_tipo_programa_secon,
+    check_es_prosec_num_aut,
+    check_vinculacion,
+    check_es_empresa_certificada_registro,
+    check_transformador_submaq,
+    check_desfase,
+    check_short_name_max_clarion,
+)
+
+
+def test_clave_vacia():
+    err = validate_row_clave_vacia({"SHORT_NAME": ""}, 1)
+    assert err is not None
+    assert err["col"] == "SHORT_NAME"
+    assert "vacía" in err["msg"]
+    assert validate_row_clave_vacia({"SHORT_NAME": "ABC"}, 1) is None
+
+
+def test_desfase():
+    assert validate_row_desfase({"COL_EXTRA": ""}, 1) is None
+    err = validate_row_desfase({"COL_EXTRA": "x"}, 1)
+    assert err is not None
+    assert "desfase" in err["msg"]
+
+
+def test_procedencia_e_n():
+    assert check_procedencia({"PROCEDENCIA": "E"}, 1) is None
+    assert check_procedencia({"PROCEDENCIA": "N"}, 1) is None
+    err = check_procedencia({"PROCEDENCIA": "X"}, 1)
+    assert err is not None
+    assert err["col"] == "PROCEDENCIA"
+    err = check_procedencia({"PROCEDENCIA": "N", "PAIS": "USA"}, 1)
+    assert err is not None
+    assert "MEX" in err["solution"]
+    err = check_procedencia({"PROCEDENCIA": "E", "PAIS": "MEX"}, 1)
+    assert err is not None
+
+
+def test_tipo_programa_secon():
+    assert check_tipo_programa_secon({"TIPO_PROGRAMA_SECON": "IMMEX", "NUM_PROGRAMA_SECON": "123", "FECHA_AUT_SECON": "01/01/2020"}, 1) is None
+    err = check_tipo_programa_secon({"TIPO_PROGRAMA_SECON": "IMMEX", "NUM_PROGRAMA_SECON": ""}, 1)
+    assert err is not None
+    assert "R" in err["col"] or "NUM_PROGRAMA" in err["col"]
+    err = check_tipo_programa_secon({"TIPO_PROGRAMA_SECON": "Ninguno", "NUM_PROGRAMA_SECON": "123"}, 1)
+    assert err is not None
+
+
+def test_es_prosec_num_aut():
+    assert check_es_prosec_num_aut({"ES_PROSEC": "SI", "NUM_AUT_PROSEC": "AUTH123"}, 1) is None
+    err = check_es_prosec_num_aut({"ES_PROSEC": "SI", "NUM_AUT_PROSEC": ""}, 1)
+    assert err is not None
+    err = check_es_prosec_num_aut({"ES_PROSEC": "NO", "NUM_AUT_PROSEC": "X"}, 1)
+    assert err is not None
+
+
+def test_vinculacion():
+    assert check_vinculacion({"VINCULACION": "0"}, 1) is None
+    assert check_vinculacion({"VINCULACION": "1"}, 1) is None
+    err = check_vinculacion({"VINCULACION": "5"}, 1)
+    assert err is not None
+
+
+def test_empresa_certificada():
+    assert check_es_empresa_certificada_registro({"ES_EMPRESA_CERTIFICADA": "SI", "REGISTRO_EMPRESA_CERT": "REG123"}, 1) is None
+    err = check_es_empresa_certificada_registro({"ES_EMPRESA_CERTIFICADA": "SI", "REGISTRO_EMPRESA_CERT": ""}, 1)
+    assert err is not None
+
+
+def test_transformador_submaq():
+    assert check_transformador_submaq({"TRANSFORMA_SUBMAQ": "T"}, 1) is None
+    assert check_transformador_submaq({"TRANSFORMA_SUBMAQ": "Ninguno"}, 1) is None
+    err = check_transformador_submaq({"TRANSFORMA_SUBMAQ": "X"}, 1)
+    assert err is not None
+
+
+def test_short_name_max_8():
+    assert check_short_name_max_clarion({"SHORT_NAME": "12345678"}, 1) is None
+    err = check_short_name_max_clarion({"SHORT_NAME": "123456789"}, 1)
+    assert err is not None
+
+
+def test_valida_toda_requiere_procedencia_y_nombre():
+    row = {"SHORT_NAME": "C1", "RFC": "RFC1", "NOMBRE": "Nom", "PROCEDENCIA": ""}
+    err = valida_toda_cliente_o_prov(row, 1, actualizar=False)
+    assert err is not None
+    assert "Procedencia" in err["msg"] or "Col.A" in err["msg"] or "obligatorios" in err["msg"]
+    row["PROCEDENCIA"] = "N"
+    row["NOMBRE"] = ""
+    err = valida_toda_cliente_o_prov(row, 1, actualizar=False)
+    assert err is not None
+    row["NOMBRE"] = "Nom"
+    row["PAIS"] = "MEX"
+    err = valida_toda_cliente_o_prov(row, 1, actualizar=False)
+    assert err is None
+
+
+def test_valida_parcial_no_requiere_nombre():
+    row = {"SHORT_NAME": "C1", "RFC": "RFC1", "TIPO": "C"}
+    err = valida_parcial_cliente_o_prov(row, 1)
+    # Sin errores de dominio (procedencia/nombre no requeridos)
+    assert err is None or err.get("col") in ("SHORT_NAME", "TIPO", "RFC")
+
+
+def test_validate_row_client_provider_full():
+    row = {
+        "PROCEDENCIA": "N",
+        "PAIS": "MEX",
+        "TIPO": "C",
+        "SHORT_NAME": "CLI1",
+        "NOMBRE": "Cliente 1",
+        "RFC": "RFC123456789",
+    }
+    err = validate_row_client_provider(row, 1, actualizar=False, existing_short_names=None)
+    assert err is None
+
+
+def test_validate_row_client_provider_actualizar_clave_no_existe():
+    row = {"SHORT_NAME": "NOEXISTE", "RFC": "RFC1", "NOMBRE": "X"}
+    err = validate_row_client_provider(row, 1, actualizar=True, existing_short_names=set())
+    assert err is not None
+    assert "Clave" in err["msg"] or "no existe" in err["msg"].lower()
+
+
+def test_validate_row_client_provider_actualizar_clave_existe_parcial():
+    row = {"SHORT_NAME": "EXIST1", "RFC": "RFC1", "TIPO": "A"}
+    err = validate_row_client_provider(row, 1, actualizar=True, existing_short_names={"EXIST1"})
+    # VALIDA_PARCIAL: no requiere PROCEDENCIA ni NOMBRE; requiere RFC al final
+    assert err is None or err.get("col") == "RFC"

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

@@ -1,52 +1,177 @@
 """
 Validaciones comunes de fila para import CSV de clientes y proveedores.
+Paridad Clarion: VALIDA_TODA_CLIENTE_O_PROV, VALIDA_PARCIAL_CLIENTE_O_PROV, VALIDACIONES_CLIENTE_O_PROV.
+Origen de reglas: código legacy Clarion (EstructuraCatClienteProv).
+Mapa columnas: A=PROCEDENCIA, B=TIPO, C=SHORT_NAME, D=NOMBRE, E=RFC, F–AG (ver template_config).
 """
-from typing import Dict, Any, Optional
+from typing import Dict, Any, Optional, Set
 
 from ..common.common_validators import (
     RFC_MAX,
     NAME_MAX,
-    SHORT_NAME_MAX,
+    SHORT_NAME_MAX_CLARION,
     CURP_MAX,
     check_required_max,
     check_max_length,
     check_tipo_client_provider,
+    check_procedencia,
+    check_short_name_max_clarion,
+    check_tipo_programa_secon,
+    check_es_prosec_num_aut,
+    check_vinculacion,
+    check_es_empresa_certificada_registro,
+    check_transformador_submaq,
+    check_desfase,
 )
 
 
-def validate_row_client_provider_required(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
-    return check_required_max(row, "RFC", RFC_MAX, line_num)
+# --- Mensajes obligatorios (VALIDA_TODA) ---
+MSG_CAMPOS_OBLIGATORIOS = "Existen campos vacíos que son obligatorios: {campos}."
+MSG_CAMPOS_OBLIGATORIOS_SOLUCION = "Revisar la línea del archivo y capturar los campos con la información correcta."
+MSG_CLAVE_VACIA = (
+    "Error: La Clave de Cliente/Proveedor está vacía y no se pueden hacer las validaciones."
+)
+MSG_CLAVE_VACIA_SOLUCION = (
+    "Capturar una Clave de Cliente/Proveedor nueva o existente a la cual desee agregar, remplazar o actualizar campos."
+)
+MSG_CLAVE_NO_EXISTE = "Error: (Col.C) Clave de Proveedor/Cliente no existe."
+MSG_CLAVE_NO_EXISTE_SOLUCION = "La clave debe existir en el catálogo cuando el modo es Actualizar."
 
 
-def validate_row_client_provider_lengths(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
+def validate_row_desfase(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
+    """Si COL_EXTRA tiene valor → advertencia de desfase."""
+    return check_desfase(row, line_num)
+
+
+def validate_row_clave_vacia(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
+    """Clave (SHORT_NAME) vacía → error inmediato."""
+    val = (row.get("SHORT_NAME") or "").strip()
+    if not val:
+        return {
+            "line": line_num,
+            "col": "SHORT_NAME",
+            "msg": MSG_CLAVE_VACIA,
+            "solution": MSG_CLAVE_VACIA_SOLUCION,
+        }
+    return None
+
+
+def validaciones_cliente_o_prov(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
+    """
+    VALIDACIONES_CLIENTE_O_PROV: reglas de dominio compartidas.
+    Col A (E/N + coherencia con L), B (C/P/A), C (máx 8), Q/R/S, T/U, V, W/X, AF.
+    """
+    err = check_procedencia(row, line_num)
+    if err:
+        return err
+    err = check_tipo_client_provider(row, line_num)
+    if err:
+        return err
+    err = check_short_name_max_clarion(row, line_num)
+    if err:
+        return err
+    err = check_tipo_programa_secon(row, line_num)
+    if err:
+        return err
+    err = check_es_prosec_num_aut(row, line_num)
+    if err:
+        return err
+    err = check_vinculacion(row, line_num)
+    if err:
+        return err
+    err = check_es_empresa_certificada_registro(row, line_num)
+    if err:
+        return err
+    err = check_transformador_submaq(row, line_num)
+    if err:
+        return err
+    return None
+
+
+def valida_toda_cliente_o_prov(
+    row: Dict[str, Any],
+    line_num: int,
+    actualizar: bool = False,
+) -> Optional[Dict[str, Any]]:
+    """
+    VALIDA_TODA: obligatorios Col A (Procedencia), Col D (Nombre) cuando no es ACT.
+    Si modo ACT y clave no existe → error se devuelve antes (en validate_row_client_provider).
+    Luego ejecuta VALIDACIONES_CLIENTE_O_PROV.
+    """
+    campos_oblig = []
+    # Col A - Procedencia obligatoria
+    if not (row.get("PROCEDENCIA") or "").strip():
+        campos_oblig.append("(Col.A) Tipo Cliente Procedencia (E/N)")
+    # Col D - Nombre obligatorio solo cuando no es actualizar
+    if not actualizar and not (row.get("NOMBRE") or "").strip():
+        campos_oblig.append("(Col.D) Nombre")
+    if campos_oblig:
+        return {
+            "line": line_num,
+            "col": "PROCEDENCIA" if not (row.get("PROCEDENCIA") or "").strip() else "NOMBRE",
+            "msg": MSG_CAMPOS_OBLIGATORIOS.format(campos=", ".join(campos_oblig)),
+            "solution": MSG_CAMPOS_OBLIGATORIOS_SOLUCION,
+        }
+    return validaciones_cliente_o_prov(row, line_num)
+
+
+def valida_parcial_cliente_o_prov(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
+    """VALIDA_PARCIAL: solo VALIDACIONES_CLIENTE_O_PROV (no exige A ni D)."""
+    return validaciones_cliente_o_prov(row, line_num)
+
+
+def validate_row_client_provider(
+    row: Dict[str, Any],
+    line_num: int,
+    actualizar: bool = False,
+    existing_short_names: Optional[Set[str]] = None,
+) -> Optional[Dict[str, Any]]:
+    """
+    Valida una fila de CSV de clientes y proveedores.
+    - Desfase (COL_EXTRA) primero.
+    - Clave vacía → error.
+    - Si actualizar y clave no existe en existing_short_names → error "Clave no existe".
+    - Si actualizar y clave existe → VALIDA_PARCIAL (solo validaciones comunes).
+    - Si no actualizar o clave no existe → VALIDA_TODA (A y D obligatorios cuando aplique, luego comunes).
+    Además se validan RFC (requerido max 30), longitudes NOMBRE/SHORT_NAME/CURP para compatibilidad.
+    """
+    err = validate_row_desfase(row, line_num)
+    if err:
+        return err
+    err = validate_row_clave_vacia(row, line_num)
+    if err:
+        return err
+
+    short_name = (row.get("SHORT_NAME") or "").strip()
+    existing = existing_short_names or set()
+    use_partial = actualizar and short_name in existing
+
+    if actualizar and short_name and short_name not in existing:
+        return {
+            "line": line_num,
+            "col": "SHORT_NAME",
+            "msg": MSG_CLAVE_NO_EXISTE,
+            "solution": MSG_CLAVE_NO_EXISTE_SOLUCION,
+        }
+
+    if use_partial:
+        err = valida_parcial_cliente_o_prov(row, line_num)
+    else:
+        err = valida_toda_cliente_o_prov(row, line_num, actualizar=actualizar)
+    if err:
+        return err
+
+    # Compatibilidad: RFC requerido y longitudes (como antes)
+    err = check_required_max(row, "RFC", RFC_MAX, line_num)
+    if err:
+        return err
     err = check_max_length(row, "NOMBRE", NAME_MAX, line_num)
     if err:
         return err
-    err = check_max_length(row, "SHORT_NAME", SHORT_NAME_MAX, line_num)
+    err = check_max_length(row, "SHORT_NAME", 10, line_num)  # modelo permite 10
     if err:
         return err
     err = check_max_length(row, "CURP", CURP_MAX, line_num)
     if err:
         return err
     return None
-
-
-def validate_row_client_provider_tipo(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
-    return check_tipo_client_provider(row, line_num)
-
-
-def validate_row_client_provider(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
-    """
-    Valida una fila de CSV de clientes y proveedores.
-    RFC requerido (max 30); TIPO opcional pero debe ser Cliente/Proveedor/Ambos; NOMBRE/SHORT_NAME/CURP longitudes.
-    """
-    err = validate_row_client_provider_required(row, line_num)
-    if err:
-        return err
-    err = validate_row_client_provider_tipo(row, line_num)
-    if err:
-        return err
-    err = validate_row_client_provider_lengths(row, line_num)
-    if err:
-        return err
-    return None