GeorgeandClaude Sonnet 5.5
fcce30eda6
- HORARIOS cubre las 24 horas: se agregan 21:30-24:00 (cada 60 min) y 05:00-06:30 (cada 30 min); antes el demonio dormía hasta las 06:30 en esos tramos. Se corrigen los comentarios, que no coincidían con los valores. tests/test_horarios.py garantiza que no haya huecos ni solapes. - La carpeta de resultados se resuelve junto al código, no relativa al cwd (un servicio de Windows puede arrancar desde cualquier carpeta). - copiar_a_red no crea carpetas locales cuando RUTA_RED_FACTURACION no es una ruta absoluta en el sistema operativo (una UNC fuera de Windows); antes el draft podía referenciar archivos que no estaban en el recurso compartido. - El log muestra la versión del proyecto (pyproject.toml) al arrancar. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
379 lines
16 KiB
Python
379 lines
16 KiB
Python
"""Extracción de datos de PDF/XML y asignación de guías de remisión."""
|
||
|
||
import os
|
||
import logging
|
||
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
|
||
if not os.path.isabs(RUTA_RED_FACTURACION):
|
||
# Fuera de Windows una ruta UNC (\\\\servidor\\recurso) no es absoluta: makedirs la crearía como una carpeta
|
||
# local con ese nombre, y el draft apuntaría a archivos que no están en el recurso compartido.
|
||
log("⚠️ RUTA_RED_FACTURACION no es una ruta absoluta en este sistema operativo: no se copia a la red", 3,
|
||
severity=logging.WARNING)
|
||
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, severity=logging.WARNING)
|
||
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, severity=logging.WARNING)
|
||
|
||
|
||
# ============================================================================
|
||
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, severity=logging.WARNING)
|
||
|
||
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, severity=logging.WARNING)
|
||
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, severity=logging.WARNING)
|