se divide pipeline.py en módulos (adjuntos, documentos, persistencia, draft_sap, notificaciones)
Las funciones anidadas de _procesar_un_correo pasan a funciones de módulo con parámetros explícitos; pipeline.py queda solo como orquestador. copiar_a_red lee las variables de entorno al ejecutarse. Se actualizan docs. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
GeorgeandClaude Sonnet 5.5 committed
1 parent
ee64e5f773
commit
7d2f3212c0
9 files changed
+1309
-1228
No files matched your search
+371
@@ -0,0 +1,371 @@
|
||||
"""Extracción de datos de PDF/XML y asignación de guías de remisión."""
|
||||
|
||||
import os
|
||||
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):
|
||||
log(f"⚠️ No se pudo copiar a red: fecha_emision inválida ({data.get('fecha_emision')})", 3)
|
||||
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:
|
||||
log(f"⚠️ Error copiando a ruta de red: {e}", 3)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
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:
|
||||
log(f"⚠️ PDF no procesado: {adjunto} - {data.get('error', 'error desconocido')}", 2)
|
||||
|
||||
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")):
|
||||
log(f"⚠️ guias_remision[{j}] inválida en {doc.get('serie','?')}-{doc.get('numero','?')} línea {i}: {g}", 2)
|
||||
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"))):
|
||||
log(f"⚠️ guia_transportista inválida en {doc.get('serie','?')}-{doc.get('numero','?')} línea {i}: {gt}", 2)
|
||||
Reference in new issue
Block a user