feature/separacion-de-tax-id-de-rfc

This commit is contained in:
2026-06-02 09:05:29 -06:00
parent 08f99f102e
commit 619ef8d7d8
32 changed files with 358 additions and 163 deletions

View File

@@ -40,32 +40,32 @@ def check_required_max(row: Dict[str, Any], col: str, max_len: int, line_num: in
return None
def check_rfc_format(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Col E (RFC): si tiene valor, debe cumplir formato RFC México."""
val = (row.get("RFC") or "").strip()
def check_rfc_format(row: Dict[str, Any], line_num: int, col: str = "RFC") -> Optional[Dict[str, Any]]:
"""Columna RFC: si tiene valor, debe cumplir formato RFC México."""
val = (row.get(col) or "").strip()
if not val:
return None
if not RFC_PATTERN.match(val.upper()):
return {
"line": line_num,
"col": "RFC",
"col": col,
"msg": "El formato del RFC es inválido.",
"solution": "Capturar en la columna E un RFC con formato válido (ej. XAXX010101000).",
"solution": "Capturar un RFC con formato válido (ej. XAXX010101000).",
}
return None
def check_tax_id_format(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Col E (TAX-ID): 2 dígitos, guión y resto alfanumérico. Ej: 12-3456789. Máx 30 caracteres."""
val = (row.get("RFC") or "").strip()
def check_tax_id_format(row: Dict[str, Any], line_num: int, col: str = "TAX_ID") -> Optional[Dict[str, Any]]:
"""Columna TAX-ID: 2 dígitos, guión y resto alfanumérico. Ej: 12-3456789. Máx 30 caracteres."""
val = (row.get(col) or "").strip()
if not val:
return None
if len(val) > 30 or not TAX_ID_PATTERN.match(val):
return {
"line": line_num,
"col": "RFC",
"col": col,
"msg": "El formato del TAX-ID es inválido.",
"solution": "Capturar en la columna E un TAX-ID con formato: 2 dígitos, guión y resto (ej. 12-3456789). Máx 30 caracteres.",
"solution": "Capturar un TAX-ID con formato: 2 dígitos, guión y resto (ej. 12-3456789). Máx 30 caracteres.",
}
return None

View File

@@ -10,6 +10,7 @@ CLIENT_PROVIDER_CSV_COLUMN_MAX_CHARS: Dict[str, int] = {
"SHORT_NAME": 10,
"NOMBRE": 256,
"RFC": 30,
"TAX_ID": 30,
"DIRECCION": 100,
"NUM_EXT": 20,
"CODIGO POSTAL": 15,

View File

@@ -9,6 +9,7 @@ El CSV se alimenta en base a las tablas/modelos:
Las validaciones (validators) aplican reglas de negocio Clarion y respetan longitudes máximas de los modelos.
"""
import logging
from datetime import datetime
from decimal import Decimal
from typing import Dict, Any, Optional, Tuple
@@ -20,8 +21,11 @@ from .common_validators import (
parse_active,
)
logger = logging.getLogger(__name__)
MAX_LEN = {
"rfc": 30,
"tax_id": 30,
"name": 256,
"short_name": 10,
"curp": 19,
@@ -84,21 +88,37 @@ def row_to_client_provider_data(
"""
Mapea fila normalizada a datos para ClientProvider, ClientProviderAddress y ClientProviderPrograms.
Devuelve (cp_data, address_data_or_none, programs_data_or_none).
Identificador fiscal unificado: solo se guarda en ClientProvider.rfc (RFC o TAX-ID según procedencia).
Se toma de columna RFC (E); si viene vacía y hay TAX_ID_PROGRAMS (AB), se usa esa para la misma columna rfc.
Identificadores fiscales separados: RFC (nacional) → cp.rfc; TAX-ID (extranjero) → cp.tax_id.
Compatibilidad: archivos legacy de columna única traían el TAX-ID en la columna RFC; si el registro es
extranjero y no hay columna TAX_ID dedicada, ese valor se rutea a tax_id (salvaguarda no silenciosa).
"""
rfc_col = _str_or_none(row_norm.get("RFC"), MAX_LEN["rfc"])
tax_id_programs_col = _str_or_none(row_norm.get("TAX_ID_PROGRAMS"), MAX_LEN["rfc"])
# Una sola columna: identificador fiscal en cp.rfc (nacional=RFC, extranjero=TAX-ID)
rfc_unified = rfc_col or tax_id_programs_col
short_name = _str_or_none(row_norm.get("SHORT_NAME"), MAX_LEN["short_name"])
if not rfc_unified 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]
is_foreign = procedencia == "E"
rfc_col = _str_or_none(row_norm.get("RFC"), MAX_LEN["rfc"])
tax_id_col = _str_or_none(row_norm.get("TAX_ID"), MAX_LEN["tax_id"])
tax_id_programs_col = _str_or_none(row_norm.get("TAX_ID_PROGRAMS"), MAX_LEN["tax_id"])
rfc_final = rfc_col
tax_id_final = tax_id_col
if is_foreign and not tax_id_final:
# Salvaguarda legacy: extranjero sin columna TAX_ID dedicada → el TAX-ID estaba en RFC (o AB).
if rfc_col:
tax_id_final = rfc_col
rfc_final = None
logger.info(
"CSV clients_providers: TAX-ID tomado de columna RFC (archivo legacy de columna única)"
)
elif tax_id_programs_col:
tax_id_final = tax_id_programs_col
short_name = _str_or_none(row_norm.get("SHORT_NAME"), MAX_LEN["short_name"])
if not rfc_final and not tax_id_final and not short_name:
return ({}, None, None)
client_or_provider = parse_client_or_provider(row_norm.get("TIPO")) or ClientOrProviderEnum.BOTH
# Vinculación 0/1/2 → string
vinc = (row_norm.get("VINCULACION") or "").strip()
@@ -113,7 +133,8 @@ def row_to_client_provider_data(
cp_data = {
"tenant_id": tenant_id,
"company_id": company_id,
"rfc": rfc_unified or None,
"rfc": rfc_final or None,
"tax_id": tax_id_final or None,
"name": _str_or_none(row_norm.get("NOMBRE"), MAX_LEN["name"]),
"short_name": short_name,
"curp": _str_or_none(row_norm.get("CURP"), MAX_LEN["curp"]),

View File

@@ -221,8 +221,8 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
skipped_details.append(
{
"line": i,
"reason": "Fila sin RFC ni Clave",
"solution": "Capturar en el CSV el RFC (columna RFC) o la clave del cliente/proveedor (columna SHORT_NAME).",
"reason": "Fila sin RFC/TAX-ID ni Clave",
"solution": "Capturar en el CSV el RFC (columna RFC), el TAX-ID (columna TAX_ID) o la clave del cliente/proveedor (columna SHORT_NAME).",
}
)
continue

View File

@@ -1,7 +1,9 @@
"""
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, FAG.
Solo se leen columnas definidas aquí; el resto se ignora.
Solo se leen columnas definidas aquí; el resto se ignora. El parseo es por nombre de cabecera (no posicional).
RFC y TAX-ID viven en columnas separadas: "RFC" (nacional) y "TAX_ID" (identificador fiscal extranjero).
Compatibilidad: archivos legacy de columna única (TAX-ID en "RFC") se rutean a tax_id según procedencia (ver mappers).
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,
@@ -38,8 +40,10 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
{"canonical": "SHORT_NAME", "aliases": ["CLAVE", "CLAVE CORTA", "NOMBRE CORTO", "SIGLAS"]},
# 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 E - RFC (solo RFC mexicano / nacional)
{"canonical": "RFC", "aliases": ["RFC NACIONAL"]},
# Col E2 - TAX-ID dedicado (identificador fiscal extranjero), separado del RFC
{"canonical": "TAX_ID", "aliases": ["TAXID", "TAX-ID", "IDENTIFICADOR FISCAL", "IDENTIFICACION FISCAL"]},
# Col F - Calles
{"canonical": "DIRECCION", "aliases": ["CALLES", "DOMICILIO", "DIRECCION FISCAL", "CALLE"]},
# Col G - Número exterior
@@ -84,8 +88,8 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
{"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 AB - Tax ID (programas). Alias "TAX ID" se quitó para no colisionar con la columna TAX_ID dedicada.
{"canonical": "TAX_ID_PROGRAMS", "aliases": ["TAXID PROGRAMS"]},
# Col AC - Broker exportación
{"canonical": "BROKER_EXPO", "aliases": ["BROKER EXPO", "BROKER EXPORTACION"]},
# Col AD - Broker importación

View File

@@ -149,32 +149,40 @@ def validate_row_client_provider(
else:
errors.extend(valida_toda_cliente_o_prov(row, line_num, actualizar=actualizar))
# Identificador fiscal requerido: Nacional = RFC, Extranjero = TAX-ID (mismo campo "RFC" en layout).
# Identificador fiscal requerido y validado por procedencia:
# Nacional (N) -> RFC en columna RFC
# Extranjero (E) -> TAX-ID en columna TAX_ID (compat: archivos legacy lo traían en columna RFC)
procedencia = (row.get("PROCEDENCIA") or "").strip().upper()[:1]
err = check_required_max(row, "RFC", RFC_MAX, line_num)
if err:
if procedencia == "E":
err = {
if procedencia == "E":
# Columna efectiva: TAX_ID dedicada o, en su defecto, RFC (archivo legacy de columna única).
effective_col = "TAX_ID" if (row.get("TAX_ID") or "").strip() else "RFC"
if not (row.get(effective_col) or "").strip():
errors.append({
"line": line_num,
"col": "RFC",
"col": "TAX_ID",
"msg": "Requerido (TAX-ID)",
"solution": "Capturar el TAX-ID del cliente/proveedor extranjero en la columna E (RFC/TAX-ID).",
}
errors.append(err)
"solution": "Capturar el TAX-ID del cliente/proveedor extranjero en la columna TAX_ID.",
})
else:
err = check_tax_id_format(row, line_num, col=effective_col)
if err:
errors.append(err)
else:
# Validar formato según procedencia: RFC (N) o TAX-ID (E).
if procedencia == "N":
err = check_rfc_format(row, line_num)
elif procedencia == "E":
err = check_tax_id_format(row, line_num)
# Nacional: RFC requerido + formato
err = check_required_max(row, "RFC", RFC_MAX, line_num)
if err:
errors.append(err)
else:
err = check_rfc_format(row, line_num)
if err:
errors.append(err)
for err in [
check_max_length(row, "NOMBRE", NAME_MAX, line_num),
check_max_length(row, "SHORT_NAME", 10, line_num), # modelo permite 10
check_max_length(row, "CURP", CURP_MAX, line_num),
check_max_length(row, "TAX_ID", RFC_MAX, line_num),
]:
if err:
errors.append(err)