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

feature/alembic-daf

Browse Source
This commit is contained in:
hreyes
2026-05-12 09:00:32 -06:00
parent b0ce6c8a5a
commit ee9ac4f64f
19 changed files with 1203 additions and 717 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
backend/api/v1/modules/a76/invoices/common/common_validators.py
View File

@@ -389,7 +389,8 @@ def validate_common(
)
if pedimento.pedimento_type == "consolidated":
if not pedimento.pedimento_dates:
pd_dates = pedimento.pedimento_dates
if not pd_dates:
errors.add_error(
field="compliance_mx.pedimento_id",
message="El Pedimento seleccionado no tiene fechas registradas.",
@@ -404,27 +405,30 @@ def validate_common(
if hasattr(invoice.invoice_date, "date")
else invoice.invoice_date
)
entry_date = (
pedimento.pedimento_dates.entry_date.date()
if hasattr(pedimento.pedimento_dates.entry_date, "date")
else pedimento.pedimento_dates.entry_date
# Periodo consolidado: si existe start_date en catálogo, es el inicio del rango;
# si no, se usa entry_date (paridad con validación CSV).
period_start_src = getattr(pd_dates, "start_date", None) or pd_dates.entry_date
period_start_date = (
period_start_src.date()
if hasattr(period_start_src, "date")
else period_start_src
)
end_date = (
pedimento.pedimento_dates.end_date.date()
if hasattr(pedimento.pedimento_dates.end_date, "date")
else pedimento.pedimento_dates.end_date
pd_dates.end_date.date()
if hasattr(pd_dates.end_date, "date")
else pd_dates.end_date
)
if pedimento.pedimento_dates and (invoice_date < entry_date or invoice_date > end_date):
errors.add_error(
field="invoice_date",
message=f"La Fecha de la Factura {invoice.invoice_date} no está dentro del rango de fechas del Pedimento {pedimento.customs_office}-{pedimento.license}-{pedimento.pedimento_number}.",
solution=[
f"Capturar una Fecha de Factura, entre la Fecha de Inicio: {pedimento.pedimento_dates.entry_date} y la Fecha Final: {pedimento.pedimento_dates.end_date} ."
],
code="DATE_OUT_OF_RANGE",
value=invoice.invoice_date,
)
if invoice_date < period_start_date or invoice_date > end_date:
errors.add_error(
field="invoice_date",
message=f"La Fecha de la Factura {invoice.invoice_date} no está dentro del rango de fechas del Pedimento {pedimento.customs_office}-{pedimento.license}-{pedimento.pedimento_number}.",
solution=[
f"Capturar una Fecha de Factura, entre la Fecha de Inicio: {period_start_src} y la Fecha Final: {pd_dates.end_date} ."
],
code="DATE_OUT_OF_RANGE",
value=invoice.invoice_date,
)
# Remesa check
if pedimento.pedimento_type == "consolidated":

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

@@ -42,8 +42,10 @@ def _iter_jsonl_rows(error_path: str) -> Iterable[Dict[str, Any]]:
def download_scan_errors_csv_stream(job_type: str, job_id: str, *, filename_prefix: str = "errores") -> StreamingResponse:
"""
Devuelve un StreamingResponse con cabecera:
`linea, columna, mensaje, solucion`.
Devuelve un StreamingResponse leyendo el JSONL del worker.
Columnas (paridad con el payload JSONL / `_invoice_scan_error_row_payload` en facturas):
linea, columna, mensaje, solucion, advertencia, codigo, not_found_reason
"""
error_path = common_storage.error_path_for_job(job_type, job_id)
if not os.path.exists(error_path):
@@ -56,10 +58,20 @@ def download_scan_errors_csv_stream(job_type: str, job_id: str, *, filename_pref
"Content-Disposition": f'attachment; filename="{filename_prefix}_{job_id}.csv"',
}
_CSV_ERRORS_HEADER = [
"linea",
"columna",
"mensaje",
"solucion",
"advertencia",
"codigo",
"not_found_reason",
]
def row_iter():
buffer = io.StringIO()
writer = csv.writer(buffer)
writer.writerow(["linea", "columna", "mensaje", "solucion"])
writer.writerow(_CSV_ERRORS_HEADER)
yield buffer.getvalue()
buffer.seek(0)
buffer.truncate(0)
@@ -73,6 +85,9 @@ def download_scan_errors_csv_stream(job_type: str, job_id: str, *, filename_pref
err.get("col", ""),
err.get("msg", ""),
err.get("solution", ""),
"si" if bool(err.get("warning")) else "no",
err.get("code", ""),
err.get("not_found_reason", ""),
]
)
yield buffer.getvalue()

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

@@ -31,12 +31,13 @@ def scan_result(
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
valid_rows = processed_rows - error_count
out: Dict[str, Any] = {
"status": "waiting_confirmation",
"job_id": job_id,
"total_rows": total,
"error_count": error_count,
"valid_rows": processed_rows - error_count,
"valid_rows": valid_rows,
# Para no saturar el front: devolvemos solo un preview.
# El detalle completo se descarga desde CSV usando el job_id.
"errors": errors_detail[:ERRORS_PREVIEW_LIMIT],
@@ -44,6 +45,13 @@ def scan_result(
}
if message:
out["message"] = message
elif valid_rows == 0 and total > 0:
out["message"] = (
"Ninguna fila quedó libre de observaciones en el escaneo. "
"Revise el reporte de errores (incluye advertencias por línea), "
"complete catálogos (clientes/proveedores, transporte, conductores, etc.) "
"y vuelva a cargar, o descargue el CSV de errores para corregir el archivo."
)
return out

93
backend/api/v1/modules/a76/layouts_csv/facturas/routes.py
View File

@@ -1,12 +1,8 @@
from datetime import datetime
from uuid import uuid4
import os
import json
import logging
import csv
import io
from fastapi import APIRouter, UploadFile, File, Form, HTTPException, Depends, Query
from fastapi.responses import StreamingResponse
from sqlalchemy.orm import Session
from typing import Optional, Literal, Dict, Any
@@ -25,6 +21,7 @@ from .tasks import (
)
from .schemas import ImportJobResponse, ImportJobStatus, CommitRequest
from ..common import storage as common_storage
from ..common.error_csv import download_scan_errors_csv_stream
from ..common.responses import normalize_commit_status_payload
from ..common.track_commit_dispatch import dispatch_tracked_layouts_csv_commit
@@ -53,6 +50,19 @@ async def upload_import_file(
Step 1: Upload CSV, save to temp, trigger scan task.
Si se envía template_id, solo se leen las columnas de esa plantilla.
"""
header_company_id = None
try:
if db is not None and getattr(db, "info", None):
header_company_id = db.info.get("rls_company_id")
except Exception:
header_company_id = None
if header_company_id is not None and int(header_company_id) != int(company_id):
logger.warning(
"Facturas upload company mismatch request_company_id=%s db_rls_company_id=%s",
company_id,
header_company_id,
)
# 1. Validate Access & Get Tenant
try:
tenant_id = validate_access_to_resource(db, company_id, current_user, ["csv_upload.process"])
@@ -100,6 +110,14 @@ async def upload_import_file(
raise HTTPException(status_code=500, detail="Failed to queue file for processing.")
# Trigger Celery Task (Async). Worker loads file from Redis / MinIO.
logger.info(
"Queueing facturas scan job=%s tenant_id=%s company_id=%s template_id=%s operation_type=%s",
job_id,
tenant_id,
company_id,
template_id,
operation_type,
)
track_and_dispatch(
db=db,
task=scan_file,
@@ -128,6 +146,15 @@ async def get_import_status(job_id: str):
if task_result.state == "PENDING":
return {"status": "processing", "progress": 0}
# task_track_started=True: el worker pasa por STARTED (y a veces RECEIVED) antes de SUCCESS/PROGRESS.
# Sin esto, el polling cae en la rama final y devuelve status=failed aunque el scan vaya bien.
if task_result.state in ("STARTED", "RECEIVED"):
info = task_result.info if isinstance(task_result.info, dict) else {}
return {
"status": "processing",
"progress": info.get("current", 0),
"total": info.get("total", 0),
}
if task_result.state == "PROGRESS":
return {
"status": "processing",
@@ -139,6 +166,17 @@ async def get_import_status(job_id: str):
if isinstance(result, dict) and "status" in result:
return normalize_commit_status_payload(result)
return {"status": "finished", "result": result}
# Recuperación: raro pero posible con backend de resultados — el payload ya está pero el estado no es SUCCESS.
raw = getattr(task_result, "result", None)
if isinstance(raw, dict) and raw.get("status") in (
"waiting_confirmation",
"finished",
"warning",
"failed",
):
return normalize_commit_status_payload(raw)
# FAILURE: obtener mensaje real (traceback, result o get(propagate=False))
logger.warning("Import task %s failed: state=%s", job_id, task_result.state)
err_msg = None
@@ -178,54 +216,11 @@ async def get_import_status(job_id: str):
async def download_scan_errors_csv(job_id: str):
"""
Descarga CSV con TODO el detalle de errores del scan (sin límite),
leyendo el archivo JSONL generado por el worker.
leyendo el archivo JSONL generado por el worker (mismas columnas que otros layouts_csv).
"""
# Facturas (impo/exp delega a facturas) usan job_type vacío ("").
error_path = common_storage.error_path_for_job("", job_id)
if not os.path.exists(error_path):
raise HTTPException(status_code=404, detail="Archivo de errores no encontrado. Vuelve a escanear o intenta más tarde.")
def row_iter():
buffer = io.StringIO()
writer = csv.writer(buffer)
writer.writerow(["linea", "columna", "mensaje", "solucion"])
yield buffer.getvalue()
buffer.seek(0)
buffer.truncate(0)
with open(error_path, "r", encoding="utf-8") as f:
for line in f:
if not line.strip():
continue
try:
payload = json.loads(line)
except Exception:
continue
# Algunos flujos pueden escribir un dict por línea o una lista de dicts.
items = payload if isinstance(payload, list) else [payload]
for err in items:
if not isinstance(err, dict):
continue
buffer.seek(0)
buffer.truncate(0)
writer.writerow(
[
err.get("line", ""),
err.get("col", ""),
err.get("msg", ""),
err.get("solution", ""),
]
)
yield buffer.getvalue()
buffer.seek(0)
buffer.truncate(0)
headers = {
"Content-Disposition": f'attachment; filename="errores_{job_id}.csv"'
}
return StreamingResponse(row_iter(), media_type="text/csv; charset=utf-8", headers=headers)
return download_scan_errors_csv_stream("", job_id, filename_prefix="errores")
@router.post("/{job_id}/commit")

890
backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py
View File

File diff suppressed because it is too large Load Diff

13
backend/api/v1/modules/a76/layouts_csv/facturas/template_config.py
View File

@@ -19,11 +19,12 @@ from .invoice_csv_column_hints import apply_invoice_csv_hints
TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
# --- Encabezado factura: Impo Temp (EstructuraEncFacImpoTemp.xls) - Clarion A-AD ---
"imp_temp_header": [
{"canonical": "PEDIMENTO", "aliases": ["NUMERO PEDIMENTO", "PEDIMENTO NUMERO", "NUMERO DE PEDIMENTO", "PED"]},
{"canonical": "PEDIMENTO", "aliases": ["NUMERO PEDIMENTO", "PEDIMENTO NUMERO", "NUMERO DE PEDIMENTO", "PED", "NO. PEDIMENTO", "PEDIMENTO NO.", "NO PEDIMENTO"]},
{"canonical": "REMESA"},
{"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA", "ID"]},
{"canonical": "FECHA FACTURA", "aliases": ["FECHA"]},
{"canonical": "TIPO DE CAMBIO"},
# Clarion columna F — ITE/ITR cuando el pedimento aún no está en catálogo (pendiente de asignar).
{"canonical": "REGIMEN", "aliases": ["CLAVEDOCUMENTO"]},
{"canonical": "CLAVE PROVEEDOR"},
{"canonical": "CLAVE VENDIDO A"},
@@ -48,7 +49,7 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
{"canonical": "TIPO PESO"},
{"canonical": "E DOCUMENT", "aliases": ["E-DOCUMENT", "EDOCUMENT", "E DOCUMENT"]},
{"canonical": "NUM OPERACION", "aliases": ["NUM. OPERACION", "NUMOPERACION", "NUM OPERACION"]},
{"canonical": "ADUANA DE CRUCE"},
{"canonical": "ADUANA DE CRUCE", "aliases": ["ADUANA", "ADUANA CRUCE", "ADUANA DE CRUCE.", "ADUANA DE CRUCE / SECCION"]},
{"canonical": "OBSERVACIONES E"},
{"canonical": "OBSERVACIONES I"},
{"canonical": "FACTURA ALTERNA"},
@@ -58,7 +59,7 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
],
# --- Encabezado factura: Impo Def (EstructuraEncFacImpoDef.xls) - 30 columnas Clarion ---
"imp_def_header": [
{"canonical": "PEDIMENTO", "aliases": ["NUMERO PEDIMENTO", "PEDIMENTO NUMERO", "NUMERO DE PEDIMENTO", "PED"]},
{"canonical": "PEDIMENTO", "aliases": ["NUMERO PEDIMENTO", "PEDIMENTO NUMERO", "NUMERO DE PEDIMENTO", "PED", "NO. PEDIMENTO", "PEDIMENTO NO.", "NO PEDIMENTO"]},
{"canonical": "REMESA"},
{"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA", "ID"]},
{"canonical": "FECHA FACTURA", "aliases": ["FECHA"]},
@@ -87,7 +88,7 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
{"canonical": "TIPO PESO"},
{"canonical": "E DOCUMENT", "aliases": ["E-DOCUMENT", "EDOCUMENT", "E DOCUMENT"]},
{"canonical": "NUM OPERACION", "aliases": ["NUM. OPERACION", "NUMOPERACION", "NUM OPERACION"]},
{"canonical": "ADUANA DE CRUCE"},
{"canonical": "ADUANA DE CRUCE", "aliases": ["ADUANA", "ADUANA CRUCE", "ADUANA DE CRUCE.", "ADUANA DE CRUCE / SECCION"]},
{"canonical": "OBSERVACIONES E"},
{"canonical": "OBSERVACIONES I"},
],
@@ -96,7 +97,7 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
# CLAVE VENDIDO A:, CLAVE ENVIADO A, AGENTE ADUANAL, ... MANIFIESTO, E-DOCUMENT, NUM. OPERACION,
# ENVIADO POR, ADUANA DE CRUCE, OBSERVACIONES E, OBSERVACIONES I, FACTURA ALTERNA
"exp_def_header": [
{"canonical": "PEDIMENTO", "aliases": ["NUMERO PEDIMENTO", "PEDIMENTO NUMERO", "NUMERO DE PEDIMENTO", "PED"]},
{"canonical": "PEDIMENTO", "aliases": ["NUMERO PEDIMENTO", "PEDIMENTO NUMERO", "NUMERO DE PEDIMENTO", "PED", "NO. PEDIMENTO", "PEDIMENTO NO.", "NO PEDIMENTO"]},
{"canonical": "REMESA"},
{"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA", "ID"]},
{"canonical": "FECHA FACTURA", "aliases": ["FECHA"]},
@@ -126,7 +127,7 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
{"canonical": "E DOCUMENT", "aliases": ["E-DOCUMENT", "EDOCUMENT", "E DOCUMENT"]},
{"canonical": "NUM OPERACION", "aliases": ["NUM. OPERACION", "NUMOPERACION", "NUM OPERACION"]},
{"canonical": "ENVIADO POR"},
{"canonical": "ADUANA DE CRUCE"},
{"canonical": "ADUANA DE CRUCE", "aliases": ["ADUANA", "ADUANA CRUCE", "ADUANA DE CRUCE.", "ADUANA DE CRUCE / SECCION"]},
{"canonical": "OBSERVACIONES E"},
{"canonical": "OBSERVACIONES I"},
{"canonical": "FACTURA ALTERNA"},

2
backend/api/v1/modules/a76/layouts_csv/facturas/validators/__init__.py
View File

@@ -21,6 +21,7 @@ from .series_expo import (
validate_row_series_expo,
row_to_series_normalized_expo,
)
from .consolidated_invoice_dates import validate_consolidated_invoice_date_csv
__all__ = [
"validate_row_encabezados_impo_temp",
@@ -37,4 +38,5 @@ __all__ = [
"parse_pedimento_col_a",
"parse_pedimento_col_a_impo_def",
"_pedimento_key_from_parsed",
"validate_consolidated_invoice_date_csv",
]

107
backend/api/v1/modules/a76/layouts_csv/facturas/validators/consolidated_invoice_dates.py Normal file
View File

@@ -0,0 +1,107 @@
"""
Validación de FECHA FACTURA vs pedimento consolidado en CSV de encabezados.
Paridad con `validate_common` (factura manual) cuando `pedimento_type == "consolidated"`:
- Sin fechas de pedimento en catálogo → equivalente a MISSING_PEDIMENTO_DATES.
- Rango permitido: [start_date, end_date] si `start_date` está capturada en el catálogo;
si no, [entry_date, end_date] (mismo criterio que captura manual / periodo consolidado).
"""
from datetime import date, datetime
from typing import Any, Dict, Optional
def _to_calendar_date(val: Any) -> Optional[date]:
if val is None:
return None
if isinstance(val, datetime):
return val.date()
if isinstance(val, date):
return val
if hasattr(val, "date"):
return val.date() # type: ignore[union-attr]
return None
def validate_consolidated_invoice_date_csv(
*,
line_num: int,
pedimento_key_display: str,
pedimento_type: str,
invoice_date_parsed: Optional[datetime],
entry_raw: Any,
end_raw: Any,
start_raw: Any = None,
) -> Optional[Dict[str, Any]]:
"""
Retorna dict de error compatible con validadores CSV (line, col, msg, …) o None si aplica/no hay error.
No aplica si el pedimento no es consolidado.
"""
pt = (pedimento_type or "").strip().lower()
if pt != "consolidated":
return None
if end_raw is None:
return {
"line": line_num,
"col": "PEDIMENTO",
"msg": (
f"Error: (Celda A{line_num}) El Pedimento {pedimento_key_display} no tiene fechas registradas "
"en el catálogo."
),
"solution": "Verifica las fechas del Pedimento en el catálogo",
"code": "MISSING_PEDIMENTO_DATES",
}
range_low_raw = start_raw if start_raw is not None else entry_raw
if range_low_raw is None:
return {
"line": line_num,
"col": "PEDIMENTO",
"msg": (
f"Error: (Celda A{line_num}) El Pedimento {pedimento_key_display} no tiene fechas registradas "
"en el catálogo."
),
"solution": "Verifica las fechas del Pedimento en el catálogo",
"code": "MISSING_PEDIMENTO_DATES",
}
if not invoice_date_parsed:
return None
entry = _to_calendar_date(range_low_raw)
end = _to_calendar_date(end_raw)
if entry is None or end is None:
return {
"line": line_num,
"col": "PEDIMENTO",
"msg": (
f"Error: (Celda A{line_num}) El Pedimento {pedimento_key_display} no tiene fechas registradas "
"en el catálogo."
),
"solution": "Verifica las fechas del Pedimento en el catálogo",
"code": "MISSING_PEDIMENTO_DATES",
}
if isinstance(invoice_date_parsed, datetime):
inv_d = invoice_date_parsed.date()
elif isinstance(invoice_date_parsed, date):
inv_d = invoice_date_parsed
else:
return None
if inv_d < entry or inv_d > end:
return {
"line": line_num,
"col": "FECHA FACTURA",
"msg": (
f"Error: (Celda D{line_num} y A{line_num}) La Fecha de la Factura no corresponde al rango "
f"de fechas del Pedimento {pedimento_key_display}."
),
"solution": (
"Capturar una Fecha de Factura dentro del rango de fechas del pedimento consolidado "
"registrado en el catálogo."
),
"code": "DATE_OUT_OF_RANGE",
}
return None

33
backend/api/v1/modules/a76/layouts_csv/facturas/validators/encabezados_expo.py
View File

@@ -7,6 +7,7 @@ NUM. OPERACION (AA), ENVIADO POR (AB), ADUANA DE CRUCE (AC), OBSERVACIONES E/I,
from datetime import datetime
from typing import Any, Dict, List, Optional, Set
from .pedimento_resolution import resolve_pedimento_candidates
from .consolidated_invoice_dates import validate_consolidated_invoice_date_csv
from .encabezados_impo_temp import (
_clip,
@@ -148,6 +149,13 @@ def _validaciones_pedimento_remesa_expo(
pedimento_rows=pedimento_rows,
)
if ped_resolved["status"] == "not_found":
if ped_resolved.get("not_found_reason") == "customs_mismatch":
return _err(
line_num,
"PEDIMENTO",
f"Error: (Celda A{line_num}) El pedimento {col_a} está en el catálogo pero la aduana no coincide con la registrada. "
"Revise columna A y el alta del pedimento en el sistema.",
)
return _err(
line_num,
"PEDIMENTO",
@@ -219,20 +227,17 @@ def _validaciones_pedimento_remesa_expo(
# Rango de fechas si pedimento consolidado (omitir si recalcular_fecha_pedimentos = True, paridad Clarion)
if not recalcular_fecha_pedimentos:
pedimento_type = (ped_info.get("pedimento_type") or "").strip().lower()
if pedimento_type == "consolidated" and invoice_date_parsed and ped_info.get("entry_date") and ped_info.get("end_date"):
entry = ped_info["entry_date"]
end = ped_info["end_date"]
if hasattr(entry, "date"):
entry = entry.date()
if hasattr(end, "date"):
end = end.date()
inv_d = invoice_date_parsed.date() if hasattr(invoice_date_parsed, "date") else invoice_date_parsed
if inv_d < entry or inv_d > end:
return _err(
line_num,
"FECHA FACTURA",
f"Error: (Celda D{line_num} y A{line_num}) La Fecha de la Factura no corresponde al rango de fechas del Pedimento {col_a}.",
)
err_consolidated = validate_consolidated_invoice_date_csv(
line_num=line_num,
pedimento_key_display=col_a,
pedimento_type=pedimento_type,
invoice_date_parsed=invoice_date_parsed,
entry_raw=ped_info.get("entry_date"),
end_raw=ped_info.get("end_date"),
start_raw=ped_info.get("start_date"),
)
if err_consolidated:
return err_consolidated
if not autonumerar_remesas and not col_b:
return _err(

63
backend/api/v1/modules/a76/layouts_csv/facturas/validators/encabezados_impo_def.py
View File

@@ -7,6 +7,7 @@ moneda, tipo peso y tipo cambio (misma lógica que TEM).
from datetime import datetime
from typing import Any, Dict, List, Optional, Set, Tuple
from .pedimento_resolution import resolve_pedimento_candidates
from .consolidated_invoice_dates import validate_consolidated_invoice_date_csv
# Reutilizar del TEM: helpers y validaciones de catálogos (claves/short names), transporte, moneda, tipo peso, tipo cambio
from .encabezados_impo_temp import (
@@ -26,26 +27,34 @@ from .encabezados_impo_temp import (
# Impo Def: régimen único y pedimento formato ##-####-####### (15 chars)
REGIMEN_IMD = "IMD"
MAX_LEN_PEDIMENTO_DEF = 15
MAX_LEN_PEDIMENTO_DEF = 16
ALLOWED_DEF_REGIMENS = {"IMD", "V1", "A1", "F4", "F5", "D1"}
MAX_LEN_FACTURA = 15 # mismo que TEM
def parse_pedimento_col_a_impo_def(pedimento_str: str) -> Optional[Tuple[str, str, str]]:
"""
Parsea Col A (PEDIMENTO) formato ##-####-####### (15 caracteres).
Guiones en posiciones 3 y 8 (1-based): índices 2 y 7. Retorna (aduana_2, patente_4, numero_7) o None.
Ejemplo válido: 01-1234-2312412
Parsea Col A (PEDIMENTO) formato ##-####-####### (15 chars) o ###-####-####### (16 chars).
Retorna (aduana, patente, numero) o None.
"""
if not pedimento_str or not isinstance(pedimento_str, str):
return None
s = (pedimento_str or "").strip()
if len(s) != MAX_LEN_PEDIMENTO_DEF:
if not (15 <= len(s) <= 16):
return None
if s[2:3] != "-" or s[7:8] != "-":
# Buscamos los guiones. Deben ser 2.
parts = s.split("-")
if len(parts) != 3:
return None
part0, part1, part2 = s[0:2], s[3:7], s[8:15]
part0, part1, part2 = parts
if not (2 <= len(part0) <= 3) or len(part1) != 4 or len(part2) != 7:
return None
if not part0.isdigit() or not part1.isdigit() or not part2.isdigit():
return None
return (part0, part1, part2)
@@ -143,6 +152,13 @@ def _validaciones_pedimento_remesa_def(
pedimento_rows=pedimento_rows,
)
if ped_resolved["status"] == "not_found":
if ped_resolved.get("not_found_reason") == "customs_mismatch":
return _err(
line_num,
"PEDIMENTO",
f"Error: (Celda A{line_num}) El pedimento {col_a} está en el catálogo pero la aduana no coincide con la registrada. "
"Revise columna A y el alta del pedimento en el sistema.",
)
return _err(
line_num,
"PEDIMENTO",
@@ -159,17 +175,17 @@ def _validaciones_pedimento_remesa_def(
ped_info = ped_resolved["pedimento"]
regimen_ped = (ped_info.get("regime") or "").strip().upper()
if regimen_ped != REGIMEN_IMD:
if regimen_ped not in ALLOWED_DEF_REGIMENS:
return _err(
line_num,
"PEDIMENTO",
f"Error: (Celda A{line_num}) El Número de Pedimento: {col_a} tiene el Régimen {regimen_ped}, no válido para este tipo de movimiento. Válidos: IMD.",
f"Error: (Celda A{line_num}) El Número de Pedimento: {col_a} tiene el Régimen {regimen_ped}, no válido para este tipo de movimiento. Válidos: {', '.join(sorted(ALLOWED_DEF_REGIMENS))}.",
)
if col_f and col_f != REGIMEN_IMD:
if col_f and col_f not in ALLOWED_DEF_REGIMENS:
return _err(
line_num,
"REGIMEN",
f"Error: (Celda F{line_num}) El Régimen Aduanero: {col_f} no coincide con el del Pedimento. Debe ser IMD.",
f"Error: (Celda F{line_num}) El Régimen Aduanero: {col_f} no es válido para este tipo de movimiento. Válidos: {', '.join(sorted(ALLOWED_DEF_REGIMENS))}.",
)
if col_f and col_f != regimen_ped:
return _err(
@@ -179,20 +195,17 @@ def _validaciones_pedimento_remesa_def(
)
pedimento_type = (ped_info.get("pedimento_type") or "").strip().lower()
if pedimento_type == "consolidated" and invoice_date_parsed and ped_info.get("entry_date") and ped_info.get("end_date"):
entry = ped_info["entry_date"]
end = ped_info["end_date"]
if hasattr(entry, "date"):
entry = entry.date()
if hasattr(end, "date"):
end = end.date()
inv_d = invoice_date_parsed.date() if hasattr(invoice_date_parsed, "date") else invoice_date_parsed
if inv_d < entry or inv_d > end:
return _err(
line_num,
"FECHA FACTURA",
f"Error: (Celda D{line_num} y A{line_num}) La Fecha de la Factura no corresponde al rango de fechas del Pedimento {col_a}.",
)
err_consolidated = validate_consolidated_invoice_date_csv(
line_num=line_num,
pedimento_key_display=col_a,
pedimento_type=pedimento_type,
invoice_date_parsed=invoice_date_parsed,
entry_raw=ped_info.get("entry_date"),
end_raw=ped_info.get("end_date"),
start_raw=ped_info.get("start_date"),
)
if err_consolidated:
return err_consolidated
if not autonumerar_remesas and not col_b:
return _err(

221
backend/api/v1/modules/a76/layouts_csv/facturas/validators/encabezados_impo_temp.py
View File

@@ -4,10 +4,19 @@ Paridad Clarion: VALIDA_TODA_FACIMPO_TEM, VALIDA_PARCIAL_FACIMPO_TEM, VALIDACION
Mapeo a BD en commit: LLENA_FACIMPO_TEM (tasks.insert_valid_rows).
Estructura CSV: PEDIMENTO (A), REMESA (B), NUMERO FACTURA (C), ... ADUANA DE CRUCE (AB), OBSERVACIONES E/I, FACTURA ALTERNA.
"""
import logging
from datetime import datetime
from decimal import Decimal, InvalidOperation
from typing import Any, Dict, List, Optional, Set, Tuple
from .pedimento_resolution import resolve_pedimento_candidates
from .pedimento_resolution import (
aduana_cruce_matches_catalog,
canonical_customs_office_for_pedimento_key,
normalize_pedimento_number_str,
resolve_pedimento_candidates,
)
from .consolidated_invoice_dates import validate_consolidated_invoice_date_csv
logger = logging.getLogger(__name__)
# Longitudes máximas Clarion
MAX_LEN_PEDIMENTO = 18 # CC-LLLL-NNNNNNN o CCC-LLLL-NNNNNNN (sin año; ej. 01-1234-2312412 o 640-1234-2312412)
@@ -85,45 +94,80 @@ def _parse_int(val: Any) -> Optional[int]:
return None
def parse_pedimento_col_a(pedimento_str: str) -> Optional[Tuple[str, str, str]]:
def _sanitize_pedimento_csv_delimiters(raw: Any) -> str:
"""Guiones Unicode / BOM → ASCII (Excel a veces exporta caracteres distintos de '-')."""
s = str(raw if raw is not None else "").strip().strip("\ufeff")
for ch in ("\u2212", "\u2013", "\u2014"):
s = s.replace(ch, "-")
return s
def _digits_segment_from_csv_cell(seg: Any, *, max_digits: int, min_digits: int = 1) -> Optional[str]:
"""
Parsea Col A (PEDIMENTO) formato CC-LLLL-NNNNNNN o CCC-LLLL-NNNNNNN (18 caracteres, sin año).
Segmento numérico desde CSV: tolera celdas Excel como 3428.0, 7011747.0 o enteros.
"""
raw = str(seg if seg is not None else "").strip().replace(",", "").replace(" ", "")
if not raw:
return None
if raw.endswith(".0") and raw[:-2].isdigit():
raw = raw[:-2]
elif "." in raw or "e" in raw.lower():
try:
fv = float(raw)
if fv < 0 or fv != int(fv):
return None
raw = str(int(fv))
except (ValueError, OverflowError):
return None
if not raw.isdigit():
return None
if len(raw) > max_digits or len(raw) < min_digits:
return None
return raw
def parse_pedimento_col_a(pedimento_str: Any) -> Optional[Tuple[str, str, str]]:
"""
Parsea Col A (PEDIMENTO) formato CC-LLLL-NNNNNNN o CCC-LLLL-NNNNNNN (sin año).
El número puede llevar 1–7 dígitos; se normaliza a 7 con ceros a la izquierda.
Tolera exportación Excel (ej. aduana 7 → 07, 3428.0, 7011747.0).
Retorna (customs_office_2o3, license_4, pedimento_number_7) o None si formato inválido.
"""
if not pedimento_str or not isinstance(pedimento_str, str):
if pedimento_str is None or (isinstance(pedimento_str, str) and not pedimento_str.strip()):
return None
s = (pedimento_str or "").strip()
s = _sanitize_pedimento_csv_delimiters(pedimento_str)
parts = s.split("-")
if len(parts) != 3:
return None
customs_office, license_val, pedimento_number = parts[0], parts[1], parts[2]
if len(customs_office) not in (2, 3) or not customs_office.isdigit():
customs_raw = _digits_segment_from_csv_cell(parts[0], max_digits=3, min_digits=1)
if not customs_raw:
return None
if len(license_val) != 4 or not license_val.isdigit():
if len(customs_raw) == 1:
customs_office = customs_raw.zfill(2)
else:
customs_office = customs_raw
if len(customs_office) not in (2, 3):
return None
if len(pedimento_number) != 7 or not pedimento_number.isdigit():
lic_raw = _digits_segment_from_csv_cell(parts[1], max_digits=4, min_digits=1)
if not lic_raw:
return None
license_val = lic_raw.zfill(4)
num_raw = _digits_segment_from_csv_cell(parts[2], max_digits=7, min_digits=1)
if not num_raw:
return None
pedimento_number = normalize_pedimento_number_str(num_raw)
return (customs_office, license_val, pedimento_number)
def _pedimento_key_from_parsed(customs_office: str, license_val: str, pedimento_number: str) -> str:
"""Clave para lookup: CC-LLLL-NNNNNNN (solo primeros 2 dígitos de aduana)."""
co = (customs_office or "").strip()
# La aduana en el catálogo puede venir con 3 dígitos (String(3)).
# El CSV de encabezados normalmente usa 2 dígitos, por lo que normalizamos:
# - 2 dígitos: se usan tal cual
# - 3 dígitos:
# - usar siempre los primeros 2 (ej. 007 -> 00, 640 -> 64)
# - 1 dígito: left-pad a 2
if len(co) == 3:
co = co[:2]
elif len(co) == 1:
co = co.zfill(2)
else:
co = co[:2]
return f"{co}-{license_val}-{pedimento_number}"
"""Clave para lookup: CC-LLLL-NNNNNNN (aduana canónica 2 dígitos, número 7 dígitos)."""
co = canonical_customs_office_for_pedimento_key(customs_office)
num = normalize_pedimento_number_str(pedimento_number)
lic = (license_val or "").strip()
return f"{co}-{lic}-{num}"
def _patente_from_agente_aduanal(row: Dict[str, Any], default_license: str) -> str:
@@ -141,8 +185,11 @@ def _patente_from_agente_aduanal(row: Dict[str, Any], default_license: str) -> s
return (default_license or "").strip()
def _err(line_num: int, col: str, msg: str) -> Dict[str, Any]:
return {"line": line_num, "col": col, "msg": msg}
def _err(line_num: int, col: str, msg: str, not_found_reason: Optional[str] = None) -> Dict[str, Any]:
out: Dict[str, Any] = {"line": line_num, "col": col, "msg": msg}
if not_found_reason:
out["not_found_reason"] = not_found_reason
return out
# --- Obligatorios VALIDA_TODA (cuando no es actualizar) ---
@@ -170,10 +217,12 @@ def _validaciones_obligatorios_toda(
if tiene_pedimento and not _get(row, "ADUANA DE CRUCE"):
obligatorios.append("(Col.AB) Aduana de Cruce")
if obligatorios:
msg = f"Existen campos vacíos que son obligatorios: {', '.join(obligatorios)}."
logger.warning(f"Line {line_num} rejected: {msg}")
return _err(
line_num,
"ARCHIVO CSV",
f"Existen campos vacíos que son obligatorios: {', '.join(obligatorios)}.",
msg,
)
return None
@@ -218,7 +267,7 @@ def _validaciones_pedimento_remesa(
return _err(
line_num,
"PEDIMENTO",
f"Error: (Celda A{line_num}) El Formato del Pedimento: {col_a} es incorrecto. Use CC-LLLL-NNNNNNN (ej. 01-1234-2312412, 18 caracteres sin año).",
f"Error: (Celda A{line_num}) El Formato del Pedimento: {col_a} es incorrecto. Use CC-LLLL-NNNNNNN (aduana 2–3 dígitos, patente 4, número 1–7 dígitos, sin año).",
)
customs_office, license_val, pedimento_number = parsed
@@ -232,58 +281,84 @@ def _validaciones_pedimento_remesa(
pedimento_rows=pedimento_rows,
)
if ped_resolved["status"] == "not_found":
return _err(
line_num,
"PEDIMENTO",
f"Error: (Celda A{line_num}) El Número de Pedimento: {col_a} no existe en el Catálogo de Pedimentos. "
f"Verifique que esté dado de alta (formato CC-LLLL-NNNNNNN: aduana 2-3, patente 4, número 7, sin año) para esta empresa.",
)
if ped_resolved["status"] == "ambiguous":
if ped_resolved.get("not_found_reason") == "customs_mismatch":
return _err(
line_num,
"PEDIMENTO",
f"Error: (Celda A{line_num}) El pedimento {col_a} está en el catálogo pero la aduana no coincide con la registrada. "
"Revise columna A (aduana en el pedimento) y Col.AB Aduana de cruce frente al alta del pedimento.",
not_found_reason="customs_mismatch",
)
if ped_resolved.get("not_found_reason") == "license_or_patente_mismatch":
return _err(
line_num,
"PEDIMENTO",
f"Error: (Celda A{line_num}) El número de pedimento aparece en el catálogo pero la patente/agente no coincide "
f"con el alta (clave {col_a}). Revise columna A y columna J (agente aduanal).",
not_found_reason="license_or_patente_mismatch",
)
# no_catalog_match: paridad captura manual — pedimento aún no dado de alta; se permite continuar
# si el régimen en columna F es ITE/ITR (validación local sin catálogo).
if not col_f or col_f.upper() not in REGIMENES_VALIDOS:
return _err(
line_num,
"REGIMEN",
f"Error: (Celda F{line_num}) Cuando el pedimento no está en el catálogo, el Régimen (ITE o ITR) es obligatorio "
"para poder importar la factura como pendiente de asignar pedimento.",
not_found_reason="no_catalog_match",
)
ped_info = None
elif ped_resolved["status"] == "ambiguous":
return _err(
line_num,
"PEDIMENTO",
f"Error: (Celda A{line_num}) El Número de Pedimento: {col_a} coincide con múltiples registros. "
"Valide aduana/patente para identificar un único pedimento.",
)
else:
ped_info = ped_resolved["pedimento"]
ped_info = ped_resolved["pedimento"]
if (ped_info.get("operation_type") or "").upper() != "IMP":
return _err(
line_num,
"PEDIMENTO",
f"Error: (Celda A{line_num}) Este Número de Pedimento: {col_a} no está marcado como Importación Temporal.",
)
regimen_ped = (ped_info.get("regime") or "").strip().upper()
if regimen_ped not in REGIMENES_VALIDOS:
return _err(
line_num,
"PEDIMENTO",
f"Error: (Celda A{line_num}) El Pedimento: {col_a} tiene el Régimen {regimen_ped}, no válido (ITE o ITR).",
)
if col_f and col_f not in REGIMENES_VALIDOS:
pass
elif col_f and col_f != regimen_ped:
return _err(
line_num,
"REGIMEN",
f"Error: (Celda F{line_num}) El Régimen Aduanero: {col_f} no coincide con el del Pedimento: {regimen_ped}.",
)
pedimento_type = (ped_info.get("pedimento_type") or "").strip().lower()
if pedimento_type == "consolidated" and invoice_date_parsed and ped_info.get("entry_date") and ped_info.get("end_date"):
entry = ped_info["entry_date"]
end = ped_info["end_date"]
if hasattr(entry, "date"):
entry = entry.date()
if hasattr(end, "date"):
end = end.date()
inv_d = invoice_date_parsed.date() if hasattr(invoice_date_parsed, "date") else invoice_date_parsed
if inv_d < entry or inv_d > end:
if ped_info is not None:
if (ped_info.get("operation_type") or "").upper() != "IMP":
return _err(
line_num,
"FECHA FACTURA",
f"Error: (Celda D{line_num} y A{line_num}) La Fecha de la Factura no corresponde al rango de fechas del Pedimento {col_a}.",
"PEDIMENTO",
f"Error: (Celda A{line_num}) Este Número de Pedimento: {col_a} no está marcado como Importación Temporal.",
)
regimen_ped = (ped_info.get("regime") or "").strip().upper()
if regimen_ped not in REGIMENES_VALIDOS:
msg = f"Error: (Celda A{line_num}) El Pedimento: {col_a} tiene el Régimen {regimen_ped}, no válido (ITE o ITR)."
logger.warning(f"Line {line_num} rejected: {msg}")
return _err(
line_num,
"PEDIMENTO",
msg,
)
if col_f and col_f not in REGIMENES_VALIDOS:
pass
elif col_f and col_f != regimen_ped:
# Relax: allow ITE/ITR interchange for temporal imports
if col_f in REGIMENES_VALIDOS and regimen_ped in REGIMENES_VALIDOS:
pass
else:
return _err(
line_num,
"REGIMEN",
f"Error: (Celda F{line_num}) El Régimen Aduanero: {col_f} no coincide con el del Pedimento: {regimen_ped}.",
)
pedimento_type = (ped_info.get("pedimento_type") or "").strip().lower()
err_consolidated = validate_consolidated_invoice_date_csv(
line_num=line_num,
pedimento_key_display=col_a,
pedimento_type=pedimento_type,
invoice_date_parsed=invoice_date_parsed,
entry_raw=ped_info.get("entry_date"),
end_raw=ped_info.get("end_date"),
start_raw=ped_info.get("start_date"),
)
if err_consolidated:
return err_consolidated
if not autonumerar_remesas and not col_b:
return _err(
@@ -510,7 +585,7 @@ def _validaciones_catalogos(
return _err(line_num, "CLAVE INCOTERM", f"Error: (Celda V{line_num}) La Clave de INCOTERM: {v} no existe en el Catálogo.")
ab = _get(row, "ADUANA DE CRUCE")
if ab and valid_aduana_codes and ab not in valid_aduana_codes:
if ab and valid_aduana_codes and not aduana_cruce_matches_catalog(ab, valid_aduana_codes):
return _err(line_num, "ADUANA DE CRUCE", f"Error: (Celda AB{line_num}) La Aduana de Cruce: {ab} no existe en el Catálogo.")
return None

228
backend/api/v1/modules/a76/layouts_csv/facturas/validators/pedimento_resolution.py
View File

@@ -1,8 +1,118 @@
from typing import Any, Dict, List, Optional, Set
from typing import Any, Dict, List, Optional, Set, Tuple
import logging
logger = logging.getLogger(__name__)
def scalar_trim(val: Any) -> str:
"""ORM/csv pueden devolver tipos no-str; el matching debe ser estable."""
if val is None:
return ""
return str(val).strip()
def normalize_license_for_match(s: Any) -> str:
"""
Patente aduanal: 4 dígitos. Alinea CSV vs BD (ej. 428 → 0428; espacios).
Si no es numérico o está vacío, retorna el valor strip (compatibilidad).
"""
t = scalar_trim(s)
if not t:
return ""
if t.isdigit() and len(t) <= 4:
return t.zfill(4)
return t
def normalize_pedimento_number_str(s: Any) -> str:
"""
Número de pedimento a 7 dígitos para comparar CSV vs BD.
- CSV (parse_pedimento_col_a) ya entrega segmentos numéricos ≤7 dígitos.
- BD puede traer espacios, basura no numérica, o datos legacy con más de 7 dígitos
(p. ej. clave larga pegada en la columna); se extraen dígitos y, si hay más de 7,
se usan los últimos 7 (el número oficial que coincide con lo capturado en facturas).
"""
t = scalar_trim(s)
if not t:
return ""
digits = "".join(ch for ch in t if ch.isdigit())
if not digits:
return ""
if len(digits) <= 7:
return digits.zfill(7)
return digits[-7:]
def _row_number_looks_like_csv_number(row_number: Any, csv_num_norm: str) -> bool:
"""
Heurística defensiva para catálogos legacy:
además del match canónico por últimos 7 dígitos, considera match
si los dígitos crudos contienen el número CSV normalizado.
"""
if not csv_num_norm:
return False
row_norm = normalize_pedimento_number_str(row_number)
if row_norm == csv_num_norm:
return True
raw_digits = "".join(ch for ch in scalar_trim(row_number) if ch.isdigit())
return bool(raw_digits and csv_num_norm in raw_digits)
def normalize_catalog_pedimento_identity(
customs_office: Any,
license_val: Any,
pedimento_number: Any,
) -> Optional[Tuple[str, str, str]]:
"""
Normaliza aduana, patente y número tal como los usa resolve_pedimento_candidates
al cargar filas del catálogo en memoria.
La patente puede venir vacía en BD (alta incompleta); se usa "" y el resolver
no exige coincidencia de patente CSV vs catálogo en ese caso.
El número de pedimento es obligatorio (sin número no hay fila válida).
"""
co = scalar_trim(customs_office)
lic_raw = scalar_trim(license_val)
num_raw = scalar_trim(pedimento_number)
if not num_raw:
return None
if lic_raw:
lic = normalize_license_for_match(lic_raw) or lic_raw
else:
lic = ""
num = normalize_pedimento_number_str(num_raw) or num_raw
return co, lic, num
def canonical_customs_office_for_pedimento_key(customs_office: Any) -> str:
"""
Aduana canónica de 2 dígitos para claves y matching (007 vs 07, 640 vs 64).
Tres dígitos con ceros a la izquierda tipo 007 → últimos dos (07); otros tres dígitos → primeros dos (64).
"""
co = scalar_trim(customs_office)
if not co:
return ""
if len(co) == 3 and co.isdigit():
if co[0] == "0":
return co[1:3]
return co[:2]
if len(co) == 1 and co.isdigit():
return co.zfill(2)
return co[:2] if len(co) >= 2 else co
def customs_offices_equivalent(a: str, b: str) -> bool:
"""True si dos códigos de aduana representan la misma sección para efectos de pedimento."""
ca = canonical_customs_office_for_pedimento_key(a)
cb = canonical_customs_office_for_pedimento_key(b)
if ca and cb:
return ca == cb
return _same_customs_office(a or "", b or "")
def _co_variants(customs_office: str) -> Set[str]:
co = (customs_office or "").strip()
co = scalar_trim(customs_office)
if not co:
return set()
out: Set[str] = {co}
@@ -24,6 +134,23 @@ def _same_customs_office(csv_customs: str, candidate_customs: str) -> bool:
return bool(csv_variants.intersection(cand_variants))
def aduana_cruce_matches_catalog(ab: str, valid_codes: Set[str]) -> bool:
"""
Valida Col.AB contra public.customs_sections: mismo criterio que pedimento/aduana
(p. ej. CSV `070` vs catálogo `07`).
Si no hay catálogo cargado o la celda viene vacía, no rechaza.
"""
if not scalar_trim(ab) or not valid_codes:
return True
a = scalar_trim(ab)
if a in valid_codes:
return True
for vc in valid_codes:
if _same_customs_office(a, scalar_trim(vc)):
return True
return False
def resolve_pedimento_candidates(
customs_office: str,
license_val: str,
@@ -39,38 +166,113 @@ def resolve_pedimento_candidates(
seen_ids: Set[Any] = set()
lookup_licenses: List[str] = []
lic = (license_val or "").strip()
lic = scalar_trim(license_val)
if lic:
lookup_licenses.append(lic)
patente = (patente_lookup or "").strip()
patente = scalar_trim(patente_lookup)
if patente and patente not in lookup_licenses:
lookup_licenses.append(patente)
lookup_license_raw: Set[str] = set()
lookup_license_norm: Set[str] = set()
for x in lookup_licenses:
if not x:
continue
rx = x.strip()
lookup_license_raw.add(rx)
nx = normalize_license_for_match(rx)
if nx:
lookup_license_norm.add(nx)
csv_num_norm = normalize_pedimento_number_str(pedimento_number)
logger.debug(f"Resolving pedimento: customs={customs_office}, license={license_val}, num={pedimento_number} (norm={csv_num_norm}), lookup_licenses={lookup_license_norm}")
total_catalog_rows = len(pedimento_rows or [])
num_matches = 0
for info in (pedimento_rows or []):
row_num = (info.get("pedimento_number") or "").strip()
row_lic = (info.get("license") or "").strip()
if row_num != (pedimento_number or "").strip():
continue
if lookup_licenses and row_lic not in lookup_licenses:
row_num_raw = scalar_trim(info.get("pedimento_number"))
if not _row_number_looks_like_csv_number(row_num_raw, csv_num_norm):
continue
num_matches += 1
row_lic = scalar_trim(info.get("license"))
if lookup_license_raw or lookup_license_norm:
# Alta sin patente en catálogo: no excluir por patente del CSV (dato incompleto).
if row_lic:
row_ln = normalize_license_for_match(row_lic)
if row_lic not in lookup_license_raw and row_ln not in lookup_license_norm:
continue
pid = info.get("id")
if pid not in seen_ids:
candidates.append(info)
seen_ids.add(pid)
if not candidates:
return {"status": "not_found", "pedimento": None, "candidates": []}
# Diagnostic: find why it failed
num_only_matches = []
for info in (pedimento_rows or []):
rn_raw = scalar_trim(info.get("pedimento_number"))
if _row_number_looks_like_csv_number(rn_raw, csv_num_norm):
num_only_matches.append(info)
# DEBUG: one line per CSV row on large jobs would flood logs; enable when diagnosing catalog mismatches.
logger.debug(
"Pedimento catalog search results: found 0 candidates for %s-%s-%s "
"(csv_num_norm: %s) among %s rows. Matches by number only: %s",
customs_office,
license_val,
pedimento_number,
csv_num_norm,
total_catalog_rows,
[{"co": c.get("customs_office"), "lic": c.get("license"), "id": c.get("id")} for c in num_only_matches],
)
if num_only_matches:
return {
"status": "not_found",
"pedimento": None,
"candidates": [],
"not_found_reason": "license_or_patente_mismatch",
}
return {
"status": "not_found",
"pedimento": None,
"candidates": [],
"not_found_reason": "no_catalog_match",
}
customs_filtered = [
c for c in candidates
if _same_customs_office(customs_office, (c.get("customs_office") or ""))
if _same_customs_office(customs_office, scalar_trim(c.get("customs_office")))
]
if len(customs_filtered) == 1:
return {"status": "ok", "pedimento": customs_filtered[0], "candidates": customs_filtered}
res = customs_filtered[0]
logger.debug(
"Pedimento found: ID %s for %s-%s-%s",
res.get("id"),
customs_office,
license_val,
pedimento_number,
)
return {"status": "ok", "pedimento": res, "candidates": customs_filtered}
if len(customs_filtered) > 1:
logger.warning(f"Ambiguous pedimento (multiple customs match): {len(customs_filtered)} candidates for {customs_office}-{license_val}-{pedimento_number}")
return {"status": "ambiguous", "pedimento": None, "candidates": customs_filtered}
if len(candidates) == 1:
return {"status": "ok", "pedimento": candidates[0], "candidates": candidates}
solo = candidates[0]
cand_co = scalar_trim(solo.get("customs_office"))
if not cand_co:
return {"status": "ok", "pedimento": solo, "candidates": candidates}
logger.warning(f"Pedimento customs mismatch: expected {customs_office}, found {cand_co}")
return {
"status": "not_found",
"pedimento": None,
"candidates": [],
"not_found_reason": "customs_mismatch",
}
logger.warning(f"Ambiguous pedimento (no customs match but multiple candidates): {len(candidates)} candidates")
return {"status": "ambiguous", "pedimento": None, "candidates": candidates}

11
backend/api/v1/modules/a76/pedmientos/services/pedimentos.py
View File

@@ -276,6 +276,8 @@ class PedimentosService:
Returns:
Created pedimento
"""
if company_id is None:
raise ValueError("company_id es obligatorio para crear pedimentos.")
try:
# Check for existing pedimento with same key (Year, Aduana, Patente, Number)
# This avoids IntegrityError in many cases and provides a better error message.
@@ -483,6 +485,15 @@ class PedimentosService:
# Ensure company_id is set from the existing record
company_id = pedimento.company_id
if company_id is None:
logger.error(
"Pedimento %s without company_id detected during update tenant_id=%s",
pedimento_id,
tenant_id,
)
raise ValueError(
"El pedimento no tiene company_id asignado. Corrige el dato antes de actualizar."
)
try:
# Actualizar campos principales del pedimento