2026-10-14 22:28:48 -05:00
|
|
|
|
"""Extracción de datos de PDF/XML y asignación de guías de remisión."""
|
|
|
|
|
|
|
|
|
|
|
|
import os
|
2026-10-14 22:42:37 -05:00
|
|
|
|
import logging
|
2026-10-14 22:28:48 -05:00
|
|
|
|
import re
|
|
|
|
|
|
import shutil
|
|
|
|
|
|
import ntpath
|
|
|
|
|
|
from datetime import datetime
|
|
|
|
|
|
|
|
|
|
|
|
import sunat_schema
|
|
|
|
|
|
import pdf_utilities
|
|
|
|
|
|
from pipeline_utils import log, ruta_carpeta
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ============================================================================
|
|
|
|
|
|
def copiar_a_red(data: dict) -> None:
|
|
|
|
|
|
"""
|
|
|
|
|
|
Copia el PDF y el XML de un comprobante ya procesado a la ruta de red
|
|
|
|
|
|
de Facturación Electrónica, organizada como::
|
|
|
|
|
|
|
|
|
|
|
|
{RUTA_RED_FACTURACION}\\{año}\\{mes}\\{día}\\{tipo}\\{RUC_EMPRESA}\\{ruc_emisor}\\
|
|
|
|
|
|
{RUC_EMPRESA}-{tipo}-{serie}-{numero}_FormatoImpreso.pdf
|
|
|
|
|
|
{RUC_EMPRESA}-{tipo}-{serie}-{numero}.xml
|
|
|
|
|
|
|
|
|
|
|
|
``RUC_EMPRESA`` es el RUC de Aberio (fijo, por ahora); ``ruc_emisor``
|
|
|
|
|
|
es el RUC del proveedor que emitió la factura (varía por comprobante).
|
|
|
|
|
|
Año/mes/día se toman de ``fecha_emision`` (mes y día sin ceros a la
|
|
|
|
|
|
izquierda). No falla el pipeline si la ruta de red no está configurada
|
|
|
|
|
|
o no está disponible: solo registra una advertencia.
|
|
|
|
|
|
|
|
|
|
|
|
Guarda las rutas resultantes en ``data["ruta_red_pdf"]`` y
|
|
|
|
|
|
``data["ruta_red_xml"]`` para que ``build_sap_draft`` las incluya en
|
|
|
|
|
|
el draft (``U_VS_ANEXO_PDF`` / ``U_VS_ANEXO_XML``).
|
|
|
|
|
|
"""
|
|
|
|
|
|
RUTA_RED_FACTURACION = os.getenv("RUTA_RED_FACTURACION", "")
|
|
|
|
|
|
RUC_EMPRESA = os.getenv("RUC_EMPRESA", "")
|
|
|
|
|
|
|
|
|
|
|
|
if not RUTA_RED_FACTURACION:
|
|
|
|
|
|
return
|
|
|
|
|
|
|
|
|
|
|
|
ruc_emisor = data.get("ruc_emisor", "")
|
|
|
|
|
|
tipo = data.get("tipo", "")
|
|
|
|
|
|
serie = data.get("serie", "")
|
|
|
|
|
|
numero = data.get("numero", "")
|
|
|
|
|
|
xml_path = data.get("ruta_xml")
|
|
|
|
|
|
pdf_path = data.get("ruta_pdf")
|
|
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
|
fecha = datetime.strptime(data.get("fecha_emision", ""), "%Y-%m-%d")
|
|
|
|
|
|
except (ValueError, TypeError):
|
2026-10-14 22:42:37 -05:00
|
|
|
|
log(f"⚠️ No se pudo copiar a red: fecha_emision inválida ({data.get('fecha_emision')})", 3, severity=logging.WARNING)
|
2026-10-14 22:28:48 -05:00
|
|
|
|
return
|
|
|
|
|
|
|
|
|
|
|
|
carpeta_destino = os.path.join(
|
|
|
|
|
|
RUTA_RED_FACTURACION,
|
|
|
|
|
|
str(fecha.year), str(fecha.month), str(fecha.day),
|
|
|
|
|
|
tipo, RUC_EMPRESA, ruc_emisor,
|
|
|
|
|
|
)
|
|
|
|
|
|
# Ruta guardada en el draft SAP (U_VS_ANEXO_PDF/XML): siempre estilo
|
|
|
|
|
|
# Windows/UNC con backslash, sin importar el separador del SO donde
|
|
|
|
|
|
# corre el script (evita mezclar "\\" y "/" al usar os.path.join en
|
|
|
|
|
|
# un host no-Windows).
|
|
|
|
|
|
carpeta_destino_unc = ntpath.join(
|
|
|
|
|
|
RUTA_RED_FACTURACION,
|
|
|
|
|
|
str(fecha.year), str(fecha.month), str(fecha.day),
|
|
|
|
|
|
tipo, RUC_EMPRESA, ruc_emisor,
|
|
|
|
|
|
)
|
|
|
|
|
|
nombre_base = f"{RUC_EMPRESA}-{tipo}-{serie}-{numero}"
|
|
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
|
os.makedirs(carpeta_destino, exist_ok=True)
|
|
|
|
|
|
|
|
|
|
|
|
if pdf_path and os.path.exists(pdf_path):
|
|
|
|
|
|
ruta_red_pdf = os.path.join(carpeta_destino, f"{nombre_base}_FormatoImpreso.pdf")
|
|
|
|
|
|
shutil.copy2(pdf_path, ruta_red_pdf)
|
|
|
|
|
|
data["ruta_red_pdf"] = ntpath.join(carpeta_destino_unc, f"{nombre_base}_FormatoImpreso.pdf")
|
|
|
|
|
|
|
|
|
|
|
|
if xml_path and os.path.exists(xml_path):
|
|
|
|
|
|
ruta_red_xml = os.path.join(carpeta_destino, f"{nombre_base}.xml")
|
|
|
|
|
|
shutil.copy2(xml_path, ruta_red_xml)
|
|
|
|
|
|
data["ruta_red_xml"] = ntpath.join(carpeta_destino_unc, f"{nombre_base}.xml")
|
|
|
|
|
|
|
|
|
|
|
|
log(f"📁 Copiado a red: {carpeta_destino}", 3)
|
|
|
|
|
|
except OSError as e:
|
2026-10-14 22:42:37 -05:00
|
|
|
|
log(f"⚠️ Error copiando a ruta de red: {e}", 3, severity=logging.WARNING)
|
2026-10-14 22:28:48 -05:00
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ============================================================================
|
|
|
|
|
|
def procesar_pdfs(email: dict, download_folder: str) -> dict:
|
|
|
|
|
|
"""
|
|
|
|
|
|
Procesa todos los PDF de los adjuntos del correo.
|
|
|
|
|
|
|
|
|
|
|
|
Por cada PDF:
|
|
|
|
|
|
1. Extrae sus datos estructurados via ``pdf_utilities.extraer_datos_pdf``.
|
|
|
|
|
|
2. Renombra el archivo al formato canónico
|
|
|
|
|
|
``{ruc}-{tipo}-{serie}-{numero}.pdf``.
|
|
|
|
|
|
3. Almacena los datos extraídos en un cache indexado por el nombre
|
|
|
|
|
|
canónico.
|
|
|
|
|
|
|
|
|
|
|
|
Los datos extraídos también se acumulan en ``email["datos_pdf"]``.
|
|
|
|
|
|
|
|
|
|
|
|
Returns:
|
|
|
|
|
|
dict — Cache ``{nombre_canonico.pdf: datos_extraídos}``.
|
|
|
|
|
|
|
|
|
|
|
|
Example:
|
|
|
|
|
|
Si el PDF es ``F002-486.pdf`` y la extracción devuelve
|
|
|
|
|
|
``ruc_emisor=20537929520, tipo_documento=FACTURA``::
|
|
|
|
|
|
|
|
|
|
|
|
cache = procesar_pdfs(email, download_folder)
|
|
|
|
|
|
# → {"20537929520-01-F002-486.pdf": {...datos...}}
|
|
|
|
|
|
# El archivo se renombra a 20537929520-01-F002-486.pdf
|
|
|
|
|
|
"""
|
|
|
|
|
|
carpeta = ruta_carpeta(download_folder)
|
|
|
|
|
|
pdf_cache: dict[str, dict] = {}
|
|
|
|
|
|
datos_pdf = []
|
|
|
|
|
|
|
|
|
|
|
|
for idx, adjunto in enumerate(email.get("adjuntos", [])):
|
|
|
|
|
|
if not adjunto.lower().endswith(".pdf"):
|
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
|
|
|
|
pdf_path = os.path.join(carpeta, adjunto)
|
|
|
|
|
|
data = pdf_utilities.extraer_datos_pdf(pdf_path)
|
|
|
|
|
|
|
|
|
|
|
|
if data and not data.get("error"):
|
|
|
|
|
|
nuevo_path = pdf_utilities.renombrar_pdf(pdf_path, data)
|
|
|
|
|
|
if nuevo_path != pdf_path:
|
|
|
|
|
|
nuevo_nombre = os.path.basename(nuevo_path)
|
|
|
|
|
|
email["adjuntos"][idx] = nuevo_nombre
|
|
|
|
|
|
data["archivo"] = nuevo_nombre
|
|
|
|
|
|
|
|
|
|
|
|
datos_pdf.append(data)
|
|
|
|
|
|
|
|
|
|
|
|
canonical = os.path.basename(nuevo_path)
|
|
|
|
|
|
pdf_cache[canonical] = data
|
|
|
|
|
|
log(f"📄 PDF procesado: {canonical} ({data.get('tipo_documento', '?')})", 3)
|
|
|
|
|
|
else:
|
2026-10-14 22:42:37 -05:00
|
|
|
|
log(f"⚠️ PDF no procesado: {adjunto} - {data.get('error', 'error desconocido')}", 2, severity=logging.WARNING)
|
2026-10-14 22:28:48 -05:00
|
|
|
|
|
|
|
|
|
|
if datos_pdf:
|
|
|
|
|
|
email["datos_pdf"] = datos_pdf
|
|
|
|
|
|
|
|
|
|
|
|
return pdf_cache
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ============================================================================
|
|
|
|
|
|
def procesar_xmls(email: dict, download_folder: str, pdf_cache: dict) -> list[dict]:
|
|
|
|
|
|
"""
|
|
|
|
|
|
Procesa todos los XML de los adjuntos del correo.
|
|
|
|
|
|
|
|
|
|
|
|
Por cada XML:
|
|
|
|
|
|
1. Extrae datos estructurados via ``sunat_schema.extraer_datos_xml``.
|
|
|
|
|
|
2. Construye el nombre canónico del PDF asociado
|
|
|
|
|
|
(``{ruc}-{tipo}-{serie}-{numero}.pdf``) y lo busca en el cache.
|
|
|
|
|
|
3. Si el XML no trae ``documentos_referencia``, los completa desde
|
|
|
|
|
|
el PDF cache (usando ``pdf_utilities.extraer_referencias``).
|
|
|
|
|
|
4. Asigna las guías de remisión (``guias_remision[]``) y
|
|
|
|
|
|
transportista (``guia_transportista``) a cada línea de detalle desde
|
|
|
|
|
|
``notes_text`` o ``documentos_referencia``.
|
|
|
|
|
|
|
|
|
|
|
|
Returns:
|
|
|
|
|
|
list[dict] — Lista de documentos extraídos listos para
|
|
|
|
|
|
``guardar_resultado``.
|
|
|
|
|
|
|
|
|
|
|
|
Example:
|
|
|
|
|
|
XML con ``DespatchDocumentReference`` ID sucio
|
|
|
|
|
|
``"GRRE: T001-127773 / GRET :EG03-11"``::
|
|
|
|
|
|
|
|
|
|
|
|
docs = procesar_xmls(email, download_folder, pdf_cache)
|
|
|
|
|
|
# → [{
|
|
|
|
|
|
# "lineas": [{
|
|
|
|
|
|
# "guias_remision": [{"tipo":"09","serie":"T001","numero":"127773"}],
|
|
|
|
|
|
# "guia_transportista": {"tipo":"31","serie":"EG03","numero":"11"}
|
|
|
|
|
|
# }]
|
|
|
|
|
|
# }]
|
|
|
|
|
|
"""
|
|
|
|
|
|
carpeta = ruta_carpeta(download_folder)
|
|
|
|
|
|
documentos = []
|
|
|
|
|
|
|
|
|
|
|
|
last_idx = len(email["adjuntos"]) - 1
|
|
|
|
|
|
for idx, a in enumerate(email["adjuntos"]):
|
|
|
|
|
|
log(a, 2, last=(idx == last_idx))
|
|
|
|
|
|
|
|
|
|
|
|
if not a.lower().endswith(".xml"):
|
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
|
|
|
|
xml_path = os.path.join(carpeta, a)
|
|
|
|
|
|
log(f"📄 Procesando XML: {a}", 3)
|
|
|
|
|
|
data = sunat_schema.extraer_datos_xml(xml_path)
|
|
|
|
|
|
|
|
|
|
|
|
if not data:
|
|
|
|
|
|
log("ℹ️ XML sin datos (no es factura tipo 01 o guía 31)", 3)
|
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
|
|
|
|
# CDR solo se renombra, no se procesa como documento
|
|
|
|
|
|
if data.get("es_cdr"):
|
|
|
|
|
|
prefijo = "CDR-"
|
|
|
|
|
|
cdr_name = f"{prefijo}{data['ruc_emisor']}-{data['tipo']}-{data['serie']}-{data['numero']}.xml"
|
|
|
|
|
|
cdr_dest = os.path.join(carpeta, cdr_name)
|
|
|
|
|
|
if cdr_dest != xml_path:
|
|
|
|
|
|
if os.path.exists(cdr_dest):
|
|
|
|
|
|
os.remove(cdr_dest)
|
|
|
|
|
|
os.rename(xml_path, cdr_dest)
|
|
|
|
|
|
email["adjuntos"][idx] = cdr_name
|
|
|
|
|
|
log(f"ℹ️ {data.get('serie','?')}-{data.get('numero','?')} (CDR) — solo renombrado", 3)
|
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
|
|
|
|
data["ruta_xml"] = xml_path
|
|
|
|
|
|
|
|
|
|
|
|
# Renombrar XML a {prefijo}{ruc}-{tipo}-{serie}-{numero}.xml
|
|
|
|
|
|
prefijo = ""
|
|
|
|
|
|
xml_canonical = f"{prefijo}{data.get('ruc_emisor', '')}-{data.get('tipo', '')}-{data.get('serie', '')}-{data.get('numero', '')}.xml"
|
|
|
|
|
|
xml_destino = os.path.join(carpeta, xml_canonical)
|
|
|
|
|
|
if xml_destino != xml_path:
|
|
|
|
|
|
if os.path.exists(xml_destino):
|
|
|
|
|
|
os.remove(xml_destino)
|
|
|
|
|
|
os.rename(xml_path, xml_destino)
|
|
|
|
|
|
email["adjuntos"][idx] = xml_canonical
|
|
|
|
|
|
data["ruta_xml"] = xml_destino
|
|
|
|
|
|
xml_path = xml_destino
|
|
|
|
|
|
|
|
|
|
|
|
# ruta_pdf: construir nombre canónico {ruc}-{tipo}-{serie}-{numero}.pdf
|
|
|
|
|
|
pdf_canonical = f"{data.get('ruc_emisor', '')}-{data.get('tipo', '')}-{data.get('serie', '')}-{data.get('numero', '')}.pdf"
|
|
|
|
|
|
pdf_path = os.path.join(carpeta, pdf_canonical)
|
|
|
|
|
|
if not os.path.exists(pdf_path):
|
|
|
|
|
|
# Fallback: reemplazar extensión .xml por .pdf
|
|
|
|
|
|
pdf_path = os.path.splitext(xml_path)[0] + ".pdf"
|
|
|
|
|
|
pdf_path = pdf_path if os.path.exists(pdf_path) else None
|
|
|
|
|
|
data["ruta_pdf"] = pdf_path
|
|
|
|
|
|
|
|
|
|
|
|
copiar_a_red(data)
|
|
|
|
|
|
|
|
|
|
|
|
# Cruzar documentos_referencia desde el PDF renombrado (pdf_cache por nombre canónico)
|
|
|
|
|
|
if not data.get("documentos_referencia"):
|
|
|
|
|
|
pdf_data = pdf_cache.get(pdf_canonical)
|
|
|
|
|
|
if pdf_data:
|
|
|
|
|
|
refs = pdf_utilities.extraer_referencias(pdf_data.get("items", []))
|
|
|
|
|
|
if refs:
|
|
|
|
|
|
data["documentos_referencia"] = refs
|
|
|
|
|
|
log(f"🔄 documentos_referencia completados desde PDF ({len(refs)} refs)", 3)
|
|
|
|
|
|
if not data.get("documentos_referencia"):
|
|
|
|
|
|
for pdf_data in email.get("datos_pdf", []):
|
|
|
|
|
|
refs = pdf_utilities.extraer_referencias(pdf_data.get("items", []))
|
|
|
|
|
|
if refs:
|
|
|
|
|
|
data["documentos_referencia"] = refs
|
|
|
|
|
|
log(f"🔄 documentos_referencia completados desde PDF (fallback, {len(refs)} refs)", 3)
|
|
|
|
|
|
break
|
|
|
|
|
|
|
|
|
|
|
|
# Asignar guias_remision[] y guia_transportista por línea
|
|
|
|
|
|
# Prioridad: notes_text (ya parseado) > documentos_referencia
|
|
|
|
|
|
refs_09 = [r for r in data.get("documentos_referencia", []) if r.get("tipo") == "09"]
|
|
|
|
|
|
refs_31 = [r for r in data.get("documentos_referencia", []) if r.get("tipo") == "31"]
|
|
|
|
|
|
|
|
|
|
|
|
for i, linea in enumerate(data.get("lineas", [])):
|
|
|
|
|
|
gr = linea.get("guias_remision") or []
|
|
|
|
|
|
gt = linea.get("guia_transportista")
|
|
|
|
|
|
|
|
|
|
|
|
if not gr and i < len(refs_09):
|
|
|
|
|
|
es_ultima = i == len(data["lineas"]) - 1
|
|
|
|
|
|
if es_ultima:
|
|
|
|
|
|
gr = list(refs_09[i:])
|
|
|
|
|
|
else:
|
|
|
|
|
|
gr = [dict(refs_09[i])]
|
|
|
|
|
|
if not gt and i < len(refs_31):
|
|
|
|
|
|
gt = dict(refs_31[i])
|
|
|
|
|
|
|
|
|
|
|
|
linea["guias_remision"] = gr
|
|
|
|
|
|
if gt:
|
|
|
|
|
|
linea["guia_transportista"] = gt
|
|
|
|
|
|
|
|
|
|
|
|
log(f"✅ Datos extraídos: {data.get('serie')}-{data.get('numero')} ({data.get('tipo_descripcion', '?')})", 3)
|
|
|
|
|
|
documentos.append(data)
|
|
|
|
|
|
|
|
|
|
|
|
# Post-procesamiento: resolver GREs desde guías de transporte referenciadas
|
|
|
|
|
|
guia_lookup = {}
|
|
|
|
|
|
for doc in documentos:
|
|
|
|
|
|
if doc.get("tipo") in ("31", "09"):
|
|
|
|
|
|
guia_lookup[(doc["tipo"], doc["serie"], doc["numero"])] = doc
|
|
|
|
|
|
|
|
|
|
|
|
for doc in documentos:
|
|
|
|
|
|
if doc.get("tipo") != "01":
|
|
|
|
|
|
continue
|
|
|
|
|
|
for ref in doc.get("documentos_referencia", []):
|
|
|
|
|
|
if ref["tipo"] == "31":
|
|
|
|
|
|
guia_doc = guia_lookup.get(("31", ref["serie"], ref["numero"]))
|
|
|
|
|
|
if guia_doc:
|
|
|
|
|
|
gre_list = [g for g in guia_doc.get("documentos_referencia", []) if g.get("tipo") == "09"]
|
|
|
|
|
|
if gre_list:
|
|
|
|
|
|
log(f"🔗 Resueltas {len(gre_list)} GRE(s) desde guía {ref['serie']}-{ref['numero']}", 3)
|
|
|
|
|
|
for i, linea in enumerate(doc.get("lineas", [])):
|
|
|
|
|
|
if not linea.get("guias_remision"):
|
|
|
|
|
|
es_ultima = i == len(doc["lineas"]) - 1
|
|
|
|
|
|
if es_ultima:
|
|
|
|
|
|
linea["guias_remision"] = gre_list[i:]
|
|
|
|
|
|
elif i < len(gre_list):
|
|
|
|
|
|
linea["guias_remision"] = [gre_list[i]]
|
|
|
|
|
|
if not linea.get("guia_transportista"):
|
|
|
|
|
|
linea["guia_transportista"] = dict(ref)
|
|
|
|
|
|
|
|
|
|
|
|
return documentos
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ============================================================================
|
|
|
|
|
|
def completar_guias(documentos: list[dict], pdf_cache: dict) -> int:
|
|
|
|
|
|
"""
|
|
|
|
|
|
Completa guías faltantes en XML desde datos del PDF.
|
|
|
|
|
|
|
|
|
|
|
|
Para cada documento en ``documentos``, verifica si todas las líneas
|
|
|
|
|
|
tienen ``guias_remision``. Si faltan, busca el PDF correspondiente en
|
|
|
|
|
|
el cache y copia las guías de las líneas del PDF.
|
|
|
|
|
|
|
|
|
|
|
|
Args:
|
|
|
|
|
|
documentos: Lista de documentos extraídos de XML.
|
|
|
|
|
|
pdf_cache: Cache de datos PDF (canonical_name → datos).
|
|
|
|
|
|
|
|
|
|
|
|
Returns:
|
|
|
|
|
|
int — Número de guías copiadas.
|
|
|
|
|
|
"""
|
|
|
|
|
|
total = 0
|
|
|
|
|
|
for xd in documentos:
|
|
|
|
|
|
has_all = all(
|
|
|
|
|
|
len(linea.get("guias_remision") or []) > 0
|
|
|
|
|
|
for linea in xd.get("lineas", [])
|
|
|
|
|
|
)
|
|
|
|
|
|
if has_all:
|
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
|
|
|
|
pdf_canonical = f"{xd.get('ruc_emisor', '')}-{xd.get('tipo', '')}-{xd.get('serie', '')}-{xd.get('numero', '')}.pdf"
|
|
|
|
|
|
pd = pdf_cache.get(pdf_canonical)
|
|
|
|
|
|
if not pd:
|
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
|
|
|
|
for i, xl in enumerate(xd.get("lineas", [])):
|
|
|
|
|
|
if xl.get("guias_remision"):
|
|
|
|
|
|
continue
|
|
|
|
|
|
pl = pd["items"][i] if i < len(pd.get("items", [])) else {}
|
|
|
|
|
|
p_gr = pl.get("guias_remision") or []
|
|
|
|
|
|
if isinstance(p_gr[0], str) if p_gr else False:
|
|
|
|
|
|
# Normalizar formato string → dict
|
|
|
|
|
|
p_gr = [
|
|
|
|
|
|
{"tipo": "09", "serie": m.group(1).upper(), "numero": m.group(2).lstrip("0") or "0"}
|
|
|
|
|
|
for g in p_gr
|
|
|
|
|
|
for m in [re.search(r"\b([TVE][A-Z0-9]{3,4})\s*-?\s*(\d+)", g)]
|
|
|
|
|
|
if m
|
|
|
|
|
|
]
|
|
|
|
|
|
if p_gr:
|
|
|
|
|
|
xl["guias_remision"] = list(p_gr)
|
|
|
|
|
|
total += len(p_gr)
|
|
|
|
|
|
if not xl.get("guia_transportista") and pl.get("guia_transportista"):
|
|
|
|
|
|
gt = pl["guia_transportista"]
|
|
|
|
|
|
if isinstance(gt, str):
|
|
|
|
|
|
m = re.search(r"\b([TVE][A-Z0-9]{3,4})\s*-?\s*(\d+)", gt)
|
|
|
|
|
|
if m:
|
|
|
|
|
|
xl["guia_transportista"] = {
|
|
|
|
|
|
"tipo": "31",
|
|
|
|
|
|
"serie": m.group(1).upper(),
|
|
|
|
|
|
"numero": m.group(2).lstrip("0") or "0",
|
|
|
|
|
|
}
|
|
|
|
|
|
elif isinstance(gt, dict):
|
|
|
|
|
|
xl["guia_transportista"] = dict(gt)
|
|
|
|
|
|
return total
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ============================================================================
|
|
|
|
|
|
def validar_guias(documentos: list[dict]) -> None:
|
|
|
|
|
|
for doc in documentos:
|
|
|
|
|
|
for i, linea in enumerate(doc.get("lineas", [])):
|
|
|
|
|
|
for j, g in enumerate(linea.get("guias_remision") or []):
|
|
|
|
|
|
if not isinstance(g, dict) or not all(k in g and g[k] for k in ("tipo", "serie", "numero")):
|
2026-10-14 22:42:37 -05:00
|
|
|
|
log(f"⚠️ guias_remision[{j}] inválida en {doc.get('serie','?')}-{doc.get('numero','?')} línea {i}: {g}", 2, severity=logging.WARNING)
|
2026-10-14 22:28:48 -05:00
|
|
|
|
gt = linea.get("guia_transportista")
|
|
|
|
|
|
if gt and (not isinstance(gt, dict) or not all(k in gt and gt[k] for k in ("tipo", "serie", "numero"))):
|
2026-10-14 22:42:37 -05:00
|
|
|
|
log(f"⚠️ guia_transportista inválida en {doc.get('serie','?')}-{doc.get('numero','?')} línea {i}: {gt}", 2, severity=logging.WARNING)
|
