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

feature/csv-correccion-clientes

Browse Source
This commit is contained in:
hreyes
2026-05-12 09:22:30 -06:00
parent ee9ac4f64f
commit e3cbb738e0
8 changed files with 105 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
backend/api/v1/modules/a76/clients_and_providers/routes.py
View File

@@ -7,7 +7,7 @@ from typing import List, Optional
from core.database import get_core_db
from core.security import get_current_user, validate_access_to_resource
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import or_
from sqlalchemy import func, or_
from sqlalchemy.orm import Session, joinedload
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
from .models import ClientOrProviderEnum
@@ -35,6 +35,9 @@ async def get_clients_and_providers(
company_id: int = Query(..., description="Company ID"),
name: Optional[str] = Query(None, description="Filter by name (contains)"),
rfc: Optional[str] = Query(None, description="Filter by RFC/TAX-ID (contains)"),
short_name: Optional[str] = Query(
None, description="Filter by short name / clave (exact match, case-insensitive)"
),
type: Optional[ClientOrProviderEnum] = Query(
None, description="Type of entity (client or provider)"
),
@@ -68,6 +71,10 @@ async def get_clients_and_providers(
if rfc:
query = query.filter(ClientProvider.rfc.ilike(f"%{rfc.strip()}%"))
if short_name:
sn = short_name.strip().upper()
query = query.filter(func.upper(ClientProvider.short_name) == sn)
if type is not None:
# Include 'both' type when filtering by client or provider
query = query.filter(

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

@@ -8,6 +8,12 @@ from typing import Dict, Any, Optional
from api.v1.modules.a76.clients_and_providers.models import ClientOrProviderEnum
def short_name_import_key(raw: Optional[str]) -> str:
"""Clave estable para upsert CSV y validación ACT (strip + upper)."""
return (raw or "").strip().upper()
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.

5
backend/api/v1/modules/a76/layouts_csv/clients_and_providers/routes.py
View File

@@ -91,6 +91,10 @@ def _get_redis():
async def upload_import_file(
file: UploadFile = File(...),
company_id: int = Query(..., description="Company ID"),
actualizar: bool = Query(
False,
description="Modo Agregar/Actualizar (ACT); si True, validación parcial para claves existentes en catálogo",
),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
@@ -114,6 +118,7 @@ async def upload_import_file(
"company_id": company_id,
"user_id": current_user.get("id"),
"template_id": "client_providers",
"actualizar": actualizar,
}
try:

182
backend/api/v1/modules/a76/layouts_csv/clients_and_providers/tasks.py
View File

@@ -19,6 +19,7 @@ from ..common import csv_reader as common_csv_reader
from .template_config import row_from_template
from .validators import validate_row_client_provider
from .common.mappers import row_to_client_provider_data
from .common.common_validators import short_name_import_key
logger = logging.getLogger(__name__)
@@ -64,8 +65,9 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str,
)
.all()
):
if (cp.short_name or "").strip():
existing_short_names.add((cp.short_name or "").strip())
sn = short_name_import_key(cp.short_name)
if sn:
existing_short_names.add(sn)
except Exception as e:
logger.warning("CP import: could not load existing short_names for ACT: %s", e)
@@ -159,8 +161,9 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
)
.all()
):
if (cp.short_name or "").strip():
existing_short_names.add((cp.short_name or "").strip())
sn = short_name_import_key(cp.short_name)
if sn:
existing_short_names.add(sn)
except Exception as e:
logger.warning("CP commit: could not load existing short_names for ACT: %s", e)
@@ -178,7 +181,6 @@ 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)
@@ -188,10 +190,7 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
)
.all()
):
rfc_key = (cp.rfc or "").strip()
if rfc_key:
existing_by_rfc[rfc_key] = cp
sn_key = (cp.short_name or "").strip()
sn_key = short_name_import_key(cp.short_name)
if sn_key:
existing_by_short_name[sn_key] = cp
@@ -227,30 +226,22 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
}
)
continue
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",
"solution": "Capturar en la columna SHORT_NAME la clave requerida para 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",
"solution": "Capturar en la columna SHORT_NAME una clave que exista en el catálogo.",
}
)
continue
# Merge: fill from existing when csv value is empty
sn_key = short_name_import_key(cp_data.get("short_name"))
if not sn_key:
skipped_invalid += 1
skipped_details.append(
{
"line": i,
"reason": "Clave (SHORT_NAME) requerida",
"solution": "Capturar en la columna SHORT_NAME la clave del cliente/proveedor.",
}
)
continue
cp_data["short_name"] = sn_key
existing = existing_by_short_name.get(sn_key)
if existing:
for k, v in cp_data.items():
if k in ("tenant_id", "company_id"):
continue
@@ -263,7 +254,6 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
setattr(existing, k, v)
session.add(existing)
updated_count += 1
# Update address if present
if address_data and existing.address:
addr = existing.address
for k, v in address_data.items():
@@ -288,7 +278,6 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
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,
@@ -307,100 +296,37 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
)
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",
"solution": "Capturar en la columna RFC el RFC requerido.",
}
new_cp = ClientProvider(**cp_data)
session.add(new_cp)
session.flush()
existing_by_short_name[sn_key] = 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"),
)
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)
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()

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

@@ -24,6 +24,7 @@ from ..common.common_validators import (
check_es_empresa_certificada_registro,
check_transformador_submaq,
check_desfase,
short_name_import_key,
)
@@ -36,8 +37,6 @@ MSG_CLAVE_VACIA = (
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_desfase(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
@@ -91,7 +90,6 @@ def valida_toda_cliente_o_prov(
) -> List[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 = []
@@ -129,9 +127,8 @@ def validate_row_client_provider(
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).
- Si actualizar y clave existe en catálogo → VALIDA_PARCIAL (solo validaciones comunes).
- Si no actualizar, o actualizar con clave nueva → 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.
"""
errors: List[Dict[str, Any]] = []
@@ -143,17 +140,9 @@ def validate_row_client_provider(
if err:
errors.append(err)
short_name = (row.get("SHORT_NAME") or "").strip()
sn_key = short_name_import_key(row.get("SHORT_NAME"))
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:
errors.append({
"line": line_num,
"col": "SHORT_NAME",
"msg": MSG_CLAVE_NO_EXISTE,
"solution": MSG_CLAVE_NO_EXISTE_SOLUCION,
})
use_partial = actualizar and sn_key in existing
if use_partial:
errors.extend(valida_parcial_cliente_o_prov(row, line_num))