ADUANASOFT/plantillas-proyectos
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature/client-provider-taxid-rfc-fix

Browse Source
This commit is contained in:
hreyes
2026-03-17 15:08:02 -06:00
parent 5a619e8bc4
commit e326ac0f0a
24 changed files with 519 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
backend/api/v1/modules/a76/clients_and_providers/dto.py
View File

@@ -6,7 +6,9 @@ Reemplaza schemas.py siguiendo enfoque DDD y estilo NestJS
from decimal import Decimal
from typing import List, Literal, Optional
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, model_validator
from .validators import is_valid_rfc, is_valid_tax_id
# DTOs para dirección
@@ -54,7 +56,6 @@ class ClientProviderProgramsDTO(BaseModel):
manufacturer_id: Optional[str] = Field(
None, max_length=25, description="Manufacturer ID"
)
tax_id: Optional[str] = Field(None, max_length=30, description="Tax ID")
broker: Optional[str] = Field(None, max_length=6, description="Broker")
import_broker: Optional[str] = Field(
None, max_length=6, description="Import broker"
@@ -116,7 +117,6 @@ class ClientProviderCreateDTO(BaseModel):
)
position: Optional[str] = Field(None, max_length=30, description="Position")
incoterm: Optional[str] = Field(None, max_length=19, description="Incoterm")
is_national_provider: Optional[bool] = None
is_active: Optional[bool] = Field(None, description="Enabled/Disabled status")
# Nested DTOs
@@ -127,6 +127,24 @@ class ClientProviderCreateDTO(BaseModel):
None, description="Programs information"
)
@model_validator(mode="after")
def validate_rfc_or_tax_id_format(self):
rfc = self.rfc
if not rfc or not (rfc := (rfc or "").strip()):
return self
proc = (self.type_nat_foreign or "N").strip().upper()[:1]
if proc == "E":
if not is_valid_tax_id(rfc):
raise ValueError(
"El TAX-ID debe tener formato: 2 dígitos, guión y resto (ej. 12-3456789). Máx 30 caracteres."
)
else:
if not is_valid_rfc(rfc):
raise ValueError(
"El RFC no tiene el formato correcto. Ejemplo: XAXX010101000."
)
return self
class Config:
from_attributes = True
@@ -157,7 +175,6 @@ class ClientProviderUpdateDTO(BaseModel):
)
position: Optional[str] = Field(None, max_length=30, description="Position")
incoterm: Optional[str] = Field(None, max_length=19, description="Incoterm")
is_national_provider: Optional[bool] = None
is_active: Optional[bool] = Field(None, description="Enabled/Disabled status")
# Nested DTOs
@@ -168,6 +185,24 @@ class ClientProviderUpdateDTO(BaseModel):
None, description="Programs information"
)
@model_validator(mode="after")
def validate_rfc_or_tax_id_format(self):
rfc = self.rfc
if not rfc or not (rfc := (rfc or "").strip()):
return self
proc = (self.type_nat_foreign or "N").strip().upper()[:1]
if proc == "E":
if not is_valid_tax_id(rfc):
raise ValueError(
"El TAX-ID debe tener formato: 2 dígitos, guión y resto (ej. 12-3456789). Máx 30 caracteres."
)
else:
if not is_valid_rfc(rfc):
raise ValueError(
"El RFC no tiene el formato correcto. Ejemplo: XAXX010101000."
)
return self
class Config:
from_attributes = True
@@ -189,7 +224,6 @@ class ClientProviderResponseDTO(BaseModel):
responsible: Optional[str] = None
position: Optional[str] = None
incoterm: Optional[str] = None
is_national_provider: Optional[bool] = None
is_active: Optional[bool] = None
tenant_id: int
company_id: int

3
backend/api/v1/modules/a76/clients_and_providers/models.py
View File

@@ -45,6 +45,7 @@ class ClientProvider(Base, TenantScopedMixin, TimestampMixin):
type_nat_foreign: Mapped[Optional[str]] = mapped_column(String(1)) # TIPO NACIONAL/EXTRANJERO
name: Mapped[Optional[str]] = mapped_column(String(256))
short_name: Mapped[Optional[str]] = mapped_column(String(10))
# Identificador fiscal único: RFC (nacional) o TAX-ID (extranjero); no usar programs.tax_id para lo mismo
rfc: Mapped[Optional[str]] = mapped_column(String(30))
curp: Mapped[Optional[str]] = mapped_column(String(19))
client_or_provider: Mapped[ClientOrProviderEnum] = mapped_column(PgEnum(ClientOrProviderEnum, name="entity_client_or_provider", create_type=True, native_enum=True),nullable=False)
@@ -55,7 +56,6 @@ class ClientProvider(Base, TenantScopedMixin, TimestampMixin):
responsible: Mapped[Optional[str]] = mapped_column(String(80))
position: Mapped[Optional[str]] = mapped_column(String(30))
incoterm: Mapped[Optional[str]] = mapped_column(String(19))
is_national_provider: Mapped[Optional[bool]] = mapped_column(Boolean)
is_active: Mapped[Optional[bool]] = mapped_column(Boolean)
# Relationships
@@ -140,7 +140,6 @@ class ClientProviderPrograms(Base, TenantScopedMixin, TimestampMixin):
prosec_authorization: Mapped[Optional[str]] = mapped_column(String(20))
secon_auth_date: Mapped[Optional[int]] = mapped_column(Integer)
manufacturer_id: Mapped[Optional[str]] = mapped_column(String(25))
tax_id: Mapped[Optional[str]] = mapped_column(String(30))
broker: Mapped[Optional[str]] = mapped_column(String(6))
import_broker: Mapped[Optional[str]] = mapped_column(String(6))
transfer_key: Mapped[Optional[str]] = mapped_column(String(8))

27
backend/api/v1/modules/a76/clients_and_providers/validators.py Normal file
View File

@@ -0,0 +1,27 @@
"""
Validadores de formato para RFC (Nacional) y TAX-ID (Extranjero) en clientes y proveedores.
"""
import re
# Formato RFC México: 3-4 letras (A-Z, &, Ñ), 6 dígitos (fecha), 3 caracteres homoclave. Ej: XAXX010101000
RFC_PATTERN = re.compile(r"^[A-Z&Ñ]{3,4}\d{6}[A-Z0-9]{3}$", re.IGNORECASE)
# TAX-ID extranjero: 2 dígitos, guión, resto alfanumérico. Ej: 12-3456789 (EIN US). Total máx 30.
TAX_ID_PATTERN = re.compile(r"^\d{2}-[A-Z0-9]{1,27}$", re.IGNORECASE)
def is_valid_rfc(value: str) -> bool:
"""Valida formato RFC mexicano. Acepta cadena vacía/None como inválida (no opcional aquí)."""
if not value or not isinstance(value, str):
return False
normalized = value.strip().upper()
return bool(normalized and RFC_PATTERN.match(normalized))
def is_valid_tax_id(value: str) -> bool:
"""Valida formato TAX-ID (extranjero): 2 dígitos, guión y resto alfanumérico. Ej: 12-3456789."""
if not value or not isinstance(value, str):
return False
normalized = value.strip()
return bool(normalized and len(normalized) <= 30 and TAX_ID_PATTERN.match(normalized))

27
backend/api/v1/modules/a76/layouts_csv/classes/template_config.py
View File

@@ -12,6 +12,22 @@ from ..common.cell_value import cell_to_str
# Valores que indican que la primera fila es cabecera (primera columna normalizada)
FIRST_COLUMN_HEADER_VALUES = ("CLAVE CLASE", "CLASE")
_ENCODING_FALLBACKS: Tuple[str, ...] = ("utf-8-sig", "utf-8", "cp1252", "latin-1")
def _read_text_sample(file_path: str, sample_bytes: int = 2048) -> str:
with open(file_path, "rb") as f:
raw = f.read(sample_bytes)
last_err: Optional[Exception] = None
for enc in _ENCODING_FALLBACKS:
try:
return raw.decode(enc)
except Exception as e:
last_err = e
if last_err:
raise last_err
return ""
def detect_headers_or_data(
file_path: str,
@@ -26,8 +42,15 @@ def detect_headers_or_data(
- Si no -> has_header=False, fieldnames=TEMPLATE_DOWNLOAD_HEADERS (la primera fila es dato).
"""
try:
with open(file_path, "r", encoding=encoding) as f:
sample = f.read(2048)
# `encoding` se mantiene por compatibilidad; si falla, hacemos fallback para CSVs tipo Excel (cp1252/latin-1).
if encoding and encoding.lower() not in ("auto", "detect"):
try:
with open(file_path, "r", encoding=encoding) as f:
sample = f.read(2048)
except Exception:
sample = _read_text_sample(file_path, sample_bytes=2048)
else:
sample = _read_text_sample(file_path, sample_bytes=2048)
except Exception:
return None, True
lines = sample.splitlines()

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

@@ -3,11 +3,16 @@ 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.
"""
import re
from typing import Dict, Any, Optional
from api.v1.modules.a76.clients_and_providers.models import ClientOrProviderEnum
RFC_MAX = 30
# Formato RFC México: 3-4 letras, 6 dígitos, 3 homoclave. TAX-ID: 2 dígitos, guión, resto. Ej: 12-3456789.
RFC_PATTERN = re.compile(r"^[A-Z&Ñ]{3,4}\d{6}[A-Z0-9]{3}$", re.IGNORECASE)
TAX_ID_PATTERN = re.compile(r"^\d{2}-[A-Z0-9]{1,27}$", re.IGNORECASE)
NAME_MAX = 256
SHORT_NAME_MAX = 10
# Clarion: Col C máx 8 caracteres
@@ -29,6 +34,36 @@ 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()
if not val:
return None
if not RFC_PATTERN.match(val.upper()):
return {
"line": line_num,
"col": "RFC",
"msg": "El formato del RFC es inválido.",
"solution": "Capturar en la columna E 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()
if not val:
return None
if len(val) > 30 or not TAX_ID_PATTERN.match(val):
return {
"line": line_num,
"col": "RFC",
"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.",
}
return None
def check_max_length(row: Dict[str, Any], col: str, max_len: int, line_num: int) -> Optional[Dict[str, Any]]:
val = (row.get(col) or "").strip()
if not val:

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

@@ -43,7 +43,6 @@ MAX_LEN = {
"prosec_authorization": 20,
"secon_authorization": 20,
"manufacturer_id": 25,
"tax_id_programs": 30,
"broker": 6,
"import_broker": 6,
"transfer_key": 8,
@@ -85,11 +84,15 @@ 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).
Para compatibilidad con Clarion: se requiere RFC o SHORT_NAME para considerar la fila válida.
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.
"""
rfc = _str_or_none(row_norm.get("RFC"), MAX_LEN["rfc"])
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 and not 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
@@ -110,7 +113,7 @@ def row_to_client_provider_data(
cp_data = {
"tenant_id": tenant_id,
"company_id": company_id,
"rfc": rfc or None,
"rfc": rfc_unified 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"]),
@@ -124,7 +127,6 @@ def row_to_client_provider_data(
"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
@@ -174,12 +176,12 @@ def row_to_client_provider_data(
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
):
# Identificador fiscal solo en ClientProvider.rfc (columna única)
programs_data = {
"program": tipo_prog[:7] if tipo_prog else None,
"program_number": num_prog,
@@ -190,7 +192,6 @@ def row_to_client_provider_data(
"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"]),

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

@@ -13,6 +13,8 @@ from ..common.common_validators import (
CURP_MAX,
check_required_max,
check_max_length,
check_rfc_format,
check_tax_id_format,
check_tipo_client_provider,
check_procedencia,
check_short_name_max_clarion,
@@ -161,8 +163,23 @@ def validate_row_client_provider(
if err:
return err
# Compatibilidad: RFC requerido y longitudes (como antes)
# Identificador fiscal requerido: Nacional = RFC, Extranjero = TAX-ID (mismo campo "RFC" en layout).
procedencia = (row.get("PROCEDENCIA") or "").strip().upper()[:1]
err = check_required_max(row, "RFC", RFC_MAX, line_num)
if err:
if procedencia == "E":
err = {
"line": line_num,
"col": "RFC",
"msg": "Requerido (TAX-ID)",
"solution": "Capturar el TAX-ID del cliente/proveedor extranjero en la columna E (RFC/TAX-ID).",
}
return err
# 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)
if err:
return err
err = check_max_length(row, "NOMBRE", NAME_MAX, line_num)

34
backend/api/v1/modules/a76/layouts_csv/common/csv_reader.py
View File

@@ -5,7 +5,7 @@ Si se pasa headerless_first_cell_values, se detecta si la primera fila es cabece
"""
import csv
import io
from typing import Iterator, Tuple, Dict, Any, Optional, List, Set
from typing import Iterator, Tuple, Dict, Any, Optional, List, Set, Sequence
def _normalize_empty_headers(headers: List[str]) -> List[str]:
@@ -21,6 +21,32 @@ def _normalize_empty_headers(headers: List[str]) -> List[str]:
return result
_ENCODING_FALLBACKS: Sequence[str] = ("utf-8-sig", "utf-8", "cp1252", "latin-1")
def _detect_text_encoding(
file_path: str,
encodings: Sequence[str] = _ENCODING_FALLBACKS,
sample_bytes: int = 8192,
) -> str:
"""
Detecta encoding por prueba de decodificación en un sample en binario.
Nota: latin-1 decodifica cualquier byte; por eso debe ir al final.
"""
with open(file_path, "rb") as f:
raw = f.read(sample_bytes)
last_err: Optional[Exception] = None
for enc in encodings:
try:
raw.decode(enc)
return enc
except Exception as e:
last_err = e
if last_err:
raise last_err
return "utf-8-sig"
def iter_csv_rows(
file_path: str,
fieldnames: Optional[List[str]] = None,
@@ -34,7 +60,8 @@ def iter_csv_rows(
(quitando BOM, strip, upper) está en headerless_first_cell_values, se trata como dato y se usan fieldnames.
headerless_second_cell_key_pattern se ignora si no se usa (reservado para otros layouts).
"""
with open(file_path, "r", encoding="utf-8-sig") as f:
encoding = _detect_text_encoding(file_path)
with open(file_path, "r", encoding=encoding) as f:
sample = f.read(2048)
f.seek(0)
try:
@@ -91,6 +118,7 @@ def iter_csv_rows(
def count_csv_rows(file_path: str, has_header: bool = True) -> int:
"""Cuenta filas del CSV. Si has_header=True (por defecto), no cuenta la cabecera."""
with open(file_path, "r", encoding="utf-8-sig") as f:
encoding = _detect_text_encoding(file_path)
with open(file_path, "r", encoding=encoding) as f:
total_lines = sum(1 for _ in f)
return total_lines if not has_header else max(0, total_lines - 1)

8
backend/api/v1/modules/a76/reports/exportacion/aviso_consolidado/service.py
View File

@@ -136,12 +136,8 @@ class AvisoConsolidadoExportacionService:
if client_obj:
# Fetch Address
c_addr = db.query(ClientProviderAddress).filter(ClientProviderAddress.client_id == target_client_id).first()
# Fetch Fiscal Data (RFC)
c_prog = db.query(ClientProviderPrograms).filter(ClientProviderPrograms.client_id == target_client_id).first()
c_rfc = ""
if c_prog and c_prog.tax_id: c_rfc = c_prog.tax_id
elif hasattr(client_obj, 'rfc'): c_rfc = client_obj.rfc
# Identificador fiscal único en ClientProvider.rfc (RFC o TAX-ID)
c_rfc = getattr(client_obj, "rfc", "") or ""
c_dir_str = "DOMICILIO NO REGISTRADO"
if c_addr:

6
backend/api/v1/modules/a76/reports/importacion/consolidados/mex/service.py
View File

@@ -127,11 +127,7 @@ class ConsolidadoImportacionMexService:
ciudad=(addr.city or "") if addr else "",
estado=(addr.state or "") if addr else "",
pais=(addr.country or "MEX") if addr else "MEX",
tax_id=(
prog.tax_id
if (prog and prog.tax_id)
else (getattr(main, "rfc", "") or "")
),
tax_id=getattr(main, "rfc", "") or "",
programa="IMMEX" if (prog and prog.program) else "",
autorizacion=prog.program_number if prog else "",
prosec=(

2
backend/api/v1/modules/a76/reports/importacion/consolidados/temporary/mex/service.py
View File

@@ -82,7 +82,7 @@ class ConsolidadoImportacionMexService:
ciudad=(addr.city or "") if addr else "",
estado=(addr.state or "") if addr else "",
pais=(addr.country or "MEX") if addr else "MEX",
tax_id=prog.tax_id if (prog and prog.tax_id) else (getattr(main, 'rfc', "") or ""),
tax_id=getattr(main, "rfc", "") or "",
programa="IMMEX" if (prog and prog.program) else "",
autorizacion=prog.program_number if prog else "",
prosec=prog.prosec_authorization if (prog and prog.prosec and prog.prosec_authorization) else "",

6
backend/api/v1/modules/a76/reports/importacion/facturas/mex/service.py
View File

@@ -159,11 +159,7 @@ class FacturaImportacionMexService:
ciudad=(addr.city or "") if addr else "",
estado=(addr.state or "") if addr else "",
pais=(addr.country or "MEX") if addr else "MEX",
tax_id=(
prog.tax_id
if (prog and prog.tax_id)
else (getattr(main, "rfc", "") or "")
),
tax_id=getattr(main, "rfc", "") or "",
programa="IMMEX" if (prog and prog.program) else "",
autorizacion=prog.program_number if prog else "",
prosec=(

2
backend/api/v1/modules/a76/reports/importacion/facturas/temporary/mex/service.py
View File

@@ -82,7 +82,7 @@ class FacturaImportacionMexService:
ciudad=(addr.city or "") if addr else "",
estado=(addr.state or "") if addr else "",
pais=(addr.country or "MEX") if addr else "MEX",
tax_id=prog.tax_id if (prog and prog.tax_id) else (getattr(main, 'rfc', "") or ""),
tax_id=getattr(main, "rfc", "") or "",
programa="IMMEX" if (prog and prog.program) else "",
autorizacion=prog.program_number if prog else "",
prosec=prog.prosec_authorization if (prog and prog.prosec and prog.prosec_authorization) else "",

6
backend/api/v1/modules/a76/reports/importacion/facturas/usa/service.py
View File

@@ -153,11 +153,7 @@ class FacturaImportacionUsaService:
ciudad=(addr.city or "") if addr else "",
estado=(addr.state or "") if addr else "",
pais=(addr.country or "USA") if addr else "USA",
tax_id=(
prog.tax_id
if (prog and prog.tax_id)
else (getattr(main, "rfc", "") or "")
),
tax_id=getattr(main, "rfc", "") or "",
programa="IMMEX" if (prog and prog.program) else "",
autorizacion=prog.program_number if prog else "",
prosec=(

2
backend/api/v1/modules/a76/reports/importacion/packing_list/service.py
View File

@@ -103,7 +103,7 @@ class PackingListService:
ciudad=(addr.city or "") if addr else "",
estado=(addr.state or "") if addr else "",
pais=(addr.country or "MEX") if addr else "MEX",
tax_id=prog.tax_id if (prog and prog.tax_id) else (getattr(main, 'rfc', "") or ""),
tax_id=getattr(main, "rfc", "") or "",
programa="IMMEX" if (prog and prog.program) else "",
autorizacion=prog.program_number if prog else "",
prosec=prog.prosec_authorization if (prog and prog.prosec and prog.prosec_authorization) else "",

12
backend/api/v1/modules/a76/reports/movements/invoices/services/database_helpers.py
View File

@@ -108,22 +108,24 @@ class DatabaseHelper:
if not client_code:
return {"name": None, "rfc": None, "tax_id": None}
client_type = 'PROVIDER' if is_supplier else 'CLIENT'
# In our schema, enum values are lowercase: 'client', 'provider', 'both'
client_type = 'provider' if is_supplier else 'client'
try:
sql = text("""
SELECT cp.name, cp.rfc, cpp.tax_id
SELECT cp.name, cp.rfc
FROM a76.clients_and_providers cp
LEFT JOIN a76.clients_and_providers_programs cpp ON cpp.client_id = cp.id
WHERE cp.id = :client_code AND cp.client_or_provider = :client_type
WHERE cp.id = :client_code
AND (cp.client_or_provider = :client_type OR cp.client_or_provider = 'both')
""")
result = db.execute(sql, {"client_code": client_code, "client_type": client_type}).fetchone()
if result:
# Unified identifier: cp.rfc contains either RFC (national) or TAX-ID (foreign)
return {
"name": result[0],
"rfc": result[1],
"tax_id": result[2]
"tax_id": result[1],
}
else:
logger.debug(f"Client {client_code} not found as {client_type}")