From 8ac071aac9693ce62518a4fc107e60852490adb3 Mon Sep 17 00:00:00 2001 From: hreyes Date: Thu, 5 Mar 2026 15:05:32 -0700 Subject: [PATCH] feature/validaciones-clarion-cliente-proveedor --- .gitignore | 1 + .../common/common_validators.py | 228 ++++++++++++++++- .../clients_and_providers/common/mappers.py | 135 +++++++++- .../clients_and_providers/tasks.py | 233 ++++++++++++++++-- .../clients_and_providers/template_config.py | 78 +++++- .../clients_and_providers/test_validators.py | 143 +++++++++++ .../validators/common.py | 179 ++++++++++++-- 7 files changed, 920 insertions(+), 77 deletions(-) create mode 100644 backend/api/v1/modules/a76/layouts_csv/clients_and_providers/test_validators.py diff --git a/.gitignore b/.gitignore index 6aad309e..539bbaeb 100644 --- a/.gitignore +++ b/.gitignore @@ -30,6 +30,7 @@ wheels/ backend/.env frontend/.env backend/SCRIPTS/ +.cursor/ # IDEs .vscode/ diff --git a/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/common/common_validators.py b/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/common/common_validators.py index 801c24ab..42f41ae9 100644 --- a/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/common/common_validators.py +++ b/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 diff --git a/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/common/mappers.py b/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/common/mappers.py index 271aa04c..c281f814 100644 --- a/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/common/mappers.py +++ b/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) diff --git a/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/tasks.py b/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/tasks.py index 521244b3..603924e7 100644 --- a/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/tasks.py +++ b/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, diff --git a/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/template_config.py b/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/template_config.py index f0968d9d..be387efc 100644 --- a/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/template_config.py +++ b/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": []}, diff --git a/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/test_validators.py b/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/test_validators.py new file mode 100644 index 00000000..5ee1e46e --- /dev/null +++ b/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" diff --git a/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/validators/common.py b/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/validators/common.py index 91c8c3dd..12b311a6 100644 --- a/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/validators/common.py +++ b/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