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

feature/carga-csv-db-expo-data-to-platform

Browse Source
This commit is contained in:
hreyes
2026-03-20 15:04:28 -06:00
parent 9e30dede1c
commit 9cfc8b3215
7 changed files with 463 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.tmp_commit_4f95.json Normal file
View File

File diff suppressed because one or more lines are too long

1
.tmp_commit_latest.json Normal file
View File

File diff suppressed because one or more lines are too long

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

@@ -9,7 +9,7 @@ campos no presentes en CSV en null. No se modifican plantillas CSV; no se invent
datos sin fuente (p. ej. LineReference solo si hay fuente explícita).
"""
import os
from datetime import datetime
from datetime import datetime, timedelta
from decimal import Decimal
import csv
import json
@@ -58,13 +58,13 @@ class ForeignKeyValidator:
self.session = session
self.tenant_id = tenant_id
self.company_id = company_id
self.cache = {} # {(model_name, value): bool}
self.cache = {} # {(model_name, field_name, is_public, value): bool}
def check_exists(self, model, value, field_name="id", is_public=False):
if value is None:
return True # Assume optional if None, or let DB handle not-null
key = (model.__name__, value)
key = (model.__name__, field_name, is_public, value)
if key in self.cache:
return self.cache[key]
@@ -198,6 +198,7 @@ def _validate_customs_broker_ref(
line_num: int,
col_name: str,
required: bool = False,
broker_lookup: Optional[Dict[str, Any]] = None,
) -> Optional[Dict[str, Any]]:
"""Valida AGENTE ADUANAL: acepta ID (entero) o clave broker_key (texto)."""
if raw_value is None or not str(raw_value).strip():
@@ -209,10 +210,24 @@ def _validate_customs_broker_ref(
"solution": f"Capturar el dato requerido en la columna {col_name}.",
}
return None
if broker_lookup is not None:
resolved_id, _mode, _normalized, _err = resolve_customs_broker_ref(raw_value, broker_lookup)
if resolved_id is None:
return {
"line": line_num,
"col": col_name,
"msg": "No existe en el catalogo",
"solution": f"Capturar en la columna {col_name} una clave que exista en el catálogo.",
}
return None
pid = parse_int(raw_value)
if pid is not None:
return validate_tenant_fk_id(validator, model, pid, line_num, col_name, required=False)
clave = str(raw_value).strip()
# Si viene numérico primero intentamos como ID.
# Si no existe ese ID, intentamos como broker_key (clave numérica).
err = validate_tenant_fk_id(validator, model, pid, line_num, col_name, required=False)
if err is None:
return None
clave = _normalize_broker_key(raw_value)
if not validator.check_exists(model, clave, field_name="broker_key"):
return {
"line": line_num,
@@ -222,6 +237,73 @@ def _validate_customs_broker_ref(
}
return None
def _normalize_broker_key(value: Any) -> str:
return str(value or "").strip().upper()
def _build_customs_broker_lookup(rows: List[Tuple[int, Optional[str]]]) -> Dict[str, Any]:
"""
Construye lookup robusto para AGENTE ADUANAL:
- id_set: IDs válidos
- key_to_id: broker_key normalizado -> id
- alias_to_id: alias P#### <-> #### solo cuando es inequívoco
"""
id_set: Set[int] = set()
key_to_id: Dict[str, int] = {}
alias_votes: Dict[str, Set[int]] = {}
for bid, bkey in rows:
id_set.add(int(bid))
norm = _normalize_broker_key(bkey)
if not norm:
continue
key_to_id[norm] = int(bid)
if norm.startswith("P") and norm[1:].isdigit():
alias_votes.setdefault(norm[1:], set()).add(int(bid))
elif norm.isdigit():
alias_votes.setdefault(f"P{norm}", set()).add(int(bid))
alias_to_id: Dict[str, int] = {}
for alias, ids in alias_votes.items():
if len(ids) == 1:
alias_to_id[alias] = next(iter(ids))
return {
"id_set": id_set,
"key_to_id": key_to_id,
"alias_to_id": alias_to_id,
}
def resolve_customs_broker_ref(
raw_value: Any,
broker_lookup: Dict[str, Any],
) -> Tuple[Optional[int], Optional[str], str, Optional[str]]:
"""
Resuelve AGENTE ADUANAL con patrón robusto:
id -> broker_key exacto -> alias P####/####.
"""
text = str(raw_value or "").strip()
if not text:
return None, None, "", None
id_set: Set[int] = broker_lookup.get("id_set", set())
key_to_id: Dict[str, int] = broker_lookup.get("key_to_id", {})
alias_to_id: Dict[str, int] = broker_lookup.get("alias_to_id", {})
pid = parse_int(raw_value)
if pid is not None and pid in id_set:
return int(pid), "id", text, None
norm = _normalize_broker_key(raw_value)
if norm in key_to_id:
return key_to_id[norm], "broker_key", norm, None
if norm in alias_to_id:
return alias_to_id[norm], "broker_key_alias", norm, None
return None, None, norm, "No existe en el catalogo"
@celery_app.task(bind=True)
def scan_file(self, job_id: str, model_target: str, config: str = None, job_type_override: Optional[str] = None):
"""Pass 1: Read CSV, Validate types, Write Errors to JSONL. Delegates to _do_scan_file."""
@@ -667,7 +749,11 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
)
processed_rows += 1
common_storage.store_error_lines(effective_job_type, job_id, error_lines_list)
# El commit usa líneas únicas para omitir filas; el resumen preliminar
# debe usar la misma base para evitar discrepancias de válidos/errores.
unique_error_lines = sorted(set(error_lines_list))
error_count = len(unique_error_lines)
common_storage.store_error_lines(effective_job_type, job_id, unique_error_lines)
return common_responses.scan_result(
job_id, processed_rows, error_count, errors_detail, total_rows_in_file=total_rows
)
@@ -991,7 +1077,11 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
)
processed_rows += 1
common_storage.store_error_lines(effective_job_type, job_id, error_lines_list)
# El commit usa líneas únicas para omitir filas; el resumen preliminar
# debe usar la misma base para evitar discrepancias de válidos/errores.
unique_error_lines = sorted(set(error_lines_list))
error_count = len(unique_error_lines)
common_storage.store_error_lines(effective_job_type, job_id, unique_error_lines)
return common_responses.scan_result(
job_id, processed_rows, error_count, errors_detail, total_rows_in_file=total_rows
)
@@ -2197,7 +2287,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
invoice_exists_by_number[n] = True
invoice_processed_by_number[n] = bool(is_upd)
pedimento_data_by_key: Dict[str, List[Dict[str, Any]]] = {}
pedimento_rows: List[Dict[str, Any]] = []
for p in (
session.query(
Pedimentos.id,
@@ -2218,9 +2308,8 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
co = (p.customs_office or "").strip()
lic = (p.license or "").strip()
num = (p.pedimento_number or "").strip()
if not co or not lic or not num:
if not lic or not num:
continue
key = _pedimento_key_from_parsed(co, lic, num)
entry_date = None
end_date = None
pd = (
@@ -2232,15 +2321,16 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
end_date = pd[1]
info = {
"id": p.id,
"customs_office": co,
"license": lic,
"pedimento_number": num,
"regime": (p.regime or "").strip(),
"operation_type": (p.operation_type or "").strip(),
"pedimento_type": (p.pedimento_type or "").strip(),
"entry_date": entry_date,
"end_date": end_date,
}
if key not in pedimento_data_by_key:
pedimento_data_by_key[key] = []
pedimento_data_by_key[key].append(info)
pedimento_rows.append(info)
remesa_por_pedimento_bd: Dict[str, Set[int]] = {}
q_rem = (
@@ -2289,15 +2379,15 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
valid_sold_to_short_names.add(sn_upper)
valid_shipped_to_short_names.add(sn_upper)
valid_broker_ids: Set[int] = set()
valid_broker_claves: Set[str] = set()
for cb in session.query(CustomsBroker.id, CustomsBroker.broker_key).filter(
broker_rows = session.query(CustomsBroker.id, CustomsBroker.broker_key).filter(
CustomsBroker.tenant_id == tenant_id,
CustomsBroker.company_id == company_id,
).all():
valid_broker_ids.add(cb[0])
if cb[1] and str(cb[1]).strip():
valid_broker_claves.add(str(cb[1]).strip())
).all()
broker_lookup_scan = _build_customs_broker_lookup(broker_rows)
valid_broker_ids: Set[int] = set(broker_lookup_scan["id_set"])
valid_broker_claves: Set[str] = set(broker_lookup_scan["key_to_id"].keys()) | set(
broker_lookup_scan["alias_to_id"].keys()
)
valid_transporter_keys: Set[str] = set()
for t in session.query(Transporter.transporter_key).filter(
@@ -2415,7 +2505,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
actualizar=actualizar,
invoice_exists_by_number=invoice_exists_by_number,
invoice_processed_by_number=invoice_processed_by_number,
pedimento_data_by_key=pedimento_data_by_key,
pedimento_rows=pedimento_rows,
remesa_por_pedimento_bd=remesa_por_pedimento_bd,
remesa_por_pedimento_csv=remesa_por_pedimento_csv,
valid_provider_ids=valid_provider_ids,
@@ -2470,7 +2560,11 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
)
processed_rows += 1
common_storage.store_error_lines(effective_job_type, job_id, error_lines_list)
# El commit usa líneas únicas para omitir filas; el resumen preliminar
# debe usar la misma base para evitar discrepancias de válidos/errores.
unique_error_lines = sorted(set(error_lines_list))
error_count = len(unique_error_lines)
common_storage.store_error_lines(effective_job_type, job_id, unique_error_lines)
return common_responses.scan_result(
job_id, processed_rows, error_count, errors_detail, total_rows_in_file=total_rows
)
@@ -2546,7 +2640,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
invoice_exists_by_number[n] = True
invoice_processed_by_number[n] = bool(is_upd)
pedimento_data_by_key = {}
pedimento_rows = []
for p in (
session.query(
Pedimentos.id,
@@ -2569,9 +2663,8 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
co = (p.customs_office or "").strip()
lic = (p.license or "").strip()
num = (p.pedimento_number or "").strip()
if not co or not lic or not num:
if not lic or not num:
continue
key = _pedimento_key_from_parsed(co, lic, num)
entry_date = None
end_date = None
pd = (
@@ -2583,15 +2676,16 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
end_date = pd[1]
info = {
"id": p.id,
"customs_office": co,
"license": lic,
"pedimento_number": num,
"regime": (p.regime or "").strip(),
"operation_type": (p.operation_type or "").strip(),
"pedimento_type": (p.pedimento_type or "").strip(),
"entry_date": entry_date,
"end_date": end_date,
}
if key not in pedimento_data_by_key:
pedimento_data_by_key[key] = []
pedimento_data_by_key[key].append(info)
pedimento_rows.append(info)
remesa_por_pedimento_bd = {}
q_rem = (
@@ -2641,15 +2735,15 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
valid_sold_to_short_names.add(sn_upper)
valid_shipped_to_short_names.add(sn_upper)
valid_broker_ids = set()
valid_broker_claves = set()
for cb in session.query(CustomsBroker.id, CustomsBroker.broker_key).filter(
broker_rows = session.query(CustomsBroker.id, CustomsBroker.broker_key).filter(
CustomsBroker.tenant_id == tenant_id,
CustomsBroker.company_id == company_id,
).all():
valid_broker_ids.add(cb[0])
if cb[1] and str(cb[1]).strip():
valid_broker_claves.add(str(cb[1]).strip())
).all()
broker_lookup_scan = _build_customs_broker_lookup(broker_rows)
valid_broker_ids = set(broker_lookup_scan["id_set"])
valid_broker_claves = set(broker_lookup_scan["key_to_id"].keys()) | set(
broker_lookup_scan["alias_to_id"].keys()
)
valid_transporter_keys = set()
for t in session.query(Transporter.transporter_key).filter(
@@ -2767,7 +2861,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
actualizar=actualizar,
invoice_exists_by_number=invoice_exists_by_number,
invoice_processed_by_number=invoice_processed_by_number,
pedimento_data_by_key=pedimento_data_by_key,
pedimento_rows=pedimento_rows,
remesa_por_pedimento_bd=remesa_por_pedimento_bd,
remesa_por_pedimento_csv=remesa_por_pedimento_csv,
valid_provider_ids=valid_provider_ids,
@@ -2910,7 +3004,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
ped_filter_op = "exp"
ped_filter_regimes = ["EXD", "ETE", "ETR"]
pedimento_data_by_key = {}
pedimento_rows = []
for p in (
session.query(
Pedimentos.id,
@@ -2933,9 +3027,8 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
co = (p.customs_office or "").strip()
lic = (p.license or "").strip()
num = (p.pedimento_number or "").strip()
if not co or not lic or not num:
if not lic or not num:
continue
key = _pedimento_key_from_parsed(co, lic, num)
entry_date = None
end_date = None
pd = (
@@ -2947,6 +3040,9 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
end_date = pd[1]
info = {
"id": p.id,
"customs_office": co,
"license": lic,
"pedimento_number": num,
"regime": (p.regime or "").strip(),
"operation_type": (p.operation_type or "").strip().upper()[:3],
"pedimento_type": (p.pedimento_type or "").strip(),
@@ -2954,9 +3050,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
"entry_date": entry_date,
"end_date": end_date,
}
if key not in pedimento_data_by_key:
pedimento_data_by_key[key] = []
pedimento_data_by_key[key].append(info)
pedimento_rows.append(info)
remesa_por_pedimento_bd = {}
q_rem = (
@@ -3006,15 +3100,15 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
valid_sold_to_short_names.add(sn_upper)
valid_shipped_to_short_names.add(sn_upper)
valid_broker_ids = set()
valid_broker_claves = set()
for cb in session.query(CustomsBroker.id, CustomsBroker.broker_key).filter(
broker_rows = session.query(CustomsBroker.id, CustomsBroker.broker_key).filter(
CustomsBroker.tenant_id == tenant_id,
CustomsBroker.company_id == company_id,
).all():
valid_broker_ids.add(cb[0])
if cb[1] and str(cb[1]).strip():
valid_broker_claves.add(str(cb[1]).strip())
).all()
broker_lookup_scan = _build_customs_broker_lookup(broker_rows)
valid_broker_ids = set(broker_lookup_scan["id_set"])
valid_broker_claves = set(broker_lookup_scan["key_to_id"].keys()) | set(
broker_lookup_scan["alias_to_id"].keys()
)
valid_transporter_keys = set()
for t in session.query(Transporter.transporter_key).filter(
@@ -3143,7 +3237,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
invoice_exists_by_number=invoice_exists_by_number,
invoice_processed_by_number=invoice_processed_by_number,
invoice_in_report_by_number=invoice_in_report_by_number,
pedimento_data_by_key=pedimento_data_by_key,
pedimento_rows=pedimento_rows,
remesa_por_pedimento_bd=remesa_por_pedimento_bd,
remesa_por_pedimento_csv=remesa_por_pedimento_csv,
valid_provider_ids=valid_provider_ids,
@@ -3849,6 +3943,22 @@ def display_date_format(date_format: Optional[str]) -> str:
def parse_date(date_text: Optional[str], date_format: Optional[str]) -> Optional[datetime.date]:
if not date_text:
return None
raw = str(date_text).strip()
# Soporte para fechas numéricas (layouts legacy):
# - Clarion: días desde 1800-12-28 (ej. 75519)
# - Excel serial: días desde 1899-12-30
if re.fullmatch(r"\d+(\.\d+)?", raw):
try:
serial = int(float(raw))
if serial > 0:
clarion_date = datetime(1800, 12, 28).date() + timedelta(days=serial)
if 1900 <= clarion_date.year <= 2200:
return clarion_date
excel_date = datetime(1899, 12, 30).date() + timedelta(days=serial)
if 1900 <= excel_date.year <= 2200:
return excel_date
except Exception:
pass
candidates = []
fmt_map = {
"dd/mm/yyyy": "%d/%m/%Y",
@@ -3860,7 +3970,7 @@ def parse_date(date_text: Optional[str], date_format: Optional[str]) -> Optional
candidates.extend(["%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y"])
for fmt in candidates:
try:
return datetime.strptime(str(date_text).strip(), fmt).date()
return datetime.strptime(raw, fmt).date()
except ValueError:
continue
return None
@@ -4029,14 +4139,22 @@ def resolve_customs_broker_id(
tenant_id: int,
company_id: int,
cache: Dict[Any, Optional[int]],
broker_lookup: Optional[Dict[str, Any]] = None,
) -> Optional[int]:
"""Resuelve ID de CustomsBroker por id (entero) o por broker_key (clave). value puede ser int o str."""
if value is None or (isinstance(value, str) and not value.strip()):
return None
if broker_lookup is not None:
resolved_id, _mode, norm, _err = resolve_customs_broker_ref(value, broker_lookup)
cache[value] = resolved_id
if norm:
cache[norm] = resolved_id
return resolved_id
if value in cache:
return cache[value]
pid = parse_int(value)
if pid is not None:
# Primero resolvemos como ID; si no existe, seguimos como broker_key.
found = (
session.query(model.id)
.filter(
@@ -4046,8 +4164,9 @@ def resolve_customs_broker_id(
)
.scalar()
)
cache[value] = found
return found
if found is not None:
cache[value] = found
return found
clave = str(value).strip()
if clave in cache:
return cache[clave]
@@ -4893,8 +5012,10 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
parse_pedimento_col_a,
_pedimento_key_from_parsed,
row_to_transport_type_clarion,
_patente_from_agente_aduanal,
)
from .validators.encabezados_impo_def import parse_pedimento_col_a_impo_def
from .validators.pedimento_resolution import resolve_pedimento_candidates
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.items.line_financials.models import LineFinancial
@@ -4987,9 +5108,19 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
customs_section_cache: Dict[str, Optional[str]] = {}
part_cache: Dict[str, Optional[int]] = {}
pedimento_id_cache: Dict[str, Optional[int]] = {}
pedimento_rows_insert: List[Dict[str, Any]] = []
broker_lookup: Optional[Dict[str, Any]] = None
shipped_by_cache: Dict[Any, Optional[int]] = {}
_fc_insert = parse_footer_config(meta.get("footer_config"))
autonumerar_remesas_insert = _fc_insert.get("autonumerar_remesas", False)
actualizar_insert = bool(meta.get("actualizar", False))
if isinstance(_fc_insert, dict):
if "actualizar" in _fc_insert:
actualizar_insert = bool(_fc_insert.get("actualizar"))
elif _fc_insert.get("mode") == "update":
actualizar_insert = True
elif _fc_insert.get("mode") == "replace":
actualizar_insert = False
class_id_by_code: Dict[str, int] = {}
class_uom_by_code: Dict[str, Optional[str]] = {}
class_fraction_by_code: Dict[str, Optional[str]] = {}
@@ -5022,6 +5153,42 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
validator = ForeignKeyValidator(session, tenant_id, company_id)
if model_target == "invoice_header":
broker_rows = session.query(CustomsBroker.id, CustomsBroker.broker_key).filter(
CustomsBroker.tenant_id == tenant_id,
CustomsBroker.company_id == company_id,
).all()
broker_lookup = _build_customs_broker_lookup(broker_rows)
for p in (
session.query(
Pedimentos.id,
Pedimentos.customs_office,
Pedimentos.license,
Pedimentos.pedimento_number,
Pedimentos.operation_type,
Pedimentos.regime,
)
.filter(
Pedimentos.tenant_id == tenant_id,
Pedimentos.company_id == company_id,
)
.all()
):
co = (p.customs_office or "").strip()
lic = (p.license or "").strip()
num = (p.pedimento_number or "").strip()
if not lic or not num:
continue
info = {
"id": p.id,
"customs_office": co,
"license": lic,
"pedimento_number": num,
"operation_type": (p.operation_type or "").strip(),
"regime": (p.regime or "").strip(),
}
pedimento_rows_insert.append(info)
error_msg_by_line: Dict[int, str] = {}
if error_path and os.path.exists(error_path):
try:
@@ -5054,6 +5221,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
for i, row in enumerate(reader, start=1):
row_norm = row_from_template(row, template_id, normalize_header)
customs_broker_id_resolved: Optional[int] = None
if i in error_lines:
skipped_invalid += 1
if model_target == 'invoice_header':
@@ -5074,13 +5242,40 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
if model_target == 'invoice_header':
invoice_number = (row_norm.get('NUMERO FACTURA') or row_norm.get('NUM FACTURA') or row_norm.get('FACTURA') or '').strip()
invoice_date = parse_date(row_norm.get('FECHA FACTURA') or row_norm.get('FECHA'), date_format)
existing_header = None
if not invoice_number or not invoice_date:
if not invoice_number:
skipped_invalid += 1
reason = "Número de factura o fecha faltante/inválida"
reason = "Número de factura faltante/inválido"
skipped_fk_details.append({"line": i, "invoice": invoice_number or "(vacío)", "reason": reason})
logger.debug(f"Row {i}: Skipped - missing invoice_number or invalid invoice_date. "
f"Invoice: {invoice_number}, Date: {row_norm.get('FECHA FACTURA') or row_norm.get('FECHA')}")
logger.debug(f"Row {i}: Skipped - missing invoice_number. Invoice: {invoice_number}")
continue
if not invoice_date and actualizar_insert:
# En modo update, si la fecha no viene parseable en CSV,
# reutilizamos la fecha existente de la factura.
existing_header = (
session.query(InvoiceHeader)
.filter(
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
InvoiceHeader.invoice_number == invoice_number,
InvoiceHeader.invoice_type == inv_type_value,
InvoiceHeader.operation_type == op_type_value,
)
.first()
)
if existing_header and existing_header.invoice_date:
invoice_date = existing_header.invoice_date
if not invoice_date:
skipped_invalid += 1
reason = "Fecha de factura faltante/inválida"
skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
logger.debug(
f"Row {i}: Skipped - invalid invoice_date. Invoice: {invoice_number}, "
f"Date: {row_norm.get('FECHA FACTURA') or row_norm.get('FECHA')}"
)
continue
# --- NEW: Foreign Key Validations ---
@@ -5138,6 +5333,23 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
continue
if inv_type_value != "MEX":
if broker_lookup is not None:
raw_agent = row_norm.get('AGENTE ADUANAL')
resolved_id, mode, normalized_agent, resolve_err = resolve_customs_broker_ref(raw_agent, broker_lookup)
if resolve_err:
skipped_missing_fk += 1
reason = (
f"AGENTE ADUANAL: {resolve_err} "
f"(valor='{str(raw_agent or '').strip()}', normalizado='{normalized_agent}')"
)
skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
continue
customs_broker_id_resolved = resolved_id
# Sustitución interna: una vez resuelto, usar el ID real para todo el pipeline.
if customs_broker_id_resolved is not None:
row_norm['AGENTE ADUANAL'] = customs_broker_id_resolved
err = _validate_customs_broker_ref(
validator,
CustomsBroker,
@@ -5145,6 +5357,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
i,
"AGENTE ADUANAL",
required=False,
broker_lookup=broker_lookup,
)
if err:
skipped_missing_fk += 1
@@ -5248,8 +5461,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
# 2. Client/Provider and broker checks are handled above
# --- 4. Check for Existing Invoice (Upsert Logic) ---
existing_header = None
if invoice_number:
if invoice_number and existing_header is None:
existing_header = (
session.query(InvoiceHeader)
.filter(
@@ -5273,33 +5485,32 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
parsed = parse_pedimento_col_a(ped_str)
if parsed:
customs_office_p, license_p, num_p = (x.strip() if x else "" for x in parsed)
key_p = _pedimento_key_from_parsed(customs_office_p, license_p, num_p)
if key_p not in pedimento_id_cache:
co_prefix = (customs_office_p or "").strip()[:2].zfill(2)
ped_query = (
session.query(Pedimentos.id)
.filter(
Pedimentos.tenant_id == tenant_id,
Pedimentos.company_id == company_id,
Pedimentos.customs_office.startswith(co_prefix),
Pedimentos.license == license_p,
Pedimentos.pedimento_number == num_p,
)
patente_lookup = _patente_from_agente_aduanal(row_norm, license_p)
cache_key_p = f"{template_id}|{_es_cambio_regimen}|{customs_office_p}|{license_p}|{num_p}|{patente_lookup}"
if cache_key_p not in pedimento_id_cache:
ped_resolved = resolve_pedimento_candidates(
customs_office=customs_office_p,
license_val=license_p,
pedimento_number=num_p,
patente_lookup=patente_lookup,
pedimento_rows=pedimento_rows_insert,
)
if template_id == "exp_def_header":
if _es_cambio_regimen == "S":
ped_query = ped_query.filter(
func.lower(Pedimentos.operation_type) == "imp",
func.upper(Pedimentos.regime) == "IMD",
)
resolved_id: Optional[int] = None
if ped_resolved["status"] == "ok" and ped_resolved["pedimento"]:
ped_info = ped_resolved["pedimento"]
op_type = str(ped_info.get("operation_type") or "").strip().lower()
regimen = str(ped_info.get("regime") or "").strip().upper()
if template_id == "exp_def_header":
if _es_cambio_regimen == "S":
if op_type == "imp" and regimen == "IMD":
resolved_id = ped_info.get("id")
else:
if op_type == "exp" and regimen in {"EXD", "ETE", "ETR"}:
resolved_id = ped_info.get("id")
else:
ped_query = ped_query.filter(
func.lower(Pedimentos.operation_type) == "exp",
func.upper(Pedimentos.regime).in_(["EXD", "ETE", "ETR"]),
)
ped_row = ped_query.first()
pedimento_id_cache[key_p] = ped_row[0] if ped_row else None
pedimento_id = pedimento_id_cache[key_p]
resolved_id = ped_info.get("id")
pedimento_id_cache[cache_key_p] = resolved_id
pedimento_id = pedimento_id_cache[cache_key_p]
if pedimento_id is not None and remesa_val is None and autonumerar_remesas_insert:
max_rem = (
session.query(func.max(InvoiceComplianceMx.remesa))
@@ -5421,13 +5632,16 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
),
customs_broker_id=(
None if inv_type_value == "MEX" else
resolve_customs_broker_id(
customs_broker_id_resolved
if customs_broker_id_resolved is not None
else resolve_customs_broker_id(
session,
CustomsBroker,
row_norm.get('AGENTE ADUANAL'),
tenant_id,
company_id,
broker_cache,
broker_lookup=broker_lookup,
)
),
edocument=(row_norm.get('E DOCUMENT') or None),

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

@@ -6,6 +6,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 .encabezados_impo_temp import (
_clip,
@@ -13,6 +14,7 @@ from .encabezados_impo_temp import (
_get,
_parse_int,
_pedimento_key_from_parsed,
_patente_from_agente_aduanal,
_validaciones_catalogos,
_validaciones_factura_longitud,
_validaciones_moneda,
@@ -99,7 +101,7 @@ def _validaciones_pedimento_remesa_expo(
row: Dict[str, Any],
line_num: int,
cambio_regimen: bool,
pedimento_data_by_key: Dict[str, List[Dict[str, Any]]],
pedimento_rows: List[Dict[str, Any]],
remesa_por_pedimento_bd: Dict[str, Set[int]],
remesa_por_pedimento_csv: Dict[str, Dict[int, str]],
autonumerar_remesas: bool,
@@ -137,8 +139,15 @@ def _validaciones_pedimento_remesa_expo(
customs_office, license_val, pedimento_number = parsed
key = _pedimento_key_from_parsed(customs_office, license_val, pedimento_number)
ped_info_list = pedimento_data_by_key.get(key)
if not ped_info_list:
patente_lookup = _patente_from_agente_aduanal(row, license_val)
ped_resolved = resolve_pedimento_candidates(
customs_office=customs_office,
license_val=license_val,
pedimento_number=pedimento_number,
patente_lookup=patente_lookup,
pedimento_rows=pedimento_rows,
)
if ped_resolved["status"] == "not_found":
return _err(
line_num,
"PEDIMENTO",
@@ -149,8 +158,15 @@ def _validaciones_pedimento_remesa_expo(
else "Darlo de alta como pedimento de Exportación."
),
)
if 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.",
)
ped_info = ped_info_list[0]
ped_info = ped_resolved["pedimento"]
op_type = (ped_info.get("operation_type") or "").strip().upper()
regimen_ped = (ped_info.get("regime") or "").strip().upper()
pedimento_code = (ped_info.get("pedimento_code") or "").strip().upper()
@@ -319,7 +335,7 @@ def validate_row_encabezados_expo(
valid_aduana_codes: Set[str],
valid_currency_codes: Set[str],
invoice_in_report_by_number: Optional[Dict[str, bool]] = None,
pedimento_data_by_key: Optional[Dict[str, List[Dict[str, Any]]]] = None,
pedimento_rows: Optional[List[Dict[str, Any]]] = None,
valid_provider_short_names: Optional[Set[str]] = None,
valid_sold_to_short_names: Optional[Set[str]] = None,
valid_shipped_to_short_names: Optional[Set[str]] = None,
@@ -398,7 +414,7 @@ def validate_row_encabezados_expo(
row,
line_num,
cambio_regimen,
pedimento_data_by_key or {},
pedimento_rows or [],
remesa_por_pedimento_bd,
remesa_por_pedimento_csv,
autonumerar_remesas,

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

@@ -6,6 +6,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
# Reutilizar del TEM: helpers y validaciones de catálogos (claves/short names), transporte, moneda, tipo peso, tipo cambio
from .encabezados_impo_temp import (
@@ -14,6 +15,7 @@ from .encabezados_impo_temp import (
_get,
_parse_int,
_pedimento_key_from_parsed,
_patente_from_agente_aduanal,
_validaciones_catalogos,
_validaciones_factura_longitud,
_validaciones_moneda,
@@ -92,7 +94,7 @@ def _validaciones_regimen_imd(row: Dict[str, Any], line_num: int) -> Optional[Di
def _validaciones_pedimento_remesa_def(
row: Dict[str, Any],
line_num: int,
pedimento_data_by_key: Dict[str, List[Dict[str, Any]]],
pedimento_rows: List[Dict[str, Any]],
remesa_por_pedimento_bd: Dict[str, Set[int]],
remesa_por_pedimento_csv: Dict[str, Dict[int, str]],
autonumerar_remesas: bool,
@@ -132,16 +134,30 @@ def _validaciones_pedimento_remesa_def(
customs_office, license_val, pedimento_number = parsed
key = _pedimento_key_from_parsed(customs_office, license_val, pedimento_number)
ped_info_list = pedimento_data_by_key.get(key)
if not ped_info_list:
patente_lookup = _patente_from_agente_aduanal(row, license_val)
ped_resolved = resolve_pedimento_candidates(
customs_office=customs_office,
license_val=license_val,
pedimento_number=pedimento_number,
patente_lookup=patente_lookup,
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. "
"Darlo de alta como pedimento de Importación Definitiva.",
)
if 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.",
)
ped_info = ped_info_list[0]
ped_info = ped_resolved["pedimento"]
regimen_ped = (ped_info.get("regime") or "").strip().upper()
if regimen_ped != REGIMEN_IMD:
return _err(
@@ -226,7 +242,7 @@ def validate_row_encabezados_impo_def(
actualizar: bool,
invoice_exists_by_number: Dict[str, bool],
invoice_processed_by_number: Dict[str, bool],
pedimento_data_by_key: Dict[str, List[Dict[str, Any]]],
pedimento_rows: List[Dict[str, Any]],
remesa_por_pedimento_bd: Dict[str, Set[int]],
remesa_por_pedimento_csv: Dict[str, Dict[int, str]],
valid_provider_ids: Set[int],
@@ -307,7 +323,7 @@ def validate_row_encabezados_impo_def(
_validaciones_pedimento_remesa_def(
row,
line_num,
pedimento_data_by_key,
pedimento_rows,
remesa_por_pedimento_bd,
remesa_por_pedimento_csv,
autonumerar_remesas,

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

@@ -7,6 +7,7 @@ Estructura CSV: PEDIMENTO (A), REMESA (B), NUMERO FACTURA (C), ... ADUANA DE CRU
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
# 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)
@@ -103,11 +104,10 @@ def _pedimento_key_from_parsed(customs_office: str, license_val: str, pedimento_
# El CSV de encabezados normalmente usa 2 dígitos, por lo que normalizamos:
# - 2 dígitos: se usan tal cual
# - 3 dígitos:
# - si inicia con 0, usar los últimos 2 para no "perder" el 0 relevante (ej. 007 -> 07)
# - si no inicia con 0, usar los primeros 2
# - usar siempre los primeros 2 (ej. 007 -> 00, 640 -> 64)
# - 1 dígito: left-pad a 2
if len(co) == 3:
co = co[1:3] if co.startswith("0") else co[:2]
co = co[:2]
elif len(co) == 1:
co = co.zfill(2)
else:
@@ -116,6 +116,21 @@ def _pedimento_key_from_parsed(customs_office: str, license_val: str, pedimento_
return f"{co}-{license_val}-{pedimento_number}"
def _patente_from_agente_aduanal(row: Dict[str, Any], default_license: str) -> str:
"""
Obtiene patente para lookup desde AGENTE ADUANAL (columna específica).
Si no es numérica o viene vacía, usa la patente parseada del PEDIMENTO.
"""
agente = _get(row, "AGENTE ADUANAL")
if agente and agente.isdigit():
if len(agente) < 4:
return agente.zfill(4)
if len(agente) > 4:
return agente[-4:]
return agente
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}
@@ -157,7 +172,7 @@ def _validaciones_obligatorios_toda(
def _validaciones_pedimento_remesa(
row: Dict[str, Any],
line_num: int,
pedimento_data_by_key: Dict[str, List[Dict[str, Any]]],
pedimento_rows: List[Dict[str, Any]],
remesa_por_pedimento_bd: Dict[str, Set[int]],
remesa_por_pedimento_csv: Dict[str, Dict[int, str]],
autonumerar_remesas: bool,
@@ -198,16 +213,30 @@ def _validaciones_pedimento_remesa(
customs_office, license_val, pedimento_number = parsed
key = _pedimento_key_from_parsed(customs_office, license_val, pedimento_number)
ped_info_list = pedimento_data_by_key.get(key)
if not ped_info_list:
patente_lookup = _patente_from_agente_aduanal(row, license_val)
ped_resolved = resolve_pedimento_candidates(
customs_office=customs_office,
license_val=license_val,
pedimento_number=pedimento_number,
patente_lookup=patente_lookup,
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":
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.",
)
ped_info = ped_info_list[0]
ped_info = ped_resolved["pedimento"]
if (ped_info.get("operation_type") or "").upper() != "IMP":
return _err(
line_num,
@@ -439,10 +468,10 @@ def _validaciones_catalogos(
return None
v = _parse_int(val)
if v is not None:
if valid_ids and v not in valid_ids:
return _err(line_num, col, f"Error: La clave en {col} no existe en el Catálogo de {catalog_name}.")
return None
clave = str(val).strip()
# Si no existe como id, intentar también como clave (p.ej. "5567" como broker_key).
if not valid_ids or v in valid_ids:
return None
clave = str(val).strip().upper()
if valid_claves and clave not in valid_claves:
return _err(line_num, col, f"Error: La clave en {col} no existe en el Catálogo de {catalog_name}.")
if not valid_claves:
@@ -530,7 +559,7 @@ def validate_row_encabezados_impo_temp(
actualizar: bool,
invoice_exists_by_number: Dict[str, bool],
invoice_processed_by_number: Dict[str, bool],
pedimento_data_by_key: Dict[str, List[Dict[str, Any]]],
pedimento_rows: List[Dict[str, Any]],
remesa_por_pedimento_bd: Dict[str, Set[int]],
remesa_por_pedimento_csv: Dict[str, Dict[int, str]],
valid_provider_ids: Set[int],
@@ -601,7 +630,7 @@ def validate_row_encabezados_impo_temp(
_validaciones_pedimento_remesa(
row,
line_num,
pedimento_data_by_key,
pedimento_rows,
remesa_por_pedimento_bd,
remesa_por_pedimento_csv,
autonumerar_remesas,

76
backend/api/v1/modules/a76/layouts_csv/facturas/validators/pedimento_resolution.py Normal file
View File

@@ -0,0 +1,76 @@
from typing import Any, Dict, List, Optional, Set
def _co_variants(customs_office: str) -> Set[str]:
co = (customs_office or "").strip()
if not co:
return set()
out: Set[str] = {co}
if len(co) == 1 and co.isdigit():
out.add(co.zfill(2))
if len(co) >= 2:
out.add(co[:2])
out.add(co[-2:])
if len(co) >= 3:
out.add(co[:3])
return {x for x in out if x}
def _same_customs_office(csv_customs: str, candidate_customs: str) -> bool:
csv_variants = _co_variants(csv_customs)
cand_variants = _co_variants(candidate_customs)
if not csv_variants or not cand_variants:
return False
return bool(csv_variants.intersection(cand_variants))
def resolve_pedimento_candidates(
customs_office: str,
license_val: str,
pedimento_number: str,
patente_lookup: Optional[str],
pedimento_rows: List[Dict[str, Any]],
) -> Dict[str, Any]:
"""
Resuelve candidatos de pedimento por componentes reales.
Prioriza match exacto por aduana/patente/numero y usa fallback por patente/numero.
"""
candidates: List[Dict[str, Any]] = []
seen_ids: Set[Any] = set()
lookup_licenses: List[str] = []
lic = (license_val or "").strip()
if lic:
lookup_licenses.append(lic)
patente = (patente_lookup or "").strip()
if patente and patente not in lookup_licenses:
lookup_licenses.append(patente)
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:
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": []}
customs_filtered = [
c for c in candidates
if _same_customs_office(customs_office, (c.get("customs_office") or ""))
]
if len(customs_filtered) == 1:
return {"status": "ok", "pedimento": customs_filtered[0], "candidates": customs_filtered}
if len(customs_filtered) > 1:
return {"status": "ambiguous", "pedimento": None, "candidates": customs_filtered}
if len(candidates) == 1:
return {"status": "ok", "pedimento": candidates[0], "candidates": candidates}
return {"status": "ambiguous", "pedimento": None, "candidates": candidates}