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

feature/feature/clarion-validaciones-invoices-csv-headers-def-impo

Browse Source
This commit is contained in:
hreyes
2026-03-09 11:29:39 -06:00
parent 9d00c504c2
commit bcec74a741
5 changed files with 747 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -937,6 +937,341 @@ def scan_file(self, job_id: str, model_target: str, config: str = None):
logger.exception("Encabezados importación temporal scan failed: %s", e)
return {"status": "failed", "error": str(e)}
# --- Encabezados de Importación Definitiva: flujo específico (Clarion VALIDA_TODA_FACIMPO_DEF / VALIDA_PARCIAL) ---
if model_target == "invoice_header" and template_id == "imp_def_header":
try:
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceComplianceMx, InvoiceFinancials
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos
from api.v1.modules.a76.pedmientos.models.pedimento_dates import PedimentoDates
from api.v1.modules.a76.clients_and_providers.models import ClientProvider
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection
from api.v1.modules.public.reference_data.incoterms.models import Incoterm
from api.v1.modules.public.reference_data.currency_types.models import CurrencyType
from api.v1.modules.a76.general_catalogs.exchange_rate.models import ExchangeRate
from api.v1.modules.a76.transportation.transporters.models import Transporter
from .validators.encabezados_impo_def import (
validate_row_encabezados_impo_def,
parse_pedimento_col_a_impo_def,
)
from .validators.encabezados_impo_temp import _pedimento_key_from_parsed
DEF_INVOICE_TYPES = ("DEF", "MATDE", "EXDEF")
def _ped_key_from_row_def(ped_str: str) -> Optional[str]:
parsed = parse_pedimento_col_a_impo_def(ped_str)
if not parsed:
return None
return _pedimento_key_from_parsed(parsed[0], parsed[1], parsed[2])
_fc = parse_footer_config(meta.get("footer_config"))
actualizar = meta.get("actualizar", False)
autonumerar_remesas = meta.get("autonumerar_remesas", False)
control_remesa = bool(_fc.get("control_remesa", False))
remesa_inicio = _fc.get("remesa_inicio")
remesa_fin = _fc.get("remesa_fin")
if remesa_inicio is not None:
try:
remesa_inicio = int(remesa_inicio)
except (TypeError, ValueError):
remesa_inicio = None
if remesa_fin is not None:
try:
remesa_fin = int(remesa_fin)
except (TypeError, ValueError):
remesa_fin = None
if _fc:
if "actualizar" in _fc:
actualizar = bool(_fc["actualizar"])
if "autonumerar_remesas" in _fc:
autonumerar_remesas = bool(_fc["autonumerar_remesas"])
with CoreSessionLocal() as session:
q_inv = (
session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.is_updated)
.filter(
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
InvoiceHeader.operation_type == "imp",
InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES),
)
)
invoice_exists_by_number = {}
invoice_updated_by_number = {}
for num, iid, is_upd in q_inv.all():
if num:
n = str(num).strip()
invoice_exists_by_number[n] = True
invoice_updated_by_number[n] = bool(is_upd)
pedimento_data_by_key = {}
for p in (
session.query(
Pedimentos.id,
Pedimentos.year,
Pedimentos.customs_office,
Pedimentos.license,
Pedimentos.pedimento_number,
Pedimentos.operation_type,
Pedimentos.regime,
Pedimentos.pedimento_type,
)
.filter(
Pedimentos.tenant_id == tenant_id,
Pedimentos.company_id == company_id,
Pedimentos.operation_type == "imp",
Pedimentos.regime == "IMD",
)
.all()
):
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:
continue
key = _pedimento_key_from_parsed(co, lic, num)
entry_date = None
end_date = None
pd = (
session.query(PedimentoDates.entry_date, PedimentoDates.end_date)
.filter(PedimentoDates.pedimento_id == p.id).first()
)
if pd:
entry_date = pd[0]
end_date = pd[1]
info = {
"id": p.id,
"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)
remesa_por_pedimento_bd = {}
q_rem = (
session.query(
InvoiceComplianceMx.remesa,
Pedimentos.customs_office,
Pedimentos.license,
Pedimentos.pedimento_number,
)
.join(InvoiceHeader, InvoiceHeader.id == InvoiceComplianceMx.invoice_id)
.join(Pedimentos, Pedimentos.id == InvoiceComplianceMx.pedimento_id)
.filter(
InvoiceComplianceMx.tenant_id == tenant_id,
InvoiceComplianceMx.company_id == company_id,
InvoiceComplianceMx.pedimento_id.isnot(None),
InvoiceComplianceMx.remesa.isnot(None),
InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES),
)
)
for rem, co, lic, num in q_rem.all():
if co and lic and num and rem is not None:
key = _pedimento_key_from_parsed(
(co or "").strip(),
(lic or "").strip(),
(num or "").strip(),
)
if key not in remesa_por_pedimento_bd:
remesa_por_pedimento_bd[key] = set()
remesa_por_pedimento_bd[key].add(int(rem))
valid_provider_ids = set()
valid_sold_to_ids = set()
valid_shipped_to_ids = set()
valid_provider_short_names = set()
valid_sold_to_short_names = set()
valid_shipped_to_short_names = set()
for cp in session.query(ClientProvider.id, ClientProvider.short_name).filter(
ClientProvider.tenant_id == tenant_id,
ClientProvider.company_id == company_id,
).all():
valid_provider_ids.add(cp[0])
valid_sold_to_ids.add(cp[0])
valid_shipped_to_ids.add(cp[0])
if cp[1] and str(cp[1]).strip():
sn_upper = str(cp[1]).strip().upper()
valid_provider_short_names.add(sn_upper)
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(
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())
valid_transporter_keys = set()
for t in session.query(Transporter.transporter_key).filter(
Transporter.tenant_id == tenant_id,
Transporter.company_id == company_id,
).all():
if t[0]:
valid_transporter_keys.add((t[0] or "").strip().upper())
valid_incoterms = set()
for inc in session.query(Incoterm.code).all():
if inc[0]:
valid_incoterms.add((inc[0] or "").strip().upper())
valid_aduana_codes = set()
for cs in session.query(CustomsSection.customs_code).all():
if cs[0]:
valid_aduana_codes.add((cs[0] or "").strip())
valid_currency_codes = set()
for ct in session.query(CurrencyType.code).all():
if ct[0]:
valid_currency_codes.add((ct[0] or "").strip().upper())
exchange_rate_by_date = {}
for er in session.query(ExchangeRate.date, ExchangeRate.value).filter(
ExchangeRate.tenant_id == tenant_id,
ExchangeRate.company_id == company_id,
).all():
if er[0] and er[1] is not None:
dk = er[0].strftime("%Y-%m-%d") if hasattr(er[0], "strftime") else str(er[0])[:10]
exchange_rate_by_date[dk] = er[1]
invoice_has_partidas_by_number = {}
existing_tipo_moneda_by_number = {}
q_li_count = (
session.query(InvoiceHeader.invoice_number, func.count(LineItem.id))
.join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
.filter(
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
InvoiceHeader.operation_type == "imp",
InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES),
)
.group_by(InvoiceHeader.invoice_number)
)
for num, cnt in q_li_count.all():
if num:
invoice_has_partidas_by_number[str(num).strip()] = cnt > 0
q_fin = (
session.query(InvoiceHeader.invoice_number, InvoiceFinancials.currency)
.join(InvoiceFinancials, InvoiceFinancials.invoice_id == InvoiceHeader.id)
.filter(
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
InvoiceHeader.operation_type == "imp",
InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES),
)
)
for num, cur in q_fin.all():
if num and cur:
cur_str = (cur or "").strip().lower()
if cur_str == "foreign":
existing_tipo_moneda_by_number[str(num).strip()] = "ME"
elif cur_str == "local":
existing_tipo_moneda_by_number[str(num).strip()] = "MN"
else:
existing_tipo_moneda_by_number[str(num).strip()] = (cur_str or "").upper()[:2]
with open(file_path, "r", encoding="utf-8-sig") as f_in:
sample = f_in.read(2048)
f_in.seek(0)
try:
dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
except Exception:
dialect = "excel"
reader = csv.DictReader(f_in, dialect=dialect)
rows_list = list(reader)
remesa_por_pedimento_csv = {}
for row in rows_list:
row_norm = row_from_template(row, "imp_def_header", normalize_header)
ped = (row_norm.get("PEDIMENTO") or "").strip()
rem = row_norm.get("REMESA")
factura = (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or row_norm.get("FACTURA") or "").strip()
if not ped or not factura:
continue
key = _ped_key_from_row_def(ped)
if not key:
continue
try:
rem_int = int(rem) if rem is not None and str(rem).strip() else None
except (TypeError, ValueError):
rem_int = None
if rem_int is not None:
if key not in remesa_por_pedimento_csv:
remesa_por_pedimento_csv[key] = {}
if rem_int not in remesa_por_pedimento_csv[key]:
remesa_por_pedimento_csv[key][rem_int] = factura
error_count = 0
processed_rows = 0
error_lines_list = []
errors_detail = []
with open(error_path, "w", encoding="utf-8") as f_err:
for i, row in enumerate(rows_list, start=1):
if i % 1000 == 0:
self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows, "errors": error_count})
row_norm = row_from_template(row, "imp_def_header", normalize_header)
warnings_row = []
err = validate_row_encabezados_impo_def(
row_norm,
i,
actualizar=actualizar,
invoice_exists_by_number=invoice_exists_by_number,
invoice_updated_by_number=invoice_updated_by_number,
pedimento_data_by_key=pedimento_data_by_key,
remesa_por_pedimento_bd=remesa_por_pedimento_bd,
remesa_por_pedimento_csv=remesa_por_pedimento_csv,
valid_provider_ids=valid_provider_ids,
valid_sold_to_ids=valid_sold_to_ids,
valid_shipped_to_ids=valid_shipped_to_ids,
valid_provider_short_names=valid_provider_short_names,
valid_sold_to_short_names=valid_sold_to_short_names,
valid_shipped_to_short_names=valid_shipped_to_short_names,
valid_broker_ids=valid_broker_ids,
valid_broker_claves=valid_broker_claves,
valid_transporter_keys=valid_transporter_keys,
valid_incoterms=valid_incoterms,
valid_aduana_codes=valid_aduana_codes,
valid_currency_codes=valid_currency_codes,
exchange_rate_by_date=exchange_rate_by_date,
invoice_has_partidas_by_number=invoice_has_partidas_by_number,
existing_tipo_moneda_by_number=existing_tipo_moneda_by_number,
autonumerar_remesas=autonumerar_remesas,
control_remesa=control_remesa,
remesa_inicio=remesa_inicio,
remesa_fin=remesa_fin,
date_format=date_format,
parse_date_fn=parse_date,
warnings=warnings_row,
)
if err:
error_count += 1
error_lines_list.append(err["line"])
f_err.write(json.dumps({"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")}) + "\n")
if len(errors_detail) < 500:
errors_detail.append({"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")})
for w in warnings_row:
if len(errors_detail) < 500:
errors_detail.append({"line": w["line"], "col": w.get("col", ""), "msg": w.get("msg", ""), "warning": True})
processed_rows += 1
if error_lines_list:
common_storage.store_error_lines(JOB_TYPE, job_id, error_lines_list)
return common_responses.scan_result(
job_id, processed_rows, error_count, errors_detail, total_rows_in_file=total_rows
)
except Exception as e:
logger.exception("Encabezados importación definitiva scan failed: %s", e)
return {"status": "failed", "error": str(e)}
try:
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.clients_and_providers.models import ClientProvider
@@ -1857,6 +2192,11 @@ def insert_valid_rows(self, job_id: str, model_target: str):
# Default types from config or fallback
op_type_value = OperationType(meta.get('operation_type', 'imp').lower())
inv_type_value = normalize_public_code(footer_config.get('invoice_type') or 'TEM') or 'TEM'
_template_id_insert = meta.get("template_id") or (
"imp_temp_header" if model_target == "invoice_header" else "imp_temp_details"
)
if model_target == "invoice_header" and _template_id_insert == "imp_def_header":
inv_type_value = "DEF"
logger.info(f"Processing CSV with operation_type={op_type_value}, invoice_type={inv_type_value}, date_format={date_format}")
@@ -2103,12 +2443,13 @@ def insert_valid_rows(self, job_id: str, model_target: str):
customs_office_p, license_p, num_p = 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]
ped_row = (
session.query(Pedimentos.id)
.filter(
Pedimentos.tenant_id == tenant_id,
Pedimentos.company_id == company_id,
Pedimentos.customs_office == customs_office_p,
Pedimentos.customs_office.startswith(co_prefix),
Pedimentos.license == license_p,
Pedimentos.pedimento_number == num_p,
)

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

@@ -49,8 +49,39 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
{"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
# --- Encabezado factura: Impo Def (EstructuraEncFacImpoDef.xls) - 30 columnas Clarion ---
"imp_def_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": "TIPO DE CAMBIO"},
{"canonical": "REGIMEN", "aliases": ["CLAVEDOCUMENTO"]},
{"canonical": "CLAVE PROVEEDOR"},
{"canonical": "CLAVE VENDIDO A", "aliases": ["CLAVE VENDIDO A:"]},
{"canonical": "CLAVE ENVIADO A"},
{"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": "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"},
],
# --- Encabezado factura: Expo (EstructuraEncFacExpoCamReg.xls) - misma estructura ---
"exp_def_header": None,
# --- Partidas factura: Impo Temp (EstructuraParFacImpoTemp - paridad Clarion A-AG) ---

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

@@ -5,10 +5,16 @@ from .encabezados_impo_temp import (
parse_pedimento_col_a,
_pedimento_key_from_parsed,
)
from .encabezados_impo_def import (
validate_row_encabezados_impo_def,
parse_pedimento_col_a_impo_def,
)
__all__ = [
"validate_row_encabezados_impo_temp",
"validate_row_encabezados_impo_def",
"row_to_transport_type_clarion",
"parse_pedimento_col_a",
"parse_pedimento_col_a_impo_def",
"_pedimento_key_from_parsed",
]

363
backend/api/v1/modules/a76/layouts_csv/facturas/validators/encabezados_impo_def.py Normal file
View File

@@ -0,0 +1,363 @@
"""
Validaciones CSV para Encabezados de Facturas de Importación Definitiva.
Paridad Clarion: VALIDA_TODA_FACIMPO_DEF, VALIDA_PARCIAL_FACIMPO_DEF, VALIDACIONES_FACIMPO_DEF.
Reutiliza de encabezados_impo_temp: claves, short names y validaciones de catálogos, transporte,
moneda, tipo peso y tipo cambio (misma lógica que TEM).
"""
from datetime import datetime
from typing import Any, Dict, List, Optional, Set, Tuple
# Reutilizar del TEM: helpers y validaciones de catálogos (claves/short names), transporte, moneda, tipo peso, tipo cambio
from .encabezados_impo_temp import (
_clip,
_err,
_get,
_parse_int,
_pedimento_key_from_parsed,
_validaciones_catalogos,
_validaciones_factura_longitud,
_validaciones_moneda,
_validaciones_tipo_cambio,
_validaciones_tipo_peso,
_validaciones_transporte,
)
# Impo Def: régimen único y pedimento formato ##-####-####### (15 chars)
REGIMEN_IMD = "IMD"
MAX_LEN_PEDIMENTO_DEF = 15
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
"""
if not pedimento_str or not isinstance(pedimento_str, str):
return None
s = (pedimento_str or "").strip()
if len(s) != MAX_LEN_PEDIMENTO_DEF:
return None
if s[2:3] != "-" or s[7:8] != "-":
return None
part0, part1, part2 = s[0:2], s[3:7], s[8:15]
if not part0.isdigit() or not part1.isdigit() or not part2.isdigit():
return None
return (part0, part1, part2)
def _validaciones_obligatorios_toda_def(
row: Dict[str, Any],
line_num: int,
tiene_pedimento: bool,
) -> Optional[Dict[str, Any]]:
"""Obligatorios VALIDA_TODA para Impo Def: C, D, F, G, H, I, J. Aduana de Cruce (AB) no obligatoria."""
obligatorios: List[str] = []
if not _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA"):
obligatorios.append("(Col.C) 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 obligatorios:
return _err(
line_num,
"ARCHIVO CSV",
f"Existen campos vacíos que son obligatorios: {', '.join(obligatorios)}. Revisar para Importación Definitiva.",
)
return None
def _validaciones_regimen_imd(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Col F: solo IMD válido para Impo Def."""
f = _get(row, "REGIMEN", "CLAVEDOCUMENTO")
if f and f.upper() != REGIMEN_IMD:
return _err(
line_num,
"REGIMEN",
f"Error: (Celda F{line_num}) El Régimen Aduanero: {f} no es válido para este tipo de movimiento. Los válidos son: IMD.",
)
return None
def _validaciones_pedimento_remesa_def(
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],
invoice_date_parsed: Optional[datetime],
) -> Optional[Dict[str, Any]]:
"""Pedimento Col A: max 15 chars, formato ##-####-#######; catálogo con regime IMD. Remesa igual que TEM."""
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_DEF:
return _err(
line_num,
"PEDIMENTO",
f"Error: (Celda A{line_num}) El Pedimento: {col_a} supera la longitud de caracteres. Use formato ##-####-#######.",
)
parsed = parse_pedimento_col_a_impo_def(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 ##-####-#######.",
)
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. "
"Darlo de alta como pedimento de Importación Definitiva.",
)
ped_info = ped_info_list[0]
regimen_ped = (ped_info.get("regime") or "").strip().upper()
if regimen_ped != REGIMEN_IMD:
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.",
)
if col_f and col_f != REGIMEN_IMD:
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.",
)
if col_f and col_f != regimen_ped:
return _err(
line_num,
"REGIMEN",
f"Error: (Celda F{line_num}) El Régimen Aduanero: {col_f} capturado es diferente al del Pedimento: {regimen_ped}. Debe ser IMD.",
)
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
def _normalize_tipo_transporte_ferro(row: Dict[str, Any]) -> Dict[str, Any]:
"""Clarion DEF acepta 'FERRO BARCAZA' (con espacio). Normaliza a FERROBARCAZA para reutilizar validación TEM."""
out = dict(row)
m = out.get("TIPO TRANSPORTE")
if m is not None and str(m).strip().upper().replace(" ", "") == "FERROBARCAZA":
out["TIPO TRANSPORTE"] = "FERROBARCAZA"
return out
def validate_row_encabezados_impo_def(
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]]:
"""
Valida una fila de CSV de Encabezados de Importación Definitiva.
Clarion: VALIDA_TODA_FACIMPO_DEF vs VALIDA_PARCIAL_FACIMPO_DEF.
Reutiliza de encabezados_impo_temp las validaciones de catálogos (claves y short names),
transporte, moneda, tipo peso y tipo cambio.
"""
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 a las facturas actualizadas.",
)
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 Definitiva 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_def(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_def(
row,
line_num,
pedimento_data_by_key,
remesa_por_pedimento_bd,
remesa_por_pedimento_csv,
autonumerar_remesas,
control_remesa,
remesa_inicio,
remesa_fin,
invoice_date_parsed,
)
if err:
return err
err = _validaciones_factura_longitud(row, line_num)
if err:
return err
err = _validaciones_regimen_imd(row, line_num)
if err:
return err
# Normalizar FERRO BARCAZA -> FERROBARCAZA para reutilizar validación TEM
row_transport = _normalize_tipo_transporte_ferro(row)
err = _validaciones_transporte(row_transport, 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
# Mismas claves y short names que TEM (Proveedor, Vendido A, Enviado A, Agente, Transportista, Incoterm, Aduana)
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

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

@@ -96,8 +96,9 @@ def parse_pedimento_col_a(pedimento_str: str) -> Optional[Tuple[str, str, str]]:
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}"
"""Clave para lookup: CC-LLLL-NNNNNNN (solo primeros 2 dígitos de aduana)."""
co = (customs_office or "").strip()[:2]
return f"{co}-{license_val}-{pedimento_number}"
def _err(line_num: int, col: str, msg: str) -> Dict[str, Any]: