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

feature/pedimentos-csv-validations

Browse Source
This commit is contained in:
hreyes
2026-03-05 10:15:15 -07:00
parent e486fd750f
commit 1f391aa4a6
16 changed files with 1272 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
backend/api/v1/modules/a76/layouts_csv/classes/TEST_DATA_README.md
View File

@@ -1,36 +0,0 @@
# Datos de prueba – Import CSV Clases de Materiales
## Archivo: `test_data_clases.csv`
### Cómo usar
1. **Catálogos necesarios**: El CSV usa códigos que deben existir en tu tenant/empresa:
- **TIPO DE MATERIAL**: p. ej. `MAT01` (catálogo Tipos de Activo Fijo / MaterialType).
- **U.M. COMERCIAL**: p. ej. `KG`, `MTR` (catálogo Unidades de Medida por tenant/company).
- **FRACCION ARANCELARIA**: mínimo 8 caracteres y que exista en Fracciones Arancelarias Sifr@ o Histórico (p. ej. `12345678`, `87654321`).
- **FRACCION AMERICANA**: que exista en catálogo Fracciones Americanas (p. ej. `1234567890123456`, `6543210987654321`).
Si no tienes esos códigos, crea al menos uno de cada catálogo o sustituye en el CSV por códigos reales de tu base.
2. **Filas válidas (para probar carga correcta)**
- Líneas 2 y 3: `CLASE01`, `CLASE02` — completas y correctas (ajusta tipos de material, U.M. y fracciones a tus catálogos).
3. **Filas que disparan errores (para probar mensajes)**
- **Línea 4**: Todas las celdas vacías → *"La columna de Clase esta vacio..."* (Col. A vacía).
- **Línea 5**: `CLASE04` sin B,D,E,F → *"Existen campos vacios que son obligatorios..."* (validación completa).
- **Línea 6**: Clase de más de 8 caracteres (`ABCDEFGHIJ`) → *"La Clase: ... supera la longitud de caracteres"*.
- **Línea 7**: Tipo de material `INVALIDO` (no en catálogo) → *"(Col. D) El Tipo de Activo Fijo: INVALIDO no existe..."*.
- **Línea 8**: U.M. `UMINE` (no en catálogo) → *"(Col. E) La Unidad de Medida Comercial: UMINE no existe..."*.
- **Línea 9**: Fracción mexicana `1234` (< 8 caracteres) → *"La Fraccion 1234 no alcanza la longitud de 8 caracteres"*.
- **Línea 10**: Fracción americana `INVALIDO` (no en catálogo) → *"(Col. G) La Fraccion Americana: INVALIDO no existe..."*.
- **Línea 11**: Tasa de depreciación `150` (> 100) → *"(Col. H) La Tasa de Depreciación: 150 no puede ser mayor al 100 %"*.
- **Línea 12**: Código producto CP `9999` (si existe catálogo CP y no está) → *"(Col. J) La clase: CLASE10 tiene asignado un código de producto inexistente"*.
### CSV sin cabecera (opcional)
Si quieres probar detección de “primera fila = datos”, usa un archivo cuya primera línea sea una fila de datos (no "CLAVE CLASE"...). El sistema usará `TEMPLATE_DOWNLOAD_HEADERS` como cabecera y la primera línea como dato.
### Valores mínimos para una fila válida
Con **Modo Actualizar** y clase ya existente, solo es obligatoria la Col. A (CLAVE CLASE).
Con **Modo Reemplazar** o clase nueva, son obligatorios: A, B, D, E, F (mínimo 8 caracteres y en catálogo), y el resto según reglas (H ≤ 100, J en catálogo CP si aplica).

14
backend/api/v1/modules/a76/layouts_csv/common/responses.py
View File

@@ -9,16 +9,24 @@ def scan_result(
processed_rows: int,
error_count: int,
errors_detail: List[Dict[str, Any]],
total_rows_in_file: Optional[int] = None,
message: Optional[str] = None,
) -> Dict[str, Any]:
"""Respuesta de scan_file (waiting_confirmation)."""
return {
"""Respuesta de scan_file (waiting_confirmation).
Si total_rows_in_file no se pasa, se usa processed_rows como total (comportamiento anterior).
"""
total = total_rows_in_file if total_rows_in_file is not None else processed_rows
out: Dict[str, Any] = {
"status": "waiting_confirmation",
"job_id": job_id,
"total_rows": processed_rows,
"total_rows": total,
"error_count": error_count,
"valid_rows": processed_rows - error_count,
"errors": errors_detail,
}
if message:
out["message"] = message
return out
def commit_result(

70
backend/api/v1/modules/a76/layouts_csv/pedmientos/common/common_validators.py
View File

@@ -1,8 +1,40 @@
"""
Validadores reutilizables para import CSV de pedimentos.
Incluye parseo de fechas (paridad exchange_rate) para FECHA_INICIO, FECHA_FINAL, FECHA_PAGO.
"""
from datetime import datetime, time
from decimal import Decimal, InvalidOperation
from typing import Dict, Any, Optional, Set
from typing import Dict, Any, Optional, Set, List
DATE_FORMATS: List[str] = ["%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y", "%d-%m-%Y", "%Y/%m/%d"]
DATE_FORMAT_PREFERENCE_MAP: Dict[str, str] = {
"dd/mm/yyyy": "%d/%m/%Y",
"mm/dd/yyyy": "%m/%d/%Y",
"yyyy-mm-dd": "%Y-%m-%d",
}
def parse_date(
val: Optional[str], date_format_preference: Optional[str] = None
) -> Optional[datetime]:
"""Parsea fecha; si date_format_preference está definido, solo se acepta ese formato."""
if not val or not str(val).strip():
return None
raw = str(val).strip()
if date_format_preference and date_format_preference in DATE_FORMAT_PREFERENCE_MAP:
fmt = DATE_FORMAT_PREFERENCE_MAP[date_format_preference]
try:
parsed = datetime.strptime(raw, fmt)
return datetime.combine(parsed.date(), time.min)
except ValueError:
return None
for fmt in DATE_FORMATS:
try:
parsed = datetime.strptime(raw, fmt)
return datetime.combine(parsed.date(), time.min)
except ValueError:
continue
return None
def check_required_max(row: Dict[str, Any], col: str, max_len: int, line_num: int) -> Optional[Dict[str, Any]]:
@@ -29,6 +61,42 @@ def check_int_in_set(
return None
def check_optional_int_in_set(
row: Dict[str, Any], col: str, valid_ids: Set[int], line_num: int
) -> Optional[Dict[str, Any]]:
"""Si la columna viene vacía no se valida; si trae valor debe ser entero y estar en valid_ids."""
val = (row.get(col) or "").strip()
if not val:
return None
try:
client_id = int(val)
except ValueError:
return {"line": line_num, "col": col, "msg": "Debe ser número entero"}
if valid_ids and client_id not in valid_ids:
return {"line": line_num, "col": col, "msg": "Cliente no existe en catálogo"}
return None
def check_optional_short_name(
row: Dict[str, Any],
col: str,
short_name_to_id: Dict[str, int],
line_num: int,
) -> Optional[Dict[str, Any]]:
"""Si la columna viene vacía no se valida; si trae valor debe ser short_name existente en catálogo. Solo string, no ID."""
val = (row.get(col) or "").strip()
if not val:
return None
key = val.upper()
if key not in short_name_to_id:
return {
"line": line_num,
"col": col,
"msg": "Cliente no encontrado (short_name no existe en catálogo)",
}
return None
def check_in_set(
row: Dict[str, Any],
col: str,

123
backend/api/v1/modules/a76/layouts_csv/pedmientos/common/fk_loader.py
View File

@@ -1,25 +1,63 @@
"""
Carga de conjuntos FK para validación de import CSV de pedimentos.
Clarion: clave+regimen+tipo (CodePedimentoRegimen), aduana (CustomsSection), patente (CustomsBroker.license), existing_pedimento_keys.
Legacy: valid_client_ids, valid_regimes, valid_pedimento_codes. Cliente por short_name: short_name_to_id.
"""
from typing import Set, Tuple
from typing import Dict, Set, Tuple
from sqlalchemy.orm import Session
def _pedimento_key(year: str, customs_office: str, license_val: str, pedimento_number: str) -> str:
"""Clave única para identificar un pedimento (tenant/company se filtra en query)."""
return f"{year}|{customs_office}|{license_val}|{pedimento_number}"
def load_pedimentos_fk_sets(
session: Session, tenant_id: int, company_id: int
) -> Tuple[Set[int], Set[str], Set[str]]:
) -> Tuple[
Set[int],
Set[str],
Set[str],
Set[Tuple[str, str, str]],
Set[str],
Set[str],
Set[str],
Set[str],
Dict[str, int],
]:
"""
Carga valid_client_ids (ClientProvider.id), valid_regimes (RegimenPedimento.code),
valid_pedimento_codes (PedimentoCode.code).
Carga todos los conjuntos FK para validación Clarion y legacy.
Returns:
valid_client_ids,
valid_regimes,
valid_pedimento_codes,
valid_clave_regimen_tipo, # (pedimento_code, regimen_code, type_code)
valid_aduana_seccion, # customs_code
existing_pedimento_keys, # key strings para actualizar
valid_anexo22_claves, # stub vacío hasta tener catálogo
valid_patentes, # CustomsBroker.license (tenant/company)
short_name_to_id, # short_name normalizado (upper) -> client id (primera aparición gana)
"""
from api.v1.modules.a76.clients_and_providers.models import ClientProvider
from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos
from api.v1.modules.public.reference_data.code_pedimento_regimens.models import (
CodePedimentoRegimen,
)
from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection
from api.v1.modules.public.reference_data.pedimento_codes.models import PedimentoCode
from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento
valid_client_ids: Set[int] = set()
valid_regimes: Set[str] = set()
valid_pedimento_codes: Set[str] = set()
valid_clave_regimen_tipo: Set[Tuple[str, str, str]] = set()
valid_aduana_seccion: Set[str] = set()
existing_pedimento_keys: Set[str] = set()
valid_anexo22_claves: Set[str] = set()
valid_patentes: Set[str] = set()
short_name_to_id: Dict[str, int] = {}
for cp in (
session.query(ClientProvider)
@@ -27,12 +65,85 @@ def load_pedimentos_fk_sets(
ClientProvider.tenant_id == tenant_id,
ClientProvider.company_id == company_id,
)
.order_by(ClientProvider.id)
.all()
):
valid_client_ids.add(cp.id)
sn = (cp.short_name or "").strip()
if sn:
key = sn.upper()
if key not in short_name_to_id:
short_name_to_id[key] = cp.id
for r in session.query(RegimenPedimento).all():
valid_regimes.add(r.code)
for pc in session.query(PedimentoCode).all():
valid_pedimento_codes.add(pc.code)
return valid_client_ids, valid_regimes, valid_pedimento_codes
for cpr in session.query(CodePedimentoRegimen).all():
# type_code puede ser None; Clarion usa I/E. Normalizar a mayúsculas para coincidir con CSV.
t = (cpr.type_code or "").strip().upper()
if t:
valid_clave_regimen_tipo.add(
(
(cpr.pedimento_code or "").strip().upper(),
(cpr.regimen_code or "").strip().upper(),
t,
)
)
for cs in session.query(CustomsSection).all():
valid_aduana_seccion.add(cs.customs_code.strip())
for cb in (
session.query(CustomsBroker)
.filter(
CustomsBroker.tenant_id == tenant_id,
CustomsBroker.company_id == company_id,
)
.all()
):
if (cb.license or "").strip():
valid_patentes.add((cb.license or "").strip())
for p in (
session.query(
Pedimentos.year,
Pedimentos.customs_office,
Pedimentos.license,
Pedimentos.pedimento_number,
)
.filter(
Pedimentos.tenant_id == tenant_id,
Pedimentos.company_id == company_id,
)
.all()
):
existing_pedimento_keys.add(
_pedimento_key(
(p.year or "").strip(),
(p.customs_office or "").strip(),
(p.license or "").strip(),
(p.pedimento_number or "").strip(),
)
)
return (
valid_client_ids,
valid_regimes,
valid_pedimento_codes,
valid_clave_regimen_tipo,
valid_aduana_seccion,
existing_pedimento_keys,
valid_anexo22_claves,
valid_patentes,
short_name_to_id,
)
def pedimento_key_from_parsed(
year: str, customs_office: str, license_val: str, pedimento_number: str
) -> str:
"""Construye la clave para comparar con existing_pedimento_keys."""
return _pedimento_key(year, customs_office, license_val, pedimento_number)

150
backend/api/v1/modules/a76/layouts_csv/pedmientos/common/mappers.py
View File

@@ -1,9 +1,14 @@
"""
Mapeo fila CSV → datos para PedimentosCreate.
Soporta layout Clarion (PEDIMENTO, TIPO_OPERACION, CLAVE_PEDIMENTO, etc.) y legacy (AÑO, ADUANA, PATENTE, NUMERO).
"""
from datetime import datetime
from decimal import Decimal, InvalidOperation
from typing import Dict, Any, Optional
from ..template_config import is_clarion_layout, parse_pedimento_col_a
from ..common.common_validators import parse_date
def parse_decimal(val: Any) -> Optional[Decimal]:
if val is None or (isinstance(val, str) and not val.strip()):
@@ -14,15 +19,102 @@ def parse_decimal(val: Any) -> Optional[Decimal]:
return None
def row_to_pedimento_data(row_norm: Dict[str, Any]) -> Dict[str, Any]:
"""Build dict for PedimentosCreate from normalized CSV row."""
def resolve_client_id_from_short_name(
valor: str, short_name_to_id: Dict[str, int]
) -> Optional[int]:
"""Resuelve short_name (string) a client_id. Vacío → None. Misma normalización que fk_loader (strip + upper)."""
if not valor or not str(valor).strip():
return None
key = str(valor).strip().upper()
return short_name_to_id.get(key)
def _row_to_pedimento_data_clarion(
row_norm: Dict[str, Any],
short_name_to_id: Dict[str, int],
date_format_preference: Optional[str] = None,
) -> Dict[str, Any]:
"""Desde layout Clarion: Cols 1-3 (AÑO, PATENTE, NUMERO) o Col A (PEDIMENTO); Col H → customs_office; B,C,D,J y fechas. CLIENTE_SHORT_NAME opcional."""
año = (row_norm.get("AÑO") or "").strip()
patente = (row_norm.get("PATENTE") or "").strip()
numero = (row_norm.get("NUMERO") or "").strip()
if año and patente and numero:
year = año[:2]
license_val = patente[:4]
pedimento_number = numero[:7]
else:
ped = (row_norm.get("PEDIMENTO") or "").strip()
parsed = parse_pedimento_col_a(ped)
if not parsed:
raise ValueError("PEDIMENTO inválido o vacío")
year, license_val, pedimento_number = parsed
aduana = (row_norm.get("ADUANA_SECCION_CRUCE") or "").strip()[:3]
clave = (row_norm.get("CLAVE_PEDIMENTO") or "").strip()[:2]
regime = (row_norm.get("REGIMEN") or "").strip()[:3]
client_short_name = (row_norm.get("CLIENTE_SHORT_NAME") or "").strip()
client_id = resolve_client_id_from_short_name(client_short_name, short_name_to_id)
data: Dict[str, Any] = {
"year": year,
"customs_office": aduana,
"license": license_val,
"pedimento_number": pedimento_number,
"pedimento_code": clave,
"regime": regime,
}
if client_id is not None:
data["client_id"] = client_id
tipo = (row_norm.get("TIPO_OPERACION") or "").strip().upper()
if tipo == "I":
data["operation_type"] = "imp"
elif tipo == "E":
data["operation_type"] = "exp"
ind_con = (row_norm.get("INDIVIDUAL_CONSOLIDADO") or "").strip().upper() or "CON"
data["pedimento_type"] = "normal" if ind_con == "IND" else "consolidated"
status = (row_norm.get("ESTATUS") or "").strip()
if status:
data["status"] = status[:30]
data["usd_value"] = parse_decimal(row_norm.get("VALOR_USD"))
data["paid_price"] = parse_decimal(row_norm.get("PRECIO_PAGADO"))
data["gross_weight"] = parse_decimal(row_norm.get("PESO_BRUTO"))
data["exchange_rate"] = parse_decimal(row_norm.get("TIPO_CAMBIO"))
obs = (row_norm.get("OBSERVACIONES") or "").strip()
if obs:
data["observations"] = obs
# Fechas E, F, G → pedimento_dates
start_str = (row_norm.get("FECHA_INICIO") or "").strip()
end_str = (row_norm.get("FECHA_FINAL") or "").strip()
payment_str = (row_norm.get("FECHA_PAGO") or "").strip()
base = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
start_dt = parse_date(start_str, date_format_preference) if start_str else base
end_dt = parse_date(end_str, date_format_preference) if end_str else base
payment_dt = parse_date(payment_str, date_format_preference) if payment_str else base
if start_dt and end_dt:
data["pedimento_dates"] = {
"entry_date": start_dt,
"end_date": end_dt,
"start_date": start_dt,
"payment_date": payment_dt,
}
return data
def _row_to_pedimento_data_legacy(
row_norm: Dict[str, Any], short_name_to_id: Dict[str, int]
) -> Dict[str, Any]:
"""Layout legacy: AÑO, ADUANA, PATENTE, NUMERO, CLIENTE_SHORT_NAME, CODIGO_PEDIMENTO/CLAVE_PEDIMENTO, REGIMEN."""
year = (row_norm.get("AÑO") or "").strip()[:2]
customs_office = (row_norm.get("ADUANA") or "").strip()[:3]
customs_office = (row_norm.get("ADUANA_SECCION_CRUCE") or row_norm.get("ADUANA") or "").strip()[:3]
license_val = (row_norm.get("PATENTE") or "").strip()[:4]
pedimento_number = (row_norm.get("NUMERO") or "").strip()[:7]
client_id_str = (row_norm.get("CLIENTE_ID") or "").strip()
client_id = int(client_id_str) if client_id_str else None
pedimento_code = (row_norm.get("CODIGO_PEDIMENTO") or "").strip()[:2]
client_short_name = (row_norm.get("CLIENTE_SHORT_NAME") or "").strip()
client_id = resolve_client_id_from_short_name(client_short_name, short_name_to_id)
pedimento_code = (row_norm.get("CLAVE_PEDIMENTO") or row_norm.get("CODIGO_PEDIMENTO") or "").strip()[:2]
regime = (row_norm.get("REGIMEN") or "").strip()[:3]
data = {
@@ -38,22 +130,58 @@ def row_to_pedimento_data(row_norm: Dict[str, Any]) -> Dict[str, Any]:
op = (row_norm.get("TIPO_OPERACION") or "").strip().lower()
if op in ("imp", "exp"):
data["operation_type"] = op
ptype = (row_norm.get("TIPO_PEDIMENTO") or "").strip().lower()
if ptype in ("normal", "consolidated", "complementary", "automobile"):
data["pedimento_type"] = ptype
status = (row_norm.get("ESTATUS") or "").strip()
if status:
data["status"] = status[:30]
data["usd_value"] = parse_decimal(row_norm.get("VALOR_USD"))
data["paid_price"] = parse_decimal(row_norm.get("PRECIO_PAGADO"))
data["gross_weight"] = parse_decimal(row_norm.get("PESO_BRUTO"))
data["exchange_rate"] = parse_decimal(row_norm.get("TIPO_CAMBIO"))
obs = (row_norm.get("OBSERVACIONES") or "").strip()
if obs:
data["observations"] = obs
return data
def row_to_pedimento_data(
row_norm: Dict[str, Any],
short_name_to_id: Dict[str, int],
date_format_preference: Optional[str] = None,
) -> Dict[str, Any]:
"""
Build dict for PedimentosCreate from normalized CSV row.
Usa layout Clarion si existe PEDIMENTO; si no, layout legacy. Cliente por short_name → client_id.
"""
if is_clarion_layout(row_norm):
return _row_to_pedimento_data_clarion(
row_norm, short_name_to_id, date_format_preference
)
return _row_to_pedimento_data_legacy(row_norm, short_name_to_id)
def row_to_pedimento_data_merge_existing(
row_norm: Dict[str, Any],
existing: Dict[str, Any],
short_name_to_id: Dict[str, int],
date_format_preference: Optional[str] = None,
) -> Dict[str, Any]:
"""
Para modo actualizar: campos vacíos en la fila se rellenan con el pedimento existente.
"""
if is_clarion_layout(row_norm):
new_data = _row_to_pedimento_data_clarion(
row_norm, short_name_to_id, date_format_preference
)
else:
new_data = _row_to_pedimento_data_legacy(row_norm, short_name_to_id)
for key, val in new_data.items():
if val is None or (isinstance(val, str) and not val.strip()):
if key in existing and existing[key] is not None:
new_data[key] = existing[key]
if not new_data.get("pedimento_dates") and existing.get("pedimento_dates"):
new_data["pedimento_dates"] = existing["pedimento_dates"]
return new_data

24
backend/api/v1/modules/a76/layouts_csv/pedmientos/routes.py
View File

@@ -10,7 +10,7 @@ from uuid import uuid4
from fastapi import APIRouter, File, HTTPException, Query, UploadFile, Depends
from sqlalchemy.orm import Session
from typing import Dict, Any
from typing import Dict, Any, Optional
from core.celery_app import celery_app
from core.database import get_core_db
@@ -21,10 +21,10 @@ from .schemas import ImportJobResponse
from .tasks import (
scan_file,
insert_valid_rows,
PED_IMPORT_FILE_PREFIX,
PED_IMPORT_META_PREFIX,
JOB_TYPE as PED_JOB_TYPE,
PED_IMPORT_REDIS_TTL,
)
from ..common import storage as common_storage
router = APIRouter()
logger = logging.getLogger(__name__)
@@ -40,6 +40,8 @@ def _get_redis():
async def upload_import_file(
file: UploadFile = File(...),
company_id: int = Query(..., description="Company ID"),
actualizar: bool = Query(False, description="Modo actualizar: validación parcial si el pedimento existe"),
dateFormat: Optional[str] = Query(None, description="Formato de fecha: dd/mm/yyyy, mm/dd/yyyy, yyyy-mm-dd"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
@@ -58,22 +60,26 @@ async def upload_import_file(
job_id = str(uuid4())
contents = await file.read()
meta_data = {
file_key, meta_key, _ = common_storage.storage_keys(PED_JOB_TYPE, job_id)
meta_data: Dict[str, Any] = {
"tenant_id": tenant_id,
"company_id": company_id,
"user_id": current_user.get("id"),
"template_id": "pedimentos",
"actualizar": actualizar,
}
if dateFormat:
meta_data["dateFormat"] = dateFormat
try:
r = _get_redis()
r.set(
f"{PED_IMPORT_FILE_PREFIX}{job_id}",
file_key,
base64.b64encode(contents),
ex=PED_IMPORT_REDIS_TTL,
)
r.set(
f"{PED_IMPORT_META_PREFIX}{job_id}",
meta_key,
json.dumps(meta_data).encode("utf-8"),
ex=PED_IMPORT_REDIS_TTL,
)
@@ -84,9 +90,11 @@ async def upload_import_file(
try:
upload_dir = layout_path("imports", "temp")
os.makedirs(upload_dir, exist_ok=True)
with open(os.path.join(upload_dir, f"ped_{job_id}.csv"), "wb") as f:
csv_path = common_storage.file_path_for_job(PED_JOB_TYPE, job_id)
with open(csv_path, "wb") as f:
f.write(contents)
with open(os.path.join(upload_dir, f"ped_{job_id}.meta.json"), "w") as f:
meta_path = csv_path.replace(".csv", ".meta.json")
with open(meta_path, "w") as f:
json.dump(meta_data, f)
except Exception as e:
logger.warning(f"Pedimentos import: local file save failed: {e}")

232
backend/api/v1/modules/a76/layouts_csv/pedmientos/tasks.py
View File

@@ -17,10 +17,15 @@ from ..common import normalize as common_normalize
from ..common import meta as common_meta
from ..common import responses as common_responses
from ..common import csv_reader as common_csv_reader
from .template_config import row_from_template
from .template_config import (
row_from_template,
is_clarion_layout,
parse_pedimento_col_a,
detect_headers_or_data,
)
from .validators import validate_row_pedimento
from .common.fk_loader import load_pedimentos_fk_sets
from .common.mappers import row_to_pedimento_data
from .common.mappers import row_to_pedimento_data, row_to_pedimento_data_merge_existing
from .common.fk_loader import load_pedimentos_fk_sets, pedimento_key_from_parsed
logger = logging.getLogger(__name__)
@@ -42,12 +47,19 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str,
file_path = common_storage.ensure_file_from_redis(JOB_TYPE, job_id, "Pedimentos import")
if not file_path:
return {"status": "failed", "error": "Archivo no encontrado (expirado o no subido). Sube de nuevo."}
if os.path.getsize(file_path) == 0:
return {"status": "failed", "error": "El archivo está vacío. Verifica que el CSV tenga contenido."}
common_storage.ensure_meta_from_redis(JOB_TYPE, job_id, file_path, "Pedimentos import")
error_path = common_storage.error_path_for_job(JOB_TYPE, job_id)
fieldnames, has_header = detect_headers_or_data(
file_path,
common_normalize.normalize_header,
parse_pedimento_col_a,
)
try:
total_rows = common_csv_reader.count_csv_rows(file_path)
total_rows = common_csv_reader.count_csv_rows(file_path, has_header=has_header)
except Exception as e:
return {"status": "failed", "error": str(e)}
@@ -56,11 +68,23 @@ 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)
date_format_preference = meta.get("dateFormat") or meta.get("date_format")
try:
with CoreSessionLocal() as session:
valid_client_ids, valid_regimes, valid_pedimento_codes = load_pedimentos_fk_sets(
session, tenant_id, company_id
)
(
valid_client_ids,
valid_regimes,
valid_pedimento_codes,
valid_clave_regimen_tipo,
valid_aduana_seccion,
existing_pedimento_keys,
valid_anexo22_claves,
valid_patentes,
short_name_to_id,
) = load_pedimentos_fk_sets(session, tenant_id, company_id)
except Exception as e:
logger.error("Pedimentos import: failed to load FK sets: %s", e)
return {"status": "failed", "error": "No se pudo cargar catálogos"}
@@ -72,13 +96,25 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str,
try:
with open(error_path, "w", encoding="utf-8") as f_err:
for i, row in common_csv_reader.iter_csv_rows(file_path):
for i, row in common_csv_reader.iter_csv_rows(file_path, fieldnames=fieldnames):
if progress_callback and i % 500 == 0:
progress_callback(i, total_rows, error_count)
row_norm = _norm_row(row)
err = validate_row_pedimento(
row_norm, i, valid_client_ids, valid_regimes, valid_pedimento_codes
row_norm,
i,
short_name_to_id,
valid_regimes,
valid_pedimento_codes,
valid_clave_regimen_tipo,
valid_aduana_seccion,
existing_pedimento_keys,
valid_anexo22_claves,
valid_patentes,
actualizar=actualizar,
raw_row=row,
date_format_preference=date_format_preference,
)
if err:
error_count += 1
@@ -118,8 +154,7 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
if not os.path.exists(alt_path):
return {"status": "failed", "error": "Archivo no encontrado (expirado). Sube y confirma de nuevo."}
file_path = alt_path
else:
common_storage.ensure_meta_from_redis(JOB_TYPE, job_id, file_path, "Pedimentos import")
common_storage.ensure_meta_from_redis(JOB_TYPE, job_id, file_path, "Pedimentos import")
error_path = common_storage.error_path_for_job(JOB_TYPE, job_id)
error_lines = common_storage.get_error_lines(JOB_TYPE, job_id, error_path)
@@ -129,35 +164,90 @@ 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)
date_format_preference = meta.get("dateFormat") or meta.get("date_format")
try:
with CoreSessionLocal() as session:
valid_client_ids, valid_regimes, valid_pedimento_codes = load_pedimentos_fk_sets(
session, tenant_id, company_id
)
(
valid_client_ids,
valid_regimes,
valid_pedimento_codes,
valid_clave_regimen_tipo,
valid_aduana_seccion,
existing_pedimento_keys,
valid_anexo22_claves,
valid_patentes,
short_name_to_id,
) = load_pedimentos_fk_sets(session, tenant_id, company_id)
except Exception as e:
logger.error("Pedimentos import: failed to load FK sets: %s", e)
return {"status": "failed", "error": "No se pudo cargar catálogos"}
from api.v1.modules.a76.pedmientos.dtos.pedimentos import PedimentosCreate
from api.v1.modules.a76.pedmientos.dtos.pedimentos import (
PedimentosCreate,
PedimentosUpdate,
)
from api.v1.modules.a76.pedmientos.dtos.pedimento_dates import PedimentoDatesCreate
from api.v1.modules.a76.pedmientos.services.pedimentos import PedimentosService
from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos
inserted_count = 0
updated_count = 0
skipped_invalid = 0
skipped_missing_fk = 0
skipped_duplicate = 0
skipped_details: List[Dict[str, Any]] = []
meta_path = common_meta.get_meta_path(file_path)
fieldnames_commit, _ = detect_headers_or_data(
file_path,
common_normalize.normalize_header,
parse_pedimento_col_a,
)
def _key_from_row(r: Dict[str, Any]) -> Optional[str]:
if is_clarion_layout(r):
año = (r.get("AÑO") or "").strip()
patente = (r.get("PATENTE") or "").strip()
numero = (r.get("NUMERO") or "").strip()
if año and patente and numero:
adu = (r.get("ADUANA_SECCION_CRUCE") or "").strip()[:3]
return pedimento_key_from_parsed(año[:2], adu, patente[:4], numero[:7])
parsed = parse_pedimento_col_a((r.get("PEDIMENTO") or "").strip())
if not parsed:
return None
y, lic, num = parsed
adu = (r.get("ADUANA_SECCION_CRUCE") or "").strip()[:3]
return pedimento_key_from_parsed(y, adu, lic, num)
y = (r.get("AÑO") or "").strip()[:2]
adu = (r.get("ADUANA_SECCION_CRUCE") or r.get("ADUANA") or "").strip()[:3]
lic = (r.get("PATENTE") or "").strip()[:4]
num = (r.get("NUMERO") or "").strip()[:7]
return pedimento_key_from_parsed(y, adu, lic, num)
try:
with CoreSessionLocal() as session:
for i, row in common_csv_reader.iter_csv_rows(file_path):
for i, row in common_csv_reader.iter_csv_rows(file_path, fieldnames=fieldnames_commit):
if i in error_lines:
continue
row_norm = _norm_row(row)
err = validate_row_pedimento(
row_norm, i, valid_client_ids, valid_regimes, valid_pedimento_codes
row_norm,
i,
short_name_to_id,
valid_regimes,
valid_pedimento_codes,
valid_clave_regimen_tipo,
valid_aduana_seccion,
existing_pedimento_keys,
valid_anexo22_claves,
valid_patentes,
actualizar=actualizar,
raw_row=row,
date_format_preference=date_format_preference,
)
if err:
skipped_invalid += 1
@@ -167,16 +257,95 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
})
continue
key = _key_from_row(row_norm)
try:
data = row_to_pedimento_data(row_norm)
if "pedimento_dates" not in data or data.get("pedimento_dates") is None:
data["pedimento_dates"] = PedimentoDatesCreate(
entry_date=datetime.now(),
end_date=datetime.now(),
# Si el pedimento ya existe: actualizar (merge) o reemplazar (crear). Si no existe: crear.
if key and key in existing_pedimento_keys:
existing = (
session.query(Pedimentos)
.filter(
Pedimentos.tenant_id == tenant_id,
Pedimentos.company_id == company_id,
Pedimentos.year == key.split("|")[0],
Pedimentos.customs_office == key.split("|")[1],
Pedimentos.license == key.split("|")[2],
Pedimentos.pedimento_number == key.split("|")[3],
)
.first()
)
create_data = PedimentosCreate(**data)
PedimentosService.create(session, create_data, tenant_id, company_id)
inserted_count += 1
if existing:
if actualizar:
# Modo actualizar: merge con existente
existing_dict = {
"year": existing.year,
"customs_office": existing.customs_office,
"license": existing.license,
"pedimento_number": existing.pedimento_number,
"client_id": existing.client_id,
"pedimento_code": existing.pedimento_code,
"regime": existing.regime,
"operation_type": existing.operation_type,
"pedimento_type": existing.pedimento_type,
"status": existing.status,
"usd_value": existing.usd_value,
"paid_price": existing.paid_price,
"gross_weight": existing.gross_weight,
"exchange_rate": existing.exchange_rate,
"observations": existing.observations,
}
if existing.pedimento_dates:
existing_dict["pedimento_dates"] = {
"entry_date": existing.pedimento_dates.entry_date,
"end_date": existing.pedimento_dates.end_date,
"start_date": getattr(
existing.pedimento_dates, "start_date", existing.pedimento_dates.entry_date
),
"payment_date": getattr(
existing.pedimento_dates, "payment_date", existing.pedimento_dates.entry_date
),
}
data = row_to_pedimento_data_merge_existing(
row_norm,
existing_dict,
short_name_to_id,
date_format_preference,
)
else:
# Modo crear: reemplazar con datos del CSV
data = row_to_pedimento_data(
row_norm, short_name_to_id, date_format_preference
)
if "pedimento_dates" not in data or data.get("pedimento_dates") is None:
data["pedimento_dates"] = PedimentoDatesCreate(
entry_date=datetime.now(),
end_date=datetime.now(),
)
update_data = {k: v for k, v in data.items() if k != "pedimento_dates"}
if data.get("pedimento_dates"):
update_data["pedimento_dates"] = PedimentoDatesCreate(
**data["pedimento_dates"]
)
update_schema = PedimentosUpdate(**update_data)
PedimentosService.update(
session, existing.id, tenant_id, update_schema, company_id
)
updated_count += 1
else:
skipped_invalid += 1
skipped_details.append({"line": i, "reason": "Pedimento no encontrado"})
else:
# No existe: crear (ambos modos)
data = row_to_pedimento_data(
row_norm, short_name_to_id, date_format_preference
)
if "pedimento_dates" not in data or data.get("pedimento_dates") is None:
data["pedimento_dates"] = PedimentoDatesCreate(
entry_date=datetime.now(),
end_date=datetime.now(),
)
create_data = PedimentosCreate(**data)
PedimentosService.create(session, create_data, tenant_id, company_id)
inserted_count += 1
except ValueError as ve:
if "Ya existe" in str(ve) or "duplicate" in str(ve).lower():
skipped_duplicate += 1
@@ -201,7 +370,12 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
)
total_skipped = skipped_invalid + skipped_missing_fk + skipped_duplicate
if inserted_count == 0 and total_skipped > 0:
if inserted_count == 0 and updated_count == 0 and total_skipped > 0:
reasons = "; ".join(
f"Línea {d.get('line', '?')}: {d.get('reason', '')}" for d in skipped_details[:5]
)
if len(skipped_details) > 5:
reasons += f" (+{len(skipped_details) - 5} más)"
return {
"status": "warning",
"inserted": 0,
@@ -210,9 +384,9 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
"skipped_missing_fk": skipped_missing_fk,
"skipped_duplicate": skipped_duplicate,
"skipped_details": skipped_details,
"message": f"No se insertaron registros. {total_skipped} rechazados.",
"message": f"No se insertaron registros. {total_skipped} rechazados. Motivos: {reasons}",
}
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",
@@ -226,7 +400,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_missing_fk": skipped_missing_fk,
"skipped_duplicate": skipped_duplicate,

186
backend/api/v1/modules/a76/layouts_csv/pedmientos/template_config.py
View File

@@ -1,22 +1,97 @@
"""
Configuración de plantilla CSV para Pedimentos (EstructuraCatPedimentos.xls).
Solo se leen columnas definidas aquí; el resto se ignora.
Layout Clarion: Cols 1-3 = AÑO (2), PATENTE (4), NUMERO (7); luego TIPO, CLAVE PEDIMENTO, REGIMEN,
fechas, ADUANA Y SECCION CRUCE, etc. Compatibilidad: PEDIMENTO (##-####-#######) y layout legacy.
"""
import csv
import io
from typing import Dict, List, Any, Optional, Tuple
from typing import Dict, List, Any, Optional
# Longitudes para validación (sin afectar modelos)
AÑO_LEN = 2
PATENTE_LEN = 4
NUMERO_LEN = 7
# Columnas Clarion (canónicos) + aliases exactos de la plantilla y legacy
TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
"pedimentos": [
# Cols 1-3: AÑO, PATENTE, NUMERO (reemplazan Col A única PEDIMENTO)
{"canonical": "AÑO", "aliases": ["YEAR", "ANIO"]},
{"canonical": "ADUANA", "aliases": ["CUSTOMS_OFFICE", "CUSTOMS OFFICE"]},
{"canonical": "PATENTE", "aliases": ["LICENCIA", "LICENSE", "LIC"]},
{"canonical": "NUMERO", "aliases": ["PEDIMENTO_NUMBER", "PEDIMENTO NUMBER", "NUMERO PEDIMENTO"]},
{"canonical": "CLIENTE_ID", "aliases": ["CLIENT_ID", "CLIENTE", "ID CLIENTE"]},
{"canonical": "TIPO_OPERACION", "aliases": ["OPERATION_TYPE", "OPERACION"]},
{"canonical": "TIPO_PEDIMENTO", "aliases": ["PEDIMENTO_TYPE", "TIPO"]},
{"canonical": "CODIGO_PEDIMENTO", "aliases": ["PEDIMENTO_CODE", "CODIGO", "CLAVE PEDIMENTO"]},
{"canonical": "REGIMEN", "aliases": ["REGIME"]},
{"canonical": "ESTATUS", "aliases": ["STATUS", "ESTADO"]},
{"canonical": "NUMERO", "aliases": ["PEDIMENTO_NUMBER", "PEDIMENTO NUMBER"]},
# Col B
{"canonical": "TIPO_OPERACION", "aliases": ["TIPO MOV(I=Importación,E=Exportación)", "TIPO MOV(I=Impotación,E=Expotación)", "TIPO MOV", "TIPO", "OPERATION_TYPE", "OPERACION"]},
# Col C
{"canonical": "CLAVE_PEDIMENTO", "aliases": ["CLAVE PEDIMENTO", "CODIGO_PEDIMENTO", "PEDIMENTO_CODE", "CODIGO"]},
# Col D
{"canonical": "REGIMEN", "aliases": ["REGIME", "CLAVE RÉGIMEN"]},
# Col E, F, G
{"canonical": "FECHA_INICIO", "aliases": ["FECHA INICIO", "FECHA INICIAL"]},
{"canonical": "FECHA_FINAL", "aliases": ["FECHA FINAL", "FECHA FIN"]},
{"canonical": "FECHA_PAGO", "aliases": ["FECHA DE PAGO", "FECHA PAGO"]},
# Col H
{"canonical": "ADUANA_SECCION_CRUCE", "aliases": ["ADUANA Y SECCION DE CRUCE", "ADUANA Y SECCION CRUCE", "ADUANA", "CUSTOMS_OFFICE", "CUSTOMS OFFICE"]},
# Col I
{"canonical": "ACUSE_ELECTRONICO", "aliases": ["ACUSE ELECTRONICO", "ACUSE ELECTRÓNICO"]},
# Col J
{"canonical": "INDIVIDUAL_CONSOLIDADO", "aliases": ["INDIVIDUAL o CONSOLIDADO (IND,CON)", "INDIVIDUAL o CONSOLIDADO", "IND/CON", "INDIVIDUAL CONSOLIDADO"]},
# Col K, L, M
{"canonical": "MET_TRANSP_ENTRADA", "aliases": ["MET TRANS ENTRADA", "METODO TRANSP ENTRADA"]},
{"canonical": "MET_TRANSP_ARRIVO", "aliases": ["MET TRANS ARRIVO", "METODO TRANSP ARRIVO"]},
{"canonical": "MET_TRANSP_SALIDA", "aliases": ["MET TRANS SALIDA", "METODO TRANSP SALIDA"]},
# Col N (desfase) y resto de columnas de la plantilla
{"canonical": "IEPS", "aliases": []},
{"canonical": "DTA", "aliases": []},
{"canonical": "CNT", "aliases": []},
{"canonical": "PREVALIDACION", "aliases": []},
{"canonical": "MONTO_TIGIE", "aliases": ["MONTO TIGIE"]},
{"canonical": "PAGO_IMPUESTO", "aliases": ["PAGO IMPUESTO? (S/N)"]},
{"canonical": "ES_MIXTO", "aliases": ["ES MIXTO (SI/NO)"]},
{"canonical": "OBS_RECTIFICA", "aliases": ["OBS RECTIFICA"]},
{"canonical": "OPCION_DESTINO", "aliases": ["OPCION DESTINO(Interior del Pais/Región Fronteriza/Franja Fronteriza)", "OPCION DESTINO"]},
{"canonical": "VALOR_IVA", "aliases": ["VALOR IVA"]},
{"canonical": "VALOR_ME", "aliases": ["VALOR ME"]},
{"canonical": "VALOR_ADUANAS", "aliases": ["VALOR ADUANAS"]},
{"canonical": "FLETE", "aliases": []},
{"canonical": "VALOR_SEGUROS", "aliases": ["VALOR SEGUROS"]},
{"canonical": "SEGUROS", "aliases": []},
{"canonical": "EMBALAJES", "aliases": []},
{"canonical": "OTROS_INCREMENTABLES", "aliases": ["OTROS INCREMENTABLES"]},
{"canonical": "ESTATUS", "aliases": ["ESTATUS (ABIERTO/CERRADO)", "STATUS", "ESTADO"]},
{"canonical": "PERSONA_REV", "aliases": ["PERSONA REV"]},
{"canonical": "FECHA_CIERRE", "aliases": ["FECHA CIERRE"]},
{"canonical": "FECHA_REVISION", "aliases": ["FECHA REVISION"]},
{"canonical": "FECHA_AUTORIZACION", "aliases": ["FECHA AUTORIZACION"]},
{"canonical": "FECHA_RECIBIDO", "aliases": ["FECHA RECIBIDO"]},
{"canonical": "REPRESENTANTE_AA", "aliases": ["REPRESENTANTE AA"]},
{"canonical": "CLAVE_DEST_ORIGEN", "aliases": ["CLAVE DEST ORIGEN"]},
{"canonical": "FECHA_ENTRADA_RECINTO", "aliases": ["FECHA ENTRADA RECINTO"]},
{"canonical": "FECHA_EXTRACCION_RECINTO", "aliases": ["FECHA EXTRACCION RECINTO"]},
{"canonical": "ERRORES", "aliases": []},
{"canonical": "FORMA_PAGO_DTA", "aliases": ["FORMA PAGO DTA"]},
{"canonical": "FORMA_PAGO_IGI", "aliases": ["FORMA PAGO IGI"]},
{"canonical": "FORMA_PAGO_PREVAL", "aliases": ["FORMA PAGO PREVAL"]},
{"canonical": "FORMA_PAGO_IVA", "aliases": ["FORMA PAGO IVA"]},
{"canonical": "RECARGOS", "aliases": []},
{"canonical": "MULTAS", "aliases": []},
{"canonical": "IVA_DE_PREV", "aliases": ["IVA DE PREV"]},
{"canonical": "CUOTAS_COMPENSATORIAS", "aliases": ["CUOTAS CONPENSATORIAS"]},
{"canonical": "IDENTIFICADORES", "aliases": []},
{"canonical": "IEPS_2", "aliases": ["IEPS 2"]},
{"canonical": "FORMA_PAGO_IEPS_2", "aliases": ["FORMA DE PAGO IEPS 2"]},
{"canonical": "DTA_2", "aliases": ["DTA 2"]},
{"canonical": "FORMA_PAGO_DTA_2", "aliases": ["FORMA DE PAGO DTA 2"]},
{"canonical": "IVA_2", "aliases": ["IVA 2"]},
{"canonical": "FORMA_PAGO_IVA_2", "aliases": ["FORMA DE PAGO IVA 2"]},
{"canonical": "IGI_2", "aliases": ["IGI 2"]},
{"canonical": "FORMA_PAGO_IGI_2", "aliases": ["FORMA DE PAGO IGI 2"]},
{"canonical": "FORMA_PAGO_PREVALIDACION_2", "aliases": ["FORMA DE PAGO PREVALIDACION 2"]},
{"canonical": "CNT_2", "aliases": ["CNT 2"]},
{"canonical": "FORMA_PAGO_CNT_2", "aliases": ["FORMA DE PAGO CNT 2"]},
# Legacy / compatibilidad (PEDIMENTO una columna ##-####-#######, y otros)
{"canonical": "PEDIMENTO", "aliases": ["NUMERO DE PEDIMENTO (##-####-######)", "NUMERO DE PEDIMENTO", "NUMERO PEDIMENTO", "PED"]},
{"canonical": "CLIENTE_SHORT_NAME", "aliases": ["CLIENTE", "SHORT_NAME", "CLIENTE_ID", "CLIENT_ID", "ID CLIENTE"]},
{"canonical": "TIPO_PEDIMENTO", "aliases": ["PEDIMENTO_TYPE", "TIPO PED"]},
{"canonical": "VALOR_USD", "aliases": ["USD_VALUE", "VALOR USD", "USD"]},
{"canonical": "PRECIO_PAGADO", "aliases": ["PAID_PRICE", "PRECIO PAGADO"]},
{"canonical": "PESO_BRUTO", "aliases": ["GROSS_WEIGHT", "PESO BRUTO", "PESO"]},
@@ -25,6 +100,61 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
],
}
# Orden de columnas para CSV sin cabecera (primera fila = datos). Usado por detect_headers_or_data.
PEDIMENTOS_TEMPLATE_ORDER: List[str] = [
item["canonical"] for item in TEMPLATE_COLUMNS["pedimentos"]
]
def _first_row_looks_like_three_cols_data(first_row: List[str]) -> bool:
"""True si las primeras 3 celdas son AÑO (2 dígitos), PATENTE (4), NUMERO (7)."""
if not first_row or len(first_row) < 3:
return False
c0 = (first_row[0] or "").strip()
c1 = (first_row[1] or "").strip()
c2 = (first_row[2] or "").strip()
return (
len(c0) <= AÑO_LEN and c0.isdigit()
and len(c1) <= PATENTE_LEN and c1.isdigit()
and len(c2) <= NUMERO_LEN and c2.isdigit()
)
def detect_headers_or_data(
file_path: str,
normalize_header_fn,
parse_pedimento_fn,
encoding: str = "utf-8-sig",
) -> Tuple[Optional[List[str]], bool]:
"""
Lee la primera línea del CSV y decide si es cabecera o dato.
- Si las primeras 3 celdas son 2, 4 y 7 dígitos (AÑO, PATENTE, NUMERO) -> has_header=False.
- Si la primera celda tiene formato ##-####-####### (compatibilidad) -> has_header=False.
- Si no -> has_header=True. Devuelve (fieldnames, has_header).
"""
try:
with open(file_path, "r", encoding=encoding) as f:
sample = f.read(2048)
except Exception:
return None, True
lines = sample.splitlines()
if not lines:
return None, True
try:
dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
except Exception:
dialect = csv.excel
reader = csv.reader(io.StringIO(lines[0]), dialect=dialect)
first_row = next(reader, None)
if not first_row:
return None, True
if _first_row_looks_like_three_cols_data(first_row):
return list(PEDIMENTOS_TEMPLATE_ORDER), False
first_cell = (first_row[0] or "").strip()
if parse_pedimento_fn(first_cell) is not None:
return list(PEDIMENTOS_TEMPLATE_ORDER), False
return None, True
def build_normalized_lookup(normalize_header_fn, template_id: str = "pedimentos") -> Dict[str, str]:
"""normalized_header -> canonical_name para plantilla pedimentos."""
@@ -51,3 +181,39 @@ def row_from_template(row: Dict[str, Any], normalize_header_fn, template_id: str
if key_norm in lookup:
out[lookup[key_norm]] = value
return out
# Formato Clarion Col A: ##-####-####### (15 chars, guiones en posiciones 3 y 8, 1-based)
PEDIMENTO_LEN = 15
PEDIMENTO_DASH_POSITIONS = (2, 7) # 0-based: pos 2 y 7 deben ser '-'
def parse_pedimento_col_a(pedimento_str: str) -> Optional[Tuple[str, str, str]]:
"""
Parsea Col A (PEDIMENTO) formato ##-####-#######.
Retorna (year_2, license_4, pedimento_number_7) o None si formato inválido.
"""
if not pedimento_str or not isinstance(pedimento_str, str):
return None
s = (pedimento_str or "").strip()
if len(s) != PEDIMENTO_LEN:
return None
if s[PEDIMENTO_DASH_POSITIONS[0]] != "-" or s[PEDIMENTO_DASH_POSITIONS[1]] != "-":
return None
year = s[0:2]
license_val = s[3:7]
pedimento_number = s[8:15]
if not year.isdigit() or not license_val.isdigit() or not pedimento_number.isdigit():
return None
return (year, license_val, pedimento_number)
def is_clarion_layout(row_norm: Dict[str, Any]) -> bool:
"""True si la fila tiene las 3 columnas AÑO, PATENTE, NUMERO con valor, o PEDIMENTO con valor (compatibilidad)."""
año = (row_norm.get("AÑO") or "").strip()
patente = (row_norm.get("PATENTE") or "").strip()
numero = (row_norm.get("NUMERO") or "").strip()
if año and patente and numero:
return True
ped = (row_norm.get("PEDIMENTO") or "").strip()
return bool(ped)

438
backend/api/v1/modules/a76/layouts_csv/pedmientos/validators/common.py
View File

@@ -1,59 +1,439 @@
"""
Validaciones comunes de fila para import CSV de pedimentos (por tipo).
Validaciones comunes de fila para import CSV de pedimentos.
Paridad Clarion: desfase (Col N), Cols 1-3 (AÑO, PATENTE, NUMERO) u Col A PEDIMENTO, obligatorios B–H (VALIDA_TODA),
VALIDACIONES_PEDIMENTO (Tipo I/E, Clave+Régimen+Tipo, Aduana, Patente, IND/CON, Método transporte).
"""
from typing import Dict, Any, Optional, Set
from typing import Dict, Any, Optional, Set, Tuple
from ..template_config import (
AÑO_LEN,
PATENTE_LEN,
NUMERO_LEN,
PEDIMENTO_LEN,
PEDIMENTO_DASH_POSITIONS,
parse_pedimento_col_a,
)
from ..common.common_validators import (
check_required_max,
check_int_in_set,
check_in_set,
check_int_in_set,
check_optional_int_in_set,
check_optional_short_name,
check_optional_max_length,
check_optional_decimal,
check_optional_in_set,
PEDIMENTO_CODE_MAX,
REGIMEN_MAX,
parse_date,
DATE_FORMAT_PREFERENCE_MAP,
)
# --- Desfase
MSG_DESFASE = "Advertencia: Podría existir un desfase en esta línea."
MSG_DESFASE_SOLUCION = "Revisar esta línea del archivo CSV y verificar cada campo esté en la posición correcta."
def validate_row_pedimento_required(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
for col, max_len in [
("AÑO", 2),
("ADUANA", 3),
("PATENTE", 4),
("NUMERO", 7),
]:
err = check_required_max(row, col, max_len, line_num)
# --- Col A ---
MSG_PEDIMENTO_VACIO = (
"Error: (Col. A) La columna de Pedimento está vacío y no se pueden hacer las validaciones. "
"Capturar en la Columna A un Pedimento nuevo o uno ya existente al cual desee actualizar campos"
)
MSG_PEDIMENTO_LONGITUD = (
"Error: (Col. A) La columna de Pedimento no cumple con la longitud o sintaxis correcta Ej. (XX-XXXX-XXXXXXX)"
)
MSG_PEDIMENTO_FORMATO = "Error: (Col. A) El Formato del Pedimento: {val} es incorrecto."
MSG_PEDIMENTO_FORMATO_SOLUCION = "Capturar en la columna A el campo Pedimento con este formato ##-####-#######."
# --- Cols 1-3 (AÑO, PATENTE, NUMERO) ---
MSG_ANIO_PATENTE_NUMERO = "Error: (Cols 1-3) AÑO, PATENTE y NUMERO son obligatorios (AÑO 2 dígitos, PATENTE 4, NUMERO 7)."
MSG_PATENTE_NO_EXISTE = "Error: (Patente) La patente {val} no está dada de alta en el catálogo de agentes aduanales."
# --- Obligatorios B–H ---
MSG_OBLIGATORIOS = "Existen campos vacios que son obligatorios, es la {campos}."
MSG_OBLIGATORIOS_SOLUCION = "Revisar la línea del archivo y capturar los campos con la información correcta."
# --- Tipo B ---
MSG_TIPO_INVALIDO = "Error: (Col. B) El Tipo de Operación: {val} no es valido."
MSG_TIPO_SOLUCION = "Capturar una opción valida, que seria I para Importación, E para Exportación."
# --- Combinación B+C+D ---
MSG_COMBINACION = "Error, (Col. B, C y D) La combinación de una operación de {b} con clave de pedimento {c} y régimen {d} no es posible."
MSG_COMBINACION_SOLUCION = "Las posibles combinaciones son {comb}."
# --- Aduana H ---
MSG_ADUANA_NO_EXISTE = "Error: (Col. H) La aduana y sección: {val} no existe."
MSG_ADUANA_SOLUCION = "Revisar que sea correcta esta aduana y seccion de cruce, de ser así actualice los catálogos fijos."
# --- Col J IND/CON ---
MSG_IND_CON_INVALIDO = "Error: (Col. J) La opción de Individual/Consolidado: {val} no es valido."
MSG_IND_CON_SOLUCION = "Capturar una opción valida: IND para Individual, CON para Consolidado: o dejar el campo vacio y automaticamente se asigna Consolidado."
# --- Transporte K/L/M ---
MSG_TRANSP_NO_EXISTE = "Error: (Col. {col}) El Metodo de Transporte: {val} no existe en el Catálogo del Anexo 22 Apendice 3."
MSG_TRANSP_SOLUCION = "Revisar si esta asignado Correctamente."
# --- Fechas ---
FECHA_MAX_LEN = 10
def validate_row_desfase_pedimento(
raw_row: Dict[str, Any], line_num: int
) -> Optional[Dict[str, Any]]:
"""Si la fila tiene al menos 16 columnas y la 16ª (Col N, IEPS/desfase) tiene valor, error de desfase. Orden: AÑO,PATENTE,NUMERO,TIPO,...,IEPS en índice 15."""
values_ordered = list(raw_row.values()) if raw_row else []
desfase_idx = 15 # IEPS en PEDIMENTOS_TEMPLATE_ORDER (tras AÑO,PATENTE,NUMERO + 12 columnas más)
if len(values_ordered) >= (desfase_idx + 1) and (values_ordered[desfase_idx] or "").strip():
return {
"line": line_num,
"col": "",
"msg": f"{MSG_DESFASE} {MSG_DESFASE_SOLUCION}",
}
return None
def validate_row_pedimento_required_col_a(
row: Dict[str, Any], line_num: int
) -> Optional[Dict[str, Any]]:
"""Col A (PEDIMENTO) obligatorio y longitud 15."""
val = (row.get("PEDIMENTO") or "").strip()
if not val:
return {"line": line_num, "col": "PEDIMENTO", "msg": MSG_PEDIMENTO_VACIO}
if len(val) != PEDIMENTO_LEN:
return {
"line": line_num,
"col": "PEDIMENTO",
"msg": f"{MSG_PEDIMENTO_LONGITUD} {MSG_PEDIMENTO_FORMATO_SOLUCION}",
}
return None
def validate_row_pedimento_format(
row: Dict[str, Any], line_num: int
) -> Optional[Dict[str, Any]]:
"""Formato ##-####-#######: guiones en posiciones 3 y 8 (1-based). Solo aplica cuando la fila trae PEDIMENTO."""
val = (row.get("PEDIMENTO") or "").strip()
if not val or len(val) != PEDIMENTO_LEN:
return None
if val[PEDIMENTO_DASH_POSITIONS[0]] != "-" or val[PEDIMENTO_DASH_POSITIONS[1]] != "-":
return {
"line": line_num,
"col": "PEDIMENTO",
"msg": f"{MSG_PEDIMENTO_FORMATO.format(val=val)} {MSG_PEDIMENTO_FORMATO_SOLUCION}",
}
return None
def validate_row_pedimento_required_three_cols(
row: Dict[str, Any], line_num: int
) -> Optional[Dict[str, Any]]:
"""Cols 1-3: AÑO (2 dígitos), PATENTE (4), NUMERO (7) obligatorios y numéricos."""
cols_missing = []
año = (row.get("AÑO") or "").strip()
patente = (row.get("PATENTE") or "").strip()
numero = (row.get("NUMERO") or "").strip()
if not año:
cols_missing.append("Col.1) AÑO")
elif len(año) > AÑO_LEN or not año.isdigit():
return {"line": line_num, "col": "AÑO", "msg": "AÑO debe ser 2 dígitos numéricos."}
if not patente:
cols_missing.append("Col.2) PATENTE")
elif len(patente) > PATENTE_LEN or not patente.isdigit():
return {"line": line_num, "col": "PATENTE", "msg": "PATENTE debe ser 4 dígitos numéricos."}
if not numero:
cols_missing.append("Col.3) NUMERO")
elif len(numero) > NUMERO_LEN or not numero.isdigit():
return {"line": line_num, "col": "NUMERO", "msg": "NUMERO debe ser 7 dígitos numéricos."}
if cols_missing:
campos = ", ".join(cols_missing)
return {
"line": line_num,
"col": "",
"msg": f"{MSG_ANIO_PATENTE_NUMERO} Faltan: {campos}.",
}
return None
def validate_row_patente(
row: Dict[str, Any],
line_num: int,
valid_patentes: Set[str],
) -> Optional[Dict[str, Any]]:
"""Si la fila trae PATENTE o PEDIMENTO (parseado), valida que la patente esté en el catálogo de agentes aduanales."""
patente_val = (row.get("PATENTE") or "").strip()
if not patente_val:
ped = (row.get("PEDIMENTO") or "").strip()
parsed = parse_pedimento_col_a(ped) if ped else None
if parsed:
_, patente_val, _ = parsed
if not patente_val:
return None
if valid_patentes and patente_val not in valid_patentes:
return {
"line": line_num,
"col": "PATENTE",
"msg": MSG_PATENTE_NO_EXISTE.format(val=patente_val),
}
return None
def validate_row_pedimento_required_full(
row: Dict[str, Any], line_num: int
) -> Optional[Dict[str, Any]]:
"""Obligatorios cuando no es actualizar: TIPO (B), CLAVE_PEDIMENTO (C), REGIMEN (D), FECHA_INICIO (E), FECHA_FINAL (F), FECHA_PAGO (G), ADUANA_SECCION_CRUCE (H)."""
cols_missing = []
col_labels = [
("TIPO_OPERACION", "Col.B) Tipo Operación"),
("CLAVE_PEDIMENTO", "Col.C) Clave Pedimento"),
("REGIMEN", "Col.D) Clave Régimen"),
("FECHA_INICIO", "Col.E) Fecha Inicio"),
("FECHA_FINAL", "Col.F) Fecha Final"),
("FECHA_PAGO", "Col.G) Fecha de Pago"),
("ADUANA_SECCION_CRUCE", "Col.H) Aduana y Sección de Cruce"),
]
for key, label in col_labels:
if not (row.get(key) or "").strip():
cols_missing.append(label)
if not cols_missing:
return None
campos = ", ".join(cols_missing)
return {
"line": line_num,
"col": "",
"msg": f"{MSG_OBLIGATORIOS.format(campos=campos)} {MSG_OBLIGATORIOS_SOLUCION}",
}
def validate_row_fecha_pedimento(
row: Dict[str, Any],
col: str,
line_num: int,
date_format_preference: Optional[str] = None,
) -> Optional[Dict[str, Any]]:
"""Valida que el campo de fecha sea válido (longitud ≤10, parseable)."""
val = (row.get(col) or "").strip()
if not val:
return None
if len(val) > FECHA_MAX_LEN:
return {
"line": line_num,
"col": col,
"msg": f"Error: ({col}) La fecha supera la longitud de caracteres.",
}
if parse_date(val, date_format_preference) is None:
format_label = (
DATE_FORMAT_PREFERENCE_MAP.get(date_format_preference, "##/##/####")
if date_format_preference
else "##/##/####"
)
return {
"line": line_num,
"col": col,
"msg": f"Error: ({col}) La fecha no coincide con el formato ({format_label}).",
}
return None
def validate_row_tipo_operacion(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Col B: si tiene valor debe ser I o E."""
val = (row.get("TIPO_OPERACION") or "").strip().upper()
if not val:
return None
if val not in ("I", "E"):
return {
"line": line_num,
"col": "TIPO_OPERACION",
"msg": f"{MSG_TIPO_INVALIDO.format(val=row.get('TIPO_OPERACION'))} {MSG_TIPO_SOLUCION}",
}
return None
def validate_row_clave_regimen_tipo(
row: Dict[str, Any],
line_num: int,
valid_clave_regimen_tipo: Set[Tuple[str, str, str]],
) -> Optional[Dict[str, Any]]:
"""Combinación B+C+D debe existir en CodePedimentoRegimen."""
b = (row.get("TIPO_OPERACION") or "").strip().upper()
c = (row.get("CLAVE_PEDIMENTO") or "").strip().upper()
d = (row.get("REGIMEN") or "").strip().upper()
if not b or not c or not d:
return None
key = (c, d, b)
if valid_clave_regimen_tipo and key not in valid_clave_regimen_tipo:
# Opcional: listar combinaciones válidas para esa clave
comb = ", ".join(
f"operación {t} con clave {cp} y régimen {r}"
for cp, r, t in valid_clave_regimen_tipo
if cp == c
) or "ninguna para esta clave"
return {
"line": line_num,
"col": "TIPO_OPERACION",
"msg": f"{MSG_COMBINACION.format(b=b, c=c, d=d)} {MSG_COMBINACION_SOLUCION.format(comb=comb)}",
}
return None
def validate_row_aduana(
row: Dict[str, Any],
line_num: int,
valid_aduana_seccion: Set[str],
) -> Optional[Dict[str, Any]]:
"""Col H: si tiene valor debe existir en CustomsSection."""
val = (row.get("ADUANA_SECCION_CRUCE") or "").strip()
if not val:
return None
# customs_code puede ser 3 caracteres; si Col H trae más, tomar primeros 3
code = val[:3] if len(val) >= 3 else val
if valid_aduana_seccion and code not in valid_aduana_seccion:
return {
"line": line_num,
"col": "ADUANA_SECCION_CRUCE",
"msg": f"{MSG_ADUANA_NO_EXISTE.format(val=val)} {MSG_ADUANA_SOLUCION}",
}
return None
def validate_row_ind_con(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Col J: si tiene valor debe ser IND o CON."""
val = (row.get("INDIVIDUAL_CONSOLIDADO") or "").strip().upper()
if not val:
return None
if val not in ("IND", "CON"):
return {
"line": line_num,
"col": "INDIVIDUAL_CONSOLIDADO",
"msg": f"{MSG_IND_CON_INVALIDO.format(val=val)} {MSG_IND_CON_SOLUCION}",
}
return None
def validate_row_transporte(
row: Dict[str, Any],
line_num: int,
col_key: str,
col_label: str,
valid_anexo22_claves: Set[str],
) -> Optional[Dict[str, Any]]:
"""Valida una columna de método de transporte (K, L o M) contra Anexo 22. Si el conjunto está vacío, no se valida."""
val = (row.get(col_key) or "").strip().upper()
if not val:
return None
if valid_anexo22_claves and val not in valid_anexo22_claves:
return {
"line": line_num,
"col": col_key,
"msg": f"{MSG_TRANSP_NO_EXISTE.format(col=col_label, val=val)} {MSG_TRANSP_SOLUCION}",
}
return None
def validaciones_pedimento(
row: Dict[str, Any],
line_num: int,
valid_clave_regimen_tipo: Set[Tuple[str, str, str]],
valid_aduana_seccion: Set[str],
valid_anexo22_claves: Set[str],
valid_patentes: Set[str],
short_name_to_id: Optional[Dict[str, int]] = None,
date_format_preference: Optional[str] = None,
) -> Optional[Dict[str, Any]]:
"""Orquestador: formato PEDIMENTO si aplica, tipo B, combinación B+C+D, aduana H, patente, IND/CON J, transporte K/L/M, fechas E/F/G. CLIENTE_SHORT_NAME opcional."""
if (row.get("PEDIMENTO") or "").strip():
err = validate_row_pedimento_format(row, line_num)
if err:
return err
# CLIENTE_SHORT_NAME opcional: si viene valor se valida que exista en catálogo; si viene vacío no se exige
if short_name_to_id is not None:
err = check_optional_short_name(row, "CLIENTE_SHORT_NAME", short_name_to_id, line_num)
if err:
return err
err = validate_row_patente(row, line_num, valid_patentes)
if err:
return err
err = validate_row_tipo_operacion(row, line_num)
if err:
return err
err = validate_row_clave_regimen_tipo(row, line_num, valid_clave_regimen_tipo)
if err:
return err
err = validate_row_aduana(row, line_num, valid_aduana_seccion)
if err:
return err
err = validate_row_ind_con(row, line_num)
if err:
return err
err = validate_row_transporte(
row, line_num, "MET_TRANSP_ENTRADA", "K", valid_anexo22_claves
)
if err:
return err
err = validate_row_transporte(
row, line_num, "MET_TRANSP_ARRIVO", "L", valid_anexo22_claves
)
if err:
return err
err = validate_row_transporte(
row, line_num, "MET_TRANSP_SALIDA", "M", valid_anexo22_claves
)
if err:
return err
for col in ("FECHA_INICIO", "FECHA_FINAL", "FECHA_PAGO"):
err = validate_row_fecha_pedimento(
row, col, line_num, date_format_preference
)
if err:
return err
return None
def validate_row_pedimento_fk(
# --- Legacy (layout sin PEDIMENTO único): mantener para compatibilidad ---
def validate_row_pedimento_required_legacy(
row: Dict[str, Any], line_num: int
) -> Optional[Dict[str, Any]]:
"""Requeridos para layout legacy: AÑO, ADUANA (ADUANA_SECCION_CRUCE), PATENTE, NUMERO."""
for col, max_len in [
("AÑO", 2),
("ADUANA_SECCION_CRUCE", 3),
("PATENTE", 4),
("NUMERO", 7),
]:
val = (row.get(col) or "").strip()
if not val:
return {"line": line_num, "col": col, "msg": "Requerido"}
if len(val) > max_len:
return {"line": line_num, "col": col, "msg": f"Máximo {max_len} caracteres"}
return None
def validate_row_pedimento_fk_legacy(
row: Dict[str, Any],
line_num: int,
valid_client_ids: Set[int],
short_name_to_id: Dict[str, int],
valid_regimes: Set[str],
valid_pedimento_codes: Set[str],
) -> Optional[Dict[str, Any]]:
err = check_int_in_set(row, "CLIENTE_ID", valid_client_ids, line_num)
"""FK para layout legacy (CLAVE_PEDIMENTO o CODIGO_PEDIMENTO, REGIMEN). CLIENTE_SHORT_NAME opcional."""
err = check_optional_short_name(row, "CLIENTE_SHORT_NAME", short_name_to_id, line_num)
if err:
return err
# Template puede dar CLAVE_PEDIMENTO o CODIGO_PEDIMENTO según alias
code = (row.get("CLAVE_PEDIMENTO") or row.get("CODIGO_PEDIMENTO") or "").strip()
if not code:
return {"line": line_num, "col": "CLAVE_PEDIMENTO", "msg": "Requerido"}
if len(code) > 2:
return {"line": line_num, "col": "CLAVE_PEDIMENTO", "msg": "Máximo 2 caracteres"}
if valid_pedimento_codes and code not in valid_pedimento_codes:
return {"line": line_num, "col": "CLAVE_PEDIMENTO", "msg": "No existe en código pedimento"}
err = check_in_set(
row, "CODIGO_PEDIMENTO", valid_pedimento_codes, line_num,
max_len=PEDIMENTO_CODE_MAX, catalog_name="código pedimento",
)
if err:
return err
err = check_in_set(
row, "REGIMEN", valid_regimes, line_num,
max_len=REGIMEN_MAX, catalog_name="régimen",
row, "REGIMEN", valid_regimes, line_num, max_len=3, catalog_name="régimen"
)
if err:
return err
return None
def validate_row_pedimento_optionals(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
def validate_row_pedimento_optionals_legacy(
row: Dict[str, Any], line_num: int
) -> Optional[Dict[str, Any]]:
"""Opcionales para layout legacy."""
err = check_optional_max_length(row, "ESTATUS", 30, line_num)
if err:
return err
@@ -62,16 +442,16 @@ def validate_row_pedimento_optionals(row: Dict[str, Any], line_num: int) -> Opti
if err:
return err
err = check_optional_in_set(
row, "TIPO_OPERACION", {"imp", "exp"}, line_num,
"Debe ser imp o exp",
row, "TIPO_OPERACION", {"imp", "exp"}, line_num, "Debe ser imp o exp"
)
if err:
return err
err = check_optional_in_set(
row, "TIPO_PEDIMENTO",
row,
"TIPO_PEDIMENTO",
{"normal", "consolidated", "complementary", "automobile"},
line_num,
"Tipo no válido (normal, consolidated, complementary, automobile)",
"Tipo no válido",
)
if err:
return err

119
backend/api/v1/modules/a76/layouts_csv/pedmientos/validators/create.py
View File

@@ -1,37 +1,126 @@
"""
Punto de entrada de validación para import de una fila pedimento.
Recibe conjuntos FK (valid_client_ids, valid_regimes, valid_pedimento_codes) para validar referencias.
Flujo Clarion:
- Modo actualizar: crear si no existe, actualizar si existe. Si existe → VALIDA_PARCIAL; si no existe → VALIDA_TODA.
- Modo crear: crear siempre; si existe reemplazar. Siempre VALIDA_TODA.
Legacy: si la fila no tiene PEDIMENTO (Col A Clarion), se usa validación por AÑO/ADUANA/PATENTE/NUMERO.
"""
from typing import Dict, Any, Optional, Set
from typing import Dict, Any, Optional, Set, Tuple
from ..template_config import is_clarion_layout, parse_pedimento_col_a
from ..common.fk_loader import pedimento_key_from_parsed
from .common import (
validate_row_pedimento_required,
validate_row_pedimento_fk,
validate_row_pedimento_optionals,
validate_row_desfase_pedimento,
validate_row_pedimento_required_col_a,
validate_row_pedimento_format,
validate_row_pedimento_required_three_cols,
validate_row_pedimento_required_full,
validaciones_pedimento,
validate_row_pedimento_required_legacy,
validate_row_pedimento_fk_legacy,
validate_row_pedimento_optionals_legacy,
)
def _clarion_pedimento_key(row: Dict[str, Any]) -> Optional[str]:
"""Construye la clave (year|customs_office|license|pedimento_number) desde fila Clarion (3 cols o PEDIMENTO)."""
año = (row.get("AÑO") or "").strip()
patente = (row.get("PATENTE") or "").strip()
numero = (row.get("NUMERO") or "").strip()
if año and patente and numero:
aduana = (row.get("ADUANA_SECCION_CRUCE") or "").strip()[:3]
return pedimento_key_from_parsed(año[:2], aduana, patente[:4], numero[:7])
parsed = parse_pedimento_col_a((row.get("PEDIMENTO") or "").strip())
if not parsed:
return None
year, license_val, pedimento_number = parsed
aduana = (row.get("ADUANA_SECCION_CRUCE") or "").strip()[:3]
return pedimento_key_from_parsed(year, aduana, license_val, pedimento_number)
def validate_row_pedimento(
row: Dict[str, Any],
line_num: int,
valid_client_ids: Set[int],
short_name_to_id: Dict[str, int],
valid_regimes: Set[str],
valid_pedimento_codes: Set[str],
valid_clave_regimen_tipo: Set[Tuple[str, str, str]],
valid_aduana_seccion: Set[str],
existing_pedimento_keys: Set[str],
valid_anexo22_claves: Set[str],
valid_patentes: Set[str],
actualizar: bool = False,
raw_row: Optional[Dict[str, Any]] = None,
date_format_preference: Optional[str] = None,
) -> Optional[Dict[str, Any]]:
"""
Valida una fila de CSV de pedimentos.
Requeridos: AÑO, ADUANA, PATENTE, NUMERO, CLIENTE_ID, CODIGO_PEDIMENTO, REGIMEN (y FKs en catálogos).
Opcionales: ESTATUS, decimales, TIPO_OPERACION, TIPO_PEDIMENTO.
- Si raw_row está presente, se valida desfase (Col N) primero.
- Layout Clarion (PEDIMENTO con valor): flujo VALIDA_TODA / VALIDA_PARCIAL según actualizar y existing_pedimento_keys.
- Layout legacy (sin PEDIMENTO): requeridos AÑO, ADUANA, PATENTE, NUMERO, CODIGO_PEDIMENTO, REGIMEN; CLIENTE_SHORT_NAME opcional.
"""
err = validate_row_pedimento_required(row, line_num)
if err:
return err
err = validate_row_pedimento_fk(
row, line_num, valid_client_ids, valid_regimes, valid_pedimento_codes
if raw_row is not None:
err = validate_row_desfase_pedimento(raw_row, line_num)
if err:
return err
if not is_clarion_layout(row):
err = validate_row_pedimento_required_legacy(row, line_num)
if err:
return err
err = validate_row_pedimento_fk_legacy(
row, line_num, short_name_to_id, valid_regimes, valid_pedimento_codes
)
if err:
return err
return validate_row_pedimento_optionals_legacy(row, line_num)
# Layout Clarion: validar Cols 1-3 (AÑO, PATENTE, NUMERO) o Col A (PEDIMENTO) si compatibilidad
has_three_cols = (
(row.get("AÑO") or "").strip()
and (row.get("PATENTE") or "").strip()
and (row.get("NUMERO") or "").strip()
)
if has_three_cols:
err = validate_row_pedimento_required_three_cols(row, line_num)
else:
err = validate_row_pedimento_required_col_a(row, line_num)
if not err:
err = validate_row_pedimento_format(row, line_num)
if err:
return err
err = validate_row_pedimento_optionals(row, line_num)
key = _clarion_pedimento_key(row)
# Actualizar: si existe → validación parcial; si no existe → validación completa (crear).
# Crear: siempre validación completa (crear siempre; si existe reemplazar).
use_partial = (
actualizar
and key is not None
and key in existing_pedimento_keys
)
if use_partial:
return validaciones_pedimento(
row,
line_num,
valid_clave_regimen_tipo,
valid_aduana_seccion,
valid_anexo22_claves,
valid_patentes,
short_name_to_id=short_name_to_id,
date_format_preference=date_format_preference,
)
err = validate_row_pedimento_required_full(row, line_num)
if err:
return err
return None
return validaciones_pedimento(
row,
line_num,
valid_clave_regimen_tipo,
valid_aduana_seccion,
valid_anexo22_claves,
valid_patentes,
short_name_to_id=short_name_to_id,
date_format_preference=date_format_preference,
)

2
backend/api/v1/modules/a76/pedmientos/dtos/pedimentos.py
View File

@@ -107,7 +107,7 @@ class PedimentosCreate(PedimentosBase):
customs_office: str = Field(..., max_length=3, description="Customs office")
license: str = Field(..., max_length=4, description="License")
pedimento_number: str = Field(..., max_length=7, description="Pedimento number")
client_id: int = Field(..., description="Client ID")
client_id: Optional[int] = Field(None, description="Client ID (opcional)")
# operation_type, pedimento_type, status son opcionales - se pueden llenar después
pedimento_code: str = Field(..., max_length=2, description="Pedimento key")
regime: str = Field(..., max_length=3, description="Regime")

4
backend/api/v1/modules/a76/pedmientos/models/pedimentos.py
View File

@@ -128,7 +128,9 @@ class Pedimentos(Base, TenantScopedMixin, TimestampMixin):
customs_office: Mapped[str] = mapped_column(String(3))
license: Mapped[str] = mapped_column(String(4))
pedimento_number: Mapped[str] = mapped_column(String(7))
client_id: Mapped[int] = mapped_column(Integer)
client_id: Mapped[Optional[int]] = mapped_column(
Integer, nullable=True
) # Opcional: no requerido para el pedimento
operation_type: Mapped[OperationType] = mapped_column(String(3))
pedimento_type: Mapped[PedimentoType] = mapped_column(String(20))
pedimento_code: Mapped[str] = mapped_column(String(2))