feature/clarion-invoices-temp-headers-csv
This commit is contained in:
hreyes
File diff suppressed because it is too large
Load Diff
@@ -12,41 +12,42 @@ from typing import Dict, List, Any, Optional
|
||||
# aliases = cabeceras alternativas que la plantilla .xls puede traer (ej. "Num Factura" → NUM FACTURA).
|
||||
|
||||
TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
|
||||
# --- Encabezado factura: Impo Temp (EstructuraEncFacImpoTemp.xls) ---
|
||||
# --- Encabezado factura: Impo Temp (EstructuraEncFacImpoTemp.xls) - Clarion A-AD ---
|
||||
"imp_temp_header": [
|
||||
{"canonical": "PEDIMENTO", "aliases": ["NUMERO PEDIMENTO", "PEDIMENTO NUMERO", "NUMERO DE PEDIMENTO", "PED"]},
|
||||
{"canonical": "REMESA"},
|
||||
{"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA", "ID"]},
|
||||
{"canonical": "FECHA FACTURA", "aliases": ["FECHA"]},
|
||||
{"canonical": "FECHA EMISION"},
|
||||
{"canonical": "TIPO DE CAMBIO"},
|
||||
{"canonical": "REGIMEN", "aliases": ["CLAVEDOCUMENTO"]},
|
||||
{"canonical": "CLAVE PROVEEDOR"},
|
||||
{"canonical": "CLAVE VENDIDO A"},
|
||||
{"canonical": "CLAVE ENVIADO A"},
|
||||
{"canonical": "REGIMEN", "aliases": ["CLAVEDOCUMENTO"]},
|
||||
{"canonical": "ADUANA DE CRUCE"},
|
||||
{"canonical": "CLAVE MONEDA"},
|
||||
{"canonical": "CLAVE INCOTERM"},
|
||||
{"canonical": "TIPO MONEDA"},
|
||||
{"canonical": "TIPO DE CAMBIO"},
|
||||
{"canonical": "TIPO PESO"},
|
||||
{"canonical": "TIPO TRANSPORTE"},
|
||||
{"canonical": "REMESA"},
|
||||
{"canonical": "AGENTE ADUANAL"},
|
||||
{"canonical": "CLAVE TRANSPORTISTA"},
|
||||
{"canonical": "NOMBRE CONDUCTOR"},
|
||||
{"canonical": "TIPO TRANSPORTE"},
|
||||
{"canonical": "NUMERO TRANSPORTE"},
|
||||
{"canonical": "TIPO MONEDA"},
|
||||
{"canonical": "CLAVE MONEDA"},
|
||||
{"canonical": "FLETES"},
|
||||
{"canonical": "VALOR SEGUROS"},
|
||||
{"canonical": "SEGUROS"},
|
||||
{"canonical": "EMBALAJES"},
|
||||
{"canonical": "OTROS INCREMENTABLES"},
|
||||
{"canonical": "NUM PROYECTO", "aliases": ["NUMPROYECTO"]},
|
||||
{"canonical": "ORDEN COMPRA", "aliases": ["ORDENCOMPRA"]},
|
||||
{"canonical": "FACTURA ALTERNA"},
|
||||
{"canonical": "FACTURA EXPO REF", "aliases": ["FACTURAEXPOREF"]},
|
||||
{"canonical": "CLAVE INCOTERM"},
|
||||
{"canonical": "PRECINTO"},
|
||||
{"canonical": "FECHA EMISION"},
|
||||
{"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": "OBSERVACIONES E"},
|
||||
{"canonical": "OBSERVACIONES I"},
|
||||
{"canonical": "E DOCUMENT"},
|
||||
{"canonical": "NUM OPERACION"},
|
||||
{"canonical": "CLAVE TRANSPORTISTA"},
|
||||
{"canonical": "NOMBRE CONDUCTOR"},
|
||||
{"canonical": "NUMERO TRANSPORTE"},
|
||||
{"canonical": "PRECINTO"},
|
||||
{"canonical": "FACTURA ALTERNA"},
|
||||
{"canonical": "NUM PROYECTO", "aliases": ["NUMPROYECTO"]},
|
||||
{"canonical": "ORDEN COMPRA", "aliases": ["ORDENCOMPRA"]},
|
||||
{"canonical": "FACTURA EXPO REF", "aliases": ["FACTURAEXPOREF"]},
|
||||
],
|
||||
# --- Encabezado factura: Impo Def (EstructuraEncFacImpoDef.xls) - misma estructura ---
|
||||
"imp_def_header": None, # se resuelve igual que imp_temp_header
|
||||
|
||||
@@ -1 +1,14 @@
|
||||
# Validators for invoice CSV imports (header, details, series).
|
||||
from .encabezados_impo_temp import (
|
||||
validate_row_encabezados_impo_temp,
|
||||
row_to_transport_type_clarion,
|
||||
parse_pedimento_col_a,
|
||||
_pedimento_key_from_parsed,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"validate_row_encabezados_impo_temp",
|
||||
"row_to_transport_type_clarion",
|
||||
"parse_pedimento_col_a",
|
||||
"_pedimento_key_from_parsed",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,650 @@
|
||||
"""
|
||||
Validaciones CSV para Encabezados de Facturas de Importación Temporal.
|
||||
Paridad Clarion: VALIDA_TODA_FACIMPO_TEM, VALIDA_PARCIAL_FACIMPO_TEM, VALIDACIONES_FACIMPO_TEM.
|
||||
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.
|
||||
"""
|
||||
from datetime import datetime
|
||||
from decimal import Decimal, InvalidOperation
|
||||
from typing import Any, Dict, List, Optional, Set, Tuple
|
||||
|
||||
# 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)
|
||||
MAX_LEN_FACTURA = 15
|
||||
|
||||
# Formato PEDIMENTO: CC-LLLL-NNNNNNN (sin año: aduana 2-3, patente 4, número 7)
|
||||
|
||||
REGIMENES_VALIDOS = frozenset({"ITE", "ITR"})
|
||||
TIPOS_MONEDA_VALIDOS = frozenset({"ME", "MN", "MC"})
|
||||
TIPO_PESO_VALIDOS = frozenset({"KILOS", "LIBRAS"})
|
||||
|
||||
# Clarion Col M → valor normalizado (minúscula para TransportType enum)
|
||||
TIPO_TRANSPORTE_CLARION_TO_NORM = {
|
||||
"NINGUNO": "none",
|
||||
"TRANSPORTE": "transport",
|
||||
"CAJA": "box",
|
||||
"PLACAS": "licence plates",
|
||||
"CAMION": "truck",
|
||||
"BUQUE": "vessel",
|
||||
"FERROBARCAZA": "rail_barge",
|
||||
"CONTENEDOR": "container",
|
||||
"PLATAFORMA": "flatbed",
|
||||
"GONDOLA": "gondola",
|
||||
"AVION": "airplane",
|
||||
}
|
||||
TIPO_TRANSPORTE_VALIDOS = frozenset(TIPO_TRANSPORTE_CLARION_TO_NORM.keys())
|
||||
|
||||
|
||||
def _clip(val: Any) -> str:
|
||||
if val is None:
|
||||
return ""
|
||||
return str(val).strip()
|
||||
|
||||
|
||||
def _get(row: Dict[str, Any], *keys: str) -> str:
|
||||
for k in keys:
|
||||
v = row.get(k)
|
||||
if v is not None and str(v).strip():
|
||||
return _clip(v)
|
||||
return ""
|
||||
|
||||
|
||||
def _parse_decimal(val: Any) -> Optional[Decimal]:
|
||||
if val is None:
|
||||
return None
|
||||
s = _clip(val)
|
||||
if not s:
|
||||
return None
|
||||
s = s.replace(",", "")
|
||||
try:
|
||||
return Decimal(s)
|
||||
except (InvalidOperation, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def _parse_int(val: Any) -> Optional[int]:
|
||||
if val is None:
|
||||
return None
|
||||
s = _clip(val)
|
||||
if not s:
|
||||
return None
|
||||
try:
|
||||
return int(s)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def parse_pedimento_col_a(pedimento_str: str) -> Optional[Tuple[str, str, str]]:
|
||||
"""
|
||||
Parsea Col A (PEDIMENTO) formato CC-LLLL-NNNNNNN o CCC-LLLL-NNNNNNN (18 caracteres, sin año).
|
||||
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):
|
||||
return None
|
||||
s = (pedimento_str or "").strip()
|
||||
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():
|
||||
return None
|
||||
if len(license_val) != 4 or not license_val.isdigit():
|
||||
return None
|
||||
if len(pedimento_number) != 7 or not pedimento_number.isdigit():
|
||||
return None
|
||||
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 (sin año; como viene en CSV)."""
|
||||
return f"{customs_office}-{license_val}-{pedimento_number}"
|
||||
|
||||
|
||||
def _err(line_num: int, col: str, msg: str) -> Dict[str, Any]:
|
||||
return {"line": line_num, "col": col, "msg": msg}
|
||||
|
||||
|
||||
# --- Obligatorios VALIDA_TODA (cuando no es actualizar) ---
|
||||
def _validaciones_obligatorios_toda(
|
||||
row: Dict[str, Any],
|
||||
line_num: int,
|
||||
tiene_pedimento: bool,
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""Obligatorios: C, D, F, G, H, I, J; AB obligatoria si A tiene valor."""
|
||||
obligatorios: List[str] = []
|
||||
if not _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA"):
|
||||
obligatorios.append("(Col.C) Número de Factura")
|
||||
if not _get(row, "FECHA FACTURA", "FECHA"):
|
||||
obligatorios.append("(Col.D) Fecha de la Factura")
|
||||
if not _get(row, "REGIMEN", "CLAVEDOCUMENTO"):
|
||||
obligatorios.append("(Col.F) Clave Régimen")
|
||||
if not _get(row, "CLAVE PROVEEDOR"):
|
||||
obligatorios.append("(Col.G) Clave del Proveedor")
|
||||
if not _get(row, "CLAVE VENDIDO A"):
|
||||
obligatorios.append("(Col.H) Clave del Vendido A")
|
||||
if not _get(row, "CLAVE ENVIADO A"):
|
||||
obligatorios.append("(Col.I) Clave del Enviado A")
|
||||
if not _get(row, "AGENTE ADUANAL"):
|
||||
obligatorios.append("(Col.J) Clave del Agente Aduanal")
|
||||
if tiene_pedimento and not _get(row, "ADUANA DE CRUCE"):
|
||||
obligatorios.append("(Col.AB) Aduana de Cruce")
|
||||
if obligatorios:
|
||||
return _err(
|
||||
line_num,
|
||||
"ARCHIVO CSV",
|
||||
f"Existen campos vacíos que son obligatorios: {', '.join(obligatorios)}.",
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
# --- Pedimento y Remesa (Col A, B) ---
|
||||
def _validaciones_pedimento_remesa(
|
||||
row: Dict[str, Any],
|
||||
line_num: int,
|
||||
pedimento_data_by_key: Dict[str, List[Dict[str, Any]]],
|
||||
remesa_por_pedimento_bd: Dict[str, Set[int]],
|
||||
remesa_por_pedimento_csv: Dict[str, Dict[int, str]],
|
||||
autonumerar_remesas: bool,
|
||||
control_remesa: bool,
|
||||
remesa_inicio: Optional[int],
|
||||
remesa_fin: Optional[int],
|
||||
date_format: Optional[str],
|
||||
invoice_date_parsed: Optional[datetime],
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""Col A: formato, longitud, catálogo, tipo I, régimen ITE/ITR, match Col F, fechas. Col B: oblig si A, no 0, rango, unicidad."""
|
||||
col_a = _get(row, "PEDIMENTO")
|
||||
col_b_raw = row.get("REMESA")
|
||||
col_b = _clip(col_b_raw)
|
||||
col_f = _get(row, "REGIMEN", "CLAVEDOCUMENTO").upper()
|
||||
|
||||
if not col_a:
|
||||
if col_b:
|
||||
return _err(
|
||||
line_num,
|
||||
"REMESA",
|
||||
f"Error: (Celda B{line_num}) Está asignado el número de Remesa y no se tiene un pedimento en (Celda A{line_num}).",
|
||||
)
|
||||
return None
|
||||
|
||||
if len(col_a) > MAX_LEN_PEDIMENTO:
|
||||
return _err(
|
||||
line_num,
|
||||
"PEDIMENTO",
|
||||
f"Error: (Celda A{line_num}) El Pedimento: {col_a} supera la longitud de caracteres.",
|
||||
)
|
||||
parsed = parse_pedimento_col_a(col_a)
|
||||
if not parsed:
|
||||
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).",
|
||||
)
|
||||
|
||||
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:
|
||||
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.",
|
||||
)
|
||||
|
||||
ped_info = ped_info_list[0]
|
||||
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:
|
||||
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}.",
|
||||
)
|
||||
|
||||
if not autonumerar_remesas and not col_b:
|
||||
return _err(
|
||||
line_num,
|
||||
"REMESA",
|
||||
f"Error: (Celda B{line_num}) El Número de Remesa está vacío y se tiene un Pedimento en la Celda A{line_num}.",
|
||||
)
|
||||
remesa_int = _parse_int(col_b_raw)
|
||||
if col_b and remesa_int is not None and remesa_int == 0:
|
||||
return _err(
|
||||
line_num,
|
||||
"REMESA",
|
||||
f"Error: (Celda B{line_num}) El Número de Remesa no puede ser 0.",
|
||||
)
|
||||
if control_remesa and remesa_int is not None and remesa_inicio is not None and remesa_fin is not None:
|
||||
if remesa_int < remesa_inicio or remesa_int > remesa_fin:
|
||||
return _err(
|
||||
line_num,
|
||||
"REMESA",
|
||||
f"Error: (Celda B{line_num}) El Número de Remesa: {col_b} está fuera del rango configurado ({remesa_inicio}-{remesa_fin}).",
|
||||
)
|
||||
|
||||
factura_actual = _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA")
|
||||
if remesa_int is not None and key in remesa_por_pedimento_csv:
|
||||
other = remesa_por_pedimento_csv[key].get(remesa_int)
|
||||
if other and other != factura_actual:
|
||||
return _err(
|
||||
line_num,
|
||||
"REMESA",
|
||||
f"Error: (Celda B{line_num}) El Número de Remesa ya está asignado a la factura {other} en este archivo CSV.",
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
# --- Factura longitud (Col C) ---
|
||||
def _validaciones_factura_longitud(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
|
||||
c = _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA")
|
||||
if c and len(c) > MAX_LEN_FACTURA:
|
||||
return _err(line_num, "NUMERO FACTURA", f"Error: (Celda C{line_num}) El Número de Factura supera la longitud de caracteres.")
|
||||
return None
|
||||
|
||||
|
||||
# --- Régimen (Col F) ---
|
||||
def _validaciones_regimen(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
|
||||
f = _get(row, "REGIMEN", "CLAVEDOCUMENTO")
|
||||
if f and f.upper() not in REGIMENES_VALIDOS:
|
||||
return _err(
|
||||
line_num,
|
||||
"REGIMEN",
|
||||
f"Error: (Celda F{line_num}) El Régimen Aduanero: {f} no es válido. Use ITE o ITR.",
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
# --- Tipo transporte y número (Col M, N) ---
|
||||
def _validaciones_transporte(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
|
||||
m = _get(row, "TIPO TRANSPORTE").upper()
|
||||
n = _get(row, "NUMERO TRANSPORTE")
|
||||
if m and m not in TIPO_TRANSPORTE_VALIDOS:
|
||||
return _err(
|
||||
line_num,
|
||||
"TIPO TRANSPORTE",
|
||||
f"Error: (Celda M{line_num}) El Tipo de Transporte: {m} no es válido. Válidos: NINGUNO, TRANSPORTE, CAJA, PLACAS, CAMION, BUQUE, FERROBARCAZA, CONTENEDOR, PLATAFORMA, GONDOLA, AVION.",
|
||||
)
|
||||
if not m and n:
|
||||
return _err(
|
||||
line_num,
|
||||
"NUMERO TRANSPORTE",
|
||||
f"Error: (Celda N{line_num}) El Tipo de Transporte está vacío y está capturado un número de transporte.",
|
||||
)
|
||||
if m == "NINGUNO" and n:
|
||||
return _err(
|
||||
line_num,
|
||||
"NUMERO TRANSPORTE",
|
||||
f"Error: (Celda N{line_num}) El Tipo de Transporte es NINGUNO y está capturado un número de transporte.",
|
||||
)
|
||||
if m and m != "NINGUNO" and not n:
|
||||
return _err(
|
||||
line_num,
|
||||
"NUMERO TRANSPORTE",
|
||||
f"Error: (Celda N{line_num}) El Tipo de Transporte es {m} y no está capturado el número de transporte.",
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
# --- Tipo moneda y clave moneda (Col O, P) ---
|
||||
def _validaciones_moneda(
|
||||
row: Dict[str, Any],
|
||||
line_num: int,
|
||||
valid_currency_codes: Set[str],
|
||||
invoice_has_partidas: Optional[bool],
|
||||
existing_tipo_moneda: Optional[str],
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
o = _get(row, "TIPO MONEDA").upper()
|
||||
p = _get(row, "CLAVE MONEDA").upper()
|
||||
if o and o not in TIPOS_MONEDA_VALIDOS:
|
||||
return _err(
|
||||
line_num,
|
||||
"TIPO MONEDA",
|
||||
f"Error: (Celda O{line_num}) La opción de Tipo Moneda: {row.get('TIPO MONEDA')} no es válida. Use ME, MN o MC.",
|
||||
)
|
||||
if o == "MC":
|
||||
if not p:
|
||||
return _err(
|
||||
line_num,
|
||||
"CLAVE MONEDA",
|
||||
f"Error: (Celda P{line_num}) La Clave de la Moneda es obligatoria cuando Tipo de Moneda es MC.",
|
||||
)
|
||||
if p and valid_currency_codes and p not in valid_currency_codes:
|
||||
return _err(
|
||||
line_num,
|
||||
"CLAVE MONEDA",
|
||||
f"Error: (Celda P{line_num}) La Clave de la Moneda: {p} no existe en el Catálogo.",
|
||||
)
|
||||
if invoice_has_partidas and existing_tipo_moneda and o and o != existing_tipo_moneda.upper():
|
||||
return _err(
|
||||
line_num,
|
||||
"TIPO MONEDA",
|
||||
f"Error: (Celda O{line_num}) No se puede cambiar el Tipo de Moneda porque la factura ya tiene partidas. Use {existing_tipo_moneda}.",
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
# --- Tipo peso (Col Y) ---
|
||||
def _validaciones_tipo_peso(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
|
||||
y = _get(row, "TIPO PESO").upper()
|
||||
if y and y not in TIPO_PESO_VALIDOS:
|
||||
return _err(
|
||||
line_num,
|
||||
"TIPO PESO",
|
||||
f"Error: (Celda Y{line_num}) La opción de Tipo de Peso: {row.get('TIPO PESO')} no es válida. Use KILOS o LIBRAS.",
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
# --- Catálogos: Proveedor, Vendido A, Enviado A, Agente, Transportista, Incoterm, Aduana ---
|
||||
def _validaciones_catalogos(
|
||||
row: Dict[str, Any],
|
||||
line_num: int,
|
||||
valid_provider_ids: Set[int],
|
||||
valid_sold_to_ids: Set[int],
|
||||
valid_shipped_to_ids: Set[int],
|
||||
valid_provider_short_names: Set[str],
|
||||
valid_sold_to_short_names: Set[str],
|
||||
valid_shipped_to_short_names: Set[str],
|
||||
valid_broker_ids: Set[int],
|
||||
valid_broker_claves: Set[str],
|
||||
valid_transporter_keys: Set[str],
|
||||
valid_incoterms: Set[str],
|
||||
valid_aduana_codes: Set[str],
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
def check_id(val: Any, col: str, catalog_name: str, valid_set: Set[int]) -> Optional[Dict[str, Any]]:
|
||||
if val is None or str(val).strip() == "":
|
||||
return None
|
||||
v = _parse_int(val)
|
||||
if v is None:
|
||||
return _err(line_num, col, f"Error: (Celda) {col} debe ser un número entero.")
|
||||
if valid_set and v not in valid_set:
|
||||
return _err(line_num, col, f"Error: La clave en {col} no existe en el Catálogo de {catalog_name}.")
|
||||
return None
|
||||
|
||||
def check_id_or_rfc(
|
||||
val: Any,
|
||||
col: str,
|
||||
catalog_name: str,
|
||||
valid_ids: Set[int],
|
||||
valid_short_names: Set[str],
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
if val is None or str(val).strip() == "":
|
||||
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
|
||||
sn_norm = str(val).strip().upper()
|
||||
if valid_short_names and sn_norm not in valid_short_names:
|
||||
return _err(line_num, col, f"Error: La clave/corta en {col} no existe en el Catálogo de {catalog_name}.")
|
||||
if not valid_short_names:
|
||||
return _err(line_num, col, f"Error: (Celda) {col} debe ser un número entero o clave corta (short name) válida.")
|
||||
return None
|
||||
|
||||
def check_id_or_clave(
|
||||
val: Any,
|
||||
col: str,
|
||||
catalog_name: str,
|
||||
valid_ids: Set[int],
|
||||
valid_claves: Set[str],
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""Para AGENTE ADUANAL: acepta id (entero) o broker_key (clave)."""
|
||||
if val is None or str(val).strip() == "":
|
||||
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()
|
||||
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:
|
||||
return _err(line_num, col, f"Error: (Celda) {col} debe ser un número entero o clave de agente aduanal válida.")
|
||||
return None
|
||||
|
||||
err = check_id_or_rfc(
|
||||
row.get("CLAVE PROVEEDOR"), "CLAVE PROVEEDOR", "Clientes/Proveedores",
|
||||
valid_provider_ids, valid_provider_short_names,
|
||||
)
|
||||
if err:
|
||||
return err
|
||||
err = check_id_or_rfc(
|
||||
row.get("CLAVE VENDIDO A"), "CLAVE VENDIDO A", "Clientes/Proveedores",
|
||||
valid_sold_to_ids, valid_sold_to_short_names,
|
||||
)
|
||||
if err:
|
||||
return err
|
||||
err = check_id_or_rfc(
|
||||
row.get("CLAVE ENVIADO A"), "CLAVE ENVIADO A", "Clientes/Proveedores",
|
||||
valid_shipped_to_ids, valid_shipped_to_short_names,
|
||||
)
|
||||
if err:
|
||||
return err
|
||||
err = check_id_or_clave(
|
||||
row.get("AGENTE ADUANAL"), "AGENTE ADUANAL", "Agentes Aduanales",
|
||||
valid_broker_ids, valid_broker_claves,
|
||||
)
|
||||
if err:
|
||||
return err
|
||||
|
||||
k = _get(row, "CLAVE TRANSPORTISTA").upper()
|
||||
if k and valid_transporter_keys and k not in valid_transporter_keys:
|
||||
return _err(line_num, "CLAVE TRANSPORTISTA", f"Error: (Celda K{line_num}) La Clave del Transportista: {k} no existe en el Catálogo.")
|
||||
|
||||
v = _get(row, "CLAVE INCOTERM").upper()
|
||||
if v and valid_incoterms and v not in valid_incoterms:
|
||||
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:
|
||||
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
|
||||
|
||||
|
||||
def _validaciones_tipo_cambio(
|
||||
row: Dict[str, Any],
|
||||
line_num: int,
|
||||
date_parsed: Optional[datetime],
|
||||
exchange_rate_by_date: Dict[str, Any],
|
||||
warnings: Optional[List[Dict[str, Any]]],
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""Si hay fecha y no hay tipo cambio, debe existir en catálogo. Si hay tipo cambio, puede advertir si difiere del catálogo."""
|
||||
d = _get(row, "FECHA FACTURA", "FECHA")
|
||||
e = row.get("TIPO DE CAMBIO")
|
||||
if not d or not date_parsed:
|
||||
return None
|
||||
date_key = date_parsed.isoformat()[:10] if hasattr(date_parsed, "isoformat") else str(date_parsed)[:10]
|
||||
catalog_tc = exchange_rate_by_date.get(date_key) if exchange_rate_by_date else None
|
||||
if not _clip(e):
|
||||
if catalog_tc is None:
|
||||
return _err(
|
||||
line_num,
|
||||
"TIPO DE CAMBIO",
|
||||
f"Error: (Celda E{line_num}) El Tipo de Cambio para la Fecha {d} no se encontró en el Catálogo.",
|
||||
)
|
||||
return None
|
||||
val_e = _parse_decimal(e)
|
||||
if val_e is not None and catalog_tc is not None:
|
||||
cat_val = catalog_tc if isinstance(catalog_tc, (Decimal, int, float)) else getattr(catalog_tc, "valor", None) or getattr(catalog_tc, "value", None)
|
||||
if cat_val is not None and abs(float(val_e) - float(cat_val)) > 0.0001 and warnings is not None:
|
||||
warnings.append({
|
||||
"line": line_num,
|
||||
"col": "TIPO DE CAMBIO",
|
||||
"msg": f"Advertencia: (Celda E{line_num}) El Tipo de Cambio capturado ({e}) difiere del Catálogo para la fecha {d}.",
|
||||
"warning": True,
|
||||
})
|
||||
return None
|
||||
|
||||
|
||||
def validate_row_encabezados_impo_temp(
|
||||
row: Dict[str, Any],
|
||||
line_num: int,
|
||||
actualizar: bool,
|
||||
invoice_exists_by_number: Dict[str, bool],
|
||||
invoice_updated_by_number: Dict[str, bool],
|
||||
pedimento_data_by_key: Dict[str, 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],
|
||||
valid_sold_to_ids: Set[int],
|
||||
valid_shipped_to_ids: Set[int],
|
||||
valid_broker_ids: Set[int],
|
||||
valid_transporter_keys: Set[str],
|
||||
valid_incoterms: Set[str],
|
||||
valid_aduana_codes: Set[str],
|
||||
valid_currency_codes: Set[str],
|
||||
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,
|
||||
valid_broker_claves: Optional[Set[str]] = None,
|
||||
exchange_rate_by_date: Optional[Dict[str, Any]] = None,
|
||||
invoice_has_partidas_by_number: Optional[Dict[str, bool]] = None,
|
||||
existing_tipo_moneda_by_number: Optional[Dict[str, str]] = None,
|
||||
autonumerar_remesas: bool = False,
|
||||
control_remesa: bool = False,
|
||||
remesa_inicio: Optional[int] = None,
|
||||
remesa_fin: Optional[int] = None,
|
||||
date_format: Optional[str] = None,
|
||||
parse_date_fn=None,
|
||||
warnings: Optional[List[Dict[str, Any]]] = None,
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Punto de entrada: valida una fila de CSV de Encabezados de Importación Temporal.
|
||||
Clarion: VALIDA_TODA_FACIMPO_TEM vs VALIDA_PARCIAL_FACIMPO_TEM según actualizar y si la factura existe.
|
||||
"""
|
||||
factura = _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA")
|
||||
if not factura:
|
||||
return _err(line_num, "NUMERO FACTURA", "Error: (Col.C) La columna de Número de Factura está vacía y no se pueden hacer las validaciones.")
|
||||
|
||||
if invoice_updated_by_number.get(factura.strip(), False):
|
||||
return _err(
|
||||
line_num,
|
||||
"NUMERO FACTURA",
|
||||
f"Error: (Celda C{line_num}) El Número de Factura: {factura} ya existe y está Actualizada, no se puede hacer cambios.",
|
||||
)
|
||||
|
||||
if actualizar and factura.strip() not in invoice_exists_by_number:
|
||||
return _err(
|
||||
line_num,
|
||||
"NUMERO FACTURA",
|
||||
f"Error: (Col.C) Factura de importación {factura} no existe (modo Actualizar).",
|
||||
)
|
||||
|
||||
use_partial = actualizar and invoice_exists_by_number.get(factura.strip(), False)
|
||||
tiene_pedimento = bool(_get(row, "PEDIMENTO"))
|
||||
|
||||
if not use_partial:
|
||||
err = _validaciones_obligatorios_toda(row, line_num, tiene_pedimento)
|
||||
if err:
|
||||
return err
|
||||
|
||||
invoice_date_parsed = None
|
||||
if parse_date_fn:
|
||||
date_str = _get(row, "FECHA FACTURA", "FECHA")
|
||||
if date_str:
|
||||
invoice_date_parsed = parse_date_fn(date_str, date_format)
|
||||
|
||||
err = _validaciones_pedimento_remesa(
|
||||
row,
|
||||
line_num,
|
||||
pedimento_data_by_key,
|
||||
remesa_por_pedimento_bd,
|
||||
remesa_por_pedimento_csv,
|
||||
autonumerar_remesas,
|
||||
control_remesa,
|
||||
remesa_inicio,
|
||||
remesa_fin,
|
||||
date_format,
|
||||
invoice_date_parsed,
|
||||
)
|
||||
if err:
|
||||
return err
|
||||
|
||||
err = _validaciones_factura_longitud(row, line_num)
|
||||
if err:
|
||||
return err
|
||||
err = _validaciones_regimen(row, line_num)
|
||||
if err:
|
||||
return err
|
||||
err = _validaciones_transporte(row, line_num)
|
||||
if err:
|
||||
return err
|
||||
|
||||
has_partidas = invoice_has_partidas_by_number.get(factura.strip(), False) if invoice_has_partidas_by_number else False
|
||||
existing_moneda = existing_tipo_moneda_by_number.get(factura.strip()) if existing_tipo_moneda_by_number else None
|
||||
err = _validaciones_moneda(
|
||||
row,
|
||||
line_num,
|
||||
valid_currency_codes or set(),
|
||||
has_partidas if use_partial else None,
|
||||
existing_moneda if use_partial else None,
|
||||
)
|
||||
if err:
|
||||
return err
|
||||
err = _validaciones_tipo_peso(row, line_num)
|
||||
if err:
|
||||
return err
|
||||
err = _validaciones_catalogos(
|
||||
row,
|
||||
line_num,
|
||||
valid_provider_ids or set(),
|
||||
valid_sold_to_ids or set(),
|
||||
valid_shipped_to_ids or set(),
|
||||
valid_provider_short_names or set(),
|
||||
valid_sold_to_short_names or set(),
|
||||
valid_shipped_to_short_names or set(),
|
||||
valid_broker_ids or set(),
|
||||
valid_broker_claves or set(),
|
||||
valid_transporter_keys or set(),
|
||||
valid_incoterms or set(),
|
||||
valid_aduana_codes or set(),
|
||||
)
|
||||
if err:
|
||||
return err
|
||||
err = _validaciones_tipo_cambio(
|
||||
row, line_num, invoice_date_parsed, exchange_rate_by_date or {}, warnings
|
||||
)
|
||||
if err:
|
||||
return err
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def row_to_transport_type_clarion(val: Optional[str]) -> Optional[str]:
|
||||
"""Mapea valor Clarion Col M a valor enum TransportType (minúscula)."""
|
||||
if not val:
|
||||
return "none"
|
||||
u = _clip(val).upper()
|
||||
return TIPO_TRANSPORTE_CLARION_TO_NORM.get(u, "none")
|
||||
Reference in New Issue
Block a user