se reorganiza tests/ y se agrega suite pytest
Los scripts manuales pasan a tools/ y los correos de ejemplo a tests/fixtures/. Suite de 95 pruebas: caracterización de la extracción sobre los 46 correos (resultados esperados en tests/golden/, fuera de la carpeta que descargar_correos_prueba borra), unitarias de draft_sap, notificaciones, documentos, adjuntos y config_validator, y flujo completo de _procesar_un_correo con SAP/HANA/Outlook simulados. pytest como dependencia dev y configuración en pyproject.toml. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
GeorgeandClaude Sonnet 5.5 committed
1 parent
b1b6f70983
commit
1d6d869111
261 files changed
+7876
-7
No files matched your search
@@ -0,0 +1,76 @@
|
||||
import json
|
||||
import os
|
||||
from glob import glob
|
||||
|
||||
CARPETA_BASE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "tests", "fixtures", "correos_prueba")
|
||||
|
||||
|
||||
def _doc_key(d):
|
||||
return (d.get("tipo"), d.get("serie"), d.get("numero"))
|
||||
|
||||
|
||||
def completar_guias_desde_pdf(xml_path: str, pdf_path: str) -> int:
|
||||
with open(xml_path, encoding="utf-8") as f:
|
||||
xml_data = json.load(f)
|
||||
try:
|
||||
with open(pdf_path, encoding="utf-8") as f:
|
||||
pdf_data = json.load(f)
|
||||
except FileNotFoundError:
|
||||
return 0
|
||||
|
||||
pdf_by_key = {_doc_key(d): d for d in pdf_data}
|
||||
total_copiadas = 0
|
||||
|
||||
for xd in xml_data:
|
||||
k = _doc_key(xd)
|
||||
|
||||
has_all_guias = all(
|
||||
len(linea.get("guias_remision") or []) > 0
|
||||
for linea in xd.get("lineas", [])
|
||||
)
|
||||
if has_all_guias:
|
||||
continue
|
||||
|
||||
pd = pdf_by_key.get(k)
|
||||
if not pd:
|
||||
continue
|
||||
|
||||
for li, xl in enumerate(xd.get("lineas", [])):
|
||||
if len(xl.get("guias_remision") or []) > 0:
|
||||
continue
|
||||
pl = pd["lineas"][li] if li < len(pd.get("lineas", [])) else {}
|
||||
p_gr = pl.get("guias_remision") or []
|
||||
if p_gr:
|
||||
xl["guias_remision"] = list(p_gr)
|
||||
total_copiadas += len(p_gr)
|
||||
# También copiar guia_transportista si XML no tiene
|
||||
if not xl.get("guia_transportista") and pl.get("guia_transportista"):
|
||||
xl["guia_transportista"] = dict(pl["guia_transportista"])
|
||||
|
||||
if total_copiadas > 0:
|
||||
with open(xml_path, "w", encoding="utf-8") as f:
|
||||
json.dump(xml_data, f, ensure_ascii=False, indent=2)
|
||||
|
||||
return total_copiadas
|
||||
|
||||
|
||||
def main():
|
||||
carpetas = sorted(glob(os.path.join(CARPETA_BASE, "[0-9][0-9][0-9]_correo")))
|
||||
total_global = 0
|
||||
for fp in carpetas:
|
||||
folder = os.path.basename(fp)
|
||||
idx = folder[:3]
|
||||
xml_path = os.path.join(fp, f"xml_resultado_{idx}.json")
|
||||
pdf_path = os.path.join(fp, f"pdf_resultado_{idx}.json")
|
||||
if not os.path.exists(xml_path):
|
||||
continue
|
||||
n = completar_guias_desde_pdf(xml_path, pdf_path)
|
||||
if n > 0:
|
||||
print(f" {folder}: {n} guías copiadas")
|
||||
total_global += n
|
||||
|
||||
print(f"\nTotal guías copiadas: {total_global}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,99 @@
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import shutil
|
||||
from datetime import datetime
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
|
||||
from dotenv import load_dotenv
|
||||
import outlook_utilities
|
||||
import email_utilities
|
||||
|
||||
|
||||
CARPETA_BASE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "tests", "fixtures", "correos_prueba")
|
||||
|
||||
|
||||
def limpiar_carpetas():
|
||||
if os.path.exists(CARPETA_BASE):
|
||||
shutil.rmtree(CARPETA_BASE)
|
||||
print(f"🗑️ Carpeta {CARPETA_BASE} eliminada")
|
||||
os.makedirs(CARPETA_BASE, exist_ok=True)
|
||||
print(f"📁 Carpeta {CARPETA_BASE} creada")
|
||||
|
||||
|
||||
def descargar_correos():
|
||||
load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), "..", ".env"))
|
||||
|
||||
EMAIL = os.getenv("EMAIL", "")
|
||||
if not EMAIL:
|
||||
print("❌ Variable EMAIL no encontrada en .env")
|
||||
return
|
||||
|
||||
print("🔑 Autenticando en Microsoft Graph...")
|
||||
token = outlook_utilities.get_access_token()
|
||||
if not token:
|
||||
print("❌ No se pudo obtener token")
|
||||
return
|
||||
|
||||
fecha_inicio = "2026-06-01"
|
||||
fecha_fin = "2026-07-31"
|
||||
print(f"📬 Descargando correos de {fecha_inicio} a {fecha_fin}...")
|
||||
|
||||
emails = outlook_utilities.get_emails(EMAIL, token, fecha_inicio, fecha_fin)
|
||||
if not emails:
|
||||
print("❌ No se encontraron correos")
|
||||
return
|
||||
|
||||
print(f"📨 Total correos encontrados: {len(emails)}")
|
||||
|
||||
indice = {}
|
||||
|
||||
for idx, email in enumerate(emails, start=1):
|
||||
correlativo = f"{idx:03d}"
|
||||
carpeta_correo = os.path.join(CARPETA_BASE, f"{correlativo}_correo")
|
||||
os.makedirs(carpeta_correo, exist_ok=True)
|
||||
|
||||
print(f"\n[{correlativo}] 📬 {email.get('fecha','')} - {email.get('subject','')[:60]}...")
|
||||
|
||||
adjuntos = outlook_utilities.download_attachments(
|
||||
EMAIL, email["id"], token, carpeta_correo
|
||||
)
|
||||
|
||||
if not adjuntos:
|
||||
print(f" ├── ⏭️ Sin adjuntos")
|
||||
os.rmdir(carpeta_correo)
|
||||
continue
|
||||
|
||||
email["adjuntos"] = adjuntos
|
||||
email_utilities.descomprimir_adjuntos([email], carpeta_correo)
|
||||
|
||||
archivos_validos = email_utilities.filtrar_archivos_validos(email.get("adjuntos", []))
|
||||
carpeta_abs = os.path.abspath(carpeta_correo)
|
||||
for archivo in email.get("adjuntos", []):
|
||||
ruta = os.path.join(carpeta_abs, archivo)
|
||||
if archivo not in archivos_validos and os.path.exists(ruta):
|
||||
os.remove(ruta)
|
||||
|
||||
if archivos_validos:
|
||||
email["adjuntos"] = archivos_validos
|
||||
|
||||
indice[correlativo] = {
|
||||
"id": email["id"],
|
||||
"fecha": email.get("fecha", ""),
|
||||
"subject": email.get("subject", ""),
|
||||
}
|
||||
print(f" └── ✅ {len(archivos_validos)} archivo(s) válido(s)")
|
||||
|
||||
ruta_indice = os.path.join(CARPETA_BASE, "indice_correos.json")
|
||||
with open(ruta_indice, "w", encoding="utf-8") as f:
|
||||
json.dump(indice, f, ensure_ascii=False, indent=2)
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f"📊 Total correos descargados: {len(indice)}")
|
||||
print(f"📄 Índice guardado en: {ruta_indice}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
limpiar_carpetas()
|
||||
descargar_correos()
|
||||
@@ -0,0 +1,48 @@
|
||||
import os
|
||||
import sys
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
|
||||
from sunat_schema import NAMESPACES, _find_text
|
||||
|
||||
ns = NAMESPACES
|
||||
ns["cbc"] = "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2"
|
||||
|
||||
CARPETA_BASE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "tests", "fixtures", "correos_prueba")
|
||||
|
||||
|
||||
def explorar_notes(ruta_carpeta):
|
||||
archivos = sorted(
|
||||
os.path.join(ruta_carpeta, f) for f in os.listdir(ruta_carpeta)
|
||||
if f.lower().endswith(".xml") and os.path.isfile(os.path.join(ruta_carpeta, f))
|
||||
)
|
||||
|
||||
for xml_path in archivos:
|
||||
tree = ET.parse(xml_path)
|
||||
root = tree.getroot()
|
||||
|
||||
if "Invoice" not in root.tag:
|
||||
continue
|
||||
|
||||
tipo = _find_text(root, "inv:InvoiceTypeCode", ns) or _find_text(root, "cbc:InvoiceTypeCode", ns) or ""
|
||||
if tipo != "01":
|
||||
continue
|
||||
|
||||
print(f"\n=== {os.path.basename(xml_path)} ===")
|
||||
lineas = root.findall(".//cac:InvoiceLine", ns)
|
||||
for i, linea in enumerate(lineas):
|
||||
notes = linea.findall("cbc:Note", ns)
|
||||
desc = _find_text(linea, "cac:Item/cbc:Description", ns) or ""
|
||||
print(f" Línea {i}: Description={desc[:80]}...")
|
||||
for note in notes:
|
||||
print(f" Note[{note.attrib.get('languageLocaleID', '?')}]: {note.text}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
carpeta = sys.argv[1] if len(sys.argv) > 1 else "001_correo"
|
||||
ruta = os.path.join(CARPETA_BASE, carpeta)
|
||||
if os.path.isdir(ruta):
|
||||
explorar_notes(ruta)
|
||||
else:
|
||||
print(f"❌ Carpeta no encontrada: {ruta}")
|
||||
@@ -0,0 +1,146 @@
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
from typing import Optional
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
|
||||
from pdf_utilities import extraer_datos_pdf, TIPO_MAP, renombrar_pdf
|
||||
from sunat_schema import split_serie_numero
|
||||
|
||||
|
||||
CARPETA_BASE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "tests", "fixtures", "correos_prueba")
|
||||
|
||||
|
||||
def normalizar_datos_pdf(data: dict) -> Optional[dict]:
|
||||
if not data or data.get("error"):
|
||||
return None
|
||||
|
||||
tipo = TIPO_MAP.get((data.get("tipo_documento") or "").upper())
|
||||
if not tipo:
|
||||
return None
|
||||
|
||||
serie_numero = data.get("serie_numero") or ""
|
||||
serie, numero = split_serie_numero(serie_numero)
|
||||
|
||||
doc = {
|
||||
"ruc_emisor": data.get("ruc_emisor"),
|
||||
"razon_social_emisor": data.get("nombre_receptor"),
|
||||
"fecha_emision": data.get("fecha_emision"),
|
||||
"tipo": tipo,
|
||||
"serie": serie,
|
||||
"numero": numero,
|
||||
"ruc_receptor": data.get("ruc_receptor"),
|
||||
"razon_social_receptor": data.get("nombre_receptor"),
|
||||
"total_factura": data.get("monto_total"),
|
||||
"documentos_referencia": [],
|
||||
"tiene_detraccion": False,
|
||||
"codigo_detraccion": None,
|
||||
"porcentaje_detraccion": None,
|
||||
"monto_detraccion": None,
|
||||
"lineas": [],
|
||||
}
|
||||
|
||||
# Detracción
|
||||
det = data.get("detraccion")
|
||||
if det and det.get("tiene"):
|
||||
doc["tiene_detraccion"] = True
|
||||
doc["porcentaje_detraccion"] = det.get("porcentaje")
|
||||
doc["monto_detraccion"] = det.get("monto")
|
||||
|
||||
# Documentos relacionados (para guía transportista)
|
||||
for rel in data.get("documentos_relacionados") or []:
|
||||
if isinstance(rel, dict):
|
||||
doc_ref = rel.get("documento", "")
|
||||
else:
|
||||
doc_ref = str(rel)
|
||||
r_serie, r_numero = split_serie_numero(doc_ref)
|
||||
if r_serie:
|
||||
doc["documentos_referencia"].append({
|
||||
"tipo": "09" if r_serie[:1] == "T" else "31",
|
||||
"serie": r_serie,
|
||||
"numero": r_numero,
|
||||
})
|
||||
|
||||
# Líneas de detalle
|
||||
for item in data.get("items") or []:
|
||||
linea = {
|
||||
"descripcion": item.get("descripcion", ""),
|
||||
"cantidad": item.get("cantidad"),
|
||||
"precio_unitario": item.get("precio_unitario"),
|
||||
"total_linea": item.get("total"),
|
||||
"guias_remision": [],
|
||||
"guia_transportista": None,
|
||||
}
|
||||
# Guías de remisión desde el PDF
|
||||
for g in item.get("guias_remision") or []:
|
||||
m = __import__("re").search(r"\b([TVE][A-Z0-9]{3,4})\s*-?\s*(\d+)", g)
|
||||
if m:
|
||||
linea["guias_remision"].append({
|
||||
"tipo": "09",
|
||||
"serie": m.group(1).upper(),
|
||||
"numero": m.group(2).lstrip("0") or "0",
|
||||
})
|
||||
# Guía transportista desde el PDF
|
||||
gt = item.get("guia_transportista")
|
||||
if gt:
|
||||
m = __import__("re").search(r"\b([TVE][A-Z0-9]{3,4})\s*-?\s*(\d+)", gt)
|
||||
if m:
|
||||
linea["guia_transportista"] = {
|
||||
"tipo": "31",
|
||||
"serie": m.group(1).upper(),
|
||||
"numero": m.group(2).lstrip("0") or "0",
|
||||
}
|
||||
doc["lineas"].append(linea)
|
||||
|
||||
return doc
|
||||
|
||||
|
||||
def procesar_carpetas():
|
||||
carpetas = sorted(
|
||||
[d for d in os.listdir(CARPETA_BASE) if os.path.isdir(os.path.join(CARPETA_BASE, d))]
|
||||
)
|
||||
|
||||
total = 0
|
||||
for carpeta in carpetas:
|
||||
ruta_carpeta = os.path.join(CARPETA_BASE, carpeta)
|
||||
archivos_pdf = sorted(
|
||||
os.path.join(ruta_carpeta, f) for f in os.listdir(ruta_carpeta)
|
||||
if f.lower().endswith(".pdf") and os.path.isfile(os.path.join(ruta_carpeta, f))
|
||||
)
|
||||
|
||||
if not archivos_pdf:
|
||||
continue
|
||||
|
||||
correlativo = carpeta.split("_")[0]
|
||||
|
||||
documentos = []
|
||||
for pdf_path in archivos_pdf:
|
||||
print(f"📄 {carpeta}/{os.path.basename(pdf_path)}")
|
||||
data = extraer_datos_pdf(pdf_path)
|
||||
if data and not data.get("error"):
|
||||
nuevo_path = renombrar_pdf(pdf_path, data)
|
||||
if nuevo_path != pdf_path:
|
||||
data["archivo"] = os.path.basename(nuevo_path)
|
||||
doc = normalizar_datos_pdf(data)
|
||||
if doc:
|
||||
doc["ruta_pdf"] = nuevo_path
|
||||
documentos.append(doc)
|
||||
desc = f"{doc.get('tipo','?')} {doc.get('serie','?')}-{doc.get('numero','?')}"
|
||||
print(f" ├── ✅ {desc}")
|
||||
else:
|
||||
print(f" ├── ⚠️ Error: {data.get('error', 'desconocido')}")
|
||||
|
||||
if documentos:
|
||||
ruta_json = os.path.join(ruta_carpeta, f"pdf_resultado_{correlativo}.json")
|
||||
with open(ruta_json, "w", encoding="utf-8") as f:
|
||||
json.dump(documentos, f, ensure_ascii=False, indent=2)
|
||||
print(f" └── 💾 Guardado: pdf_resultado_{correlativo}.json ({len(documentos)} doc(s))")
|
||||
total += len(documentos)
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f"📊 Total documentos PDF procesados: {total}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
procesar_carpetas()
|
||||
@@ -0,0 +1,117 @@
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
|
||||
from sunat_schema import extraer_datos_xml
|
||||
|
||||
|
||||
def renombrar_xml(ruta_actual: str, data: dict, carpeta: str) -> str | None:
|
||||
"""Renombra el XML a {prefijo}{ruc}-{tipo}-{serie}-{numero}.xml y retorna la nueva ruta.
|
||||
Si es un CDR (ApplicationResponse), el prefijo es 'CDR-'."""
|
||||
if not data:
|
||||
return None
|
||||
prefijo = "CDR-" if data.get("es_cdr") else ""
|
||||
nombre = f"{prefijo}{data.get('ruc_emisor', '')}-{data.get('tipo', '')}-{data.get('serie', '')}-{data.get('numero', '')}.xml"
|
||||
destino = os.path.join(carpeta, nombre)
|
||||
if destino != ruta_actual:
|
||||
if os.path.exists(destino):
|
||||
os.remove(destino)
|
||||
os.rename(ruta_actual, destino)
|
||||
return destino
|
||||
|
||||
|
||||
CARPETA_BASE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "tests", "fixtures", "correos_prueba")
|
||||
|
||||
|
||||
def procesar_carpetas():
|
||||
carpetas = sorted(
|
||||
[d for d in os.listdir(CARPETA_BASE) if os.path.isdir(os.path.join(CARPETA_BASE, d))]
|
||||
)
|
||||
|
||||
total = 0
|
||||
for carpeta in carpetas:
|
||||
ruta_carpeta = os.path.join(CARPETA_BASE, carpeta)
|
||||
archivos_xml = sorted(
|
||||
os.path.join(ruta_carpeta, f) for f in os.listdir(ruta_carpeta)
|
||||
if f.lower().endswith(".xml") and os.path.isfile(os.path.join(ruta_carpeta, f))
|
||||
)
|
||||
|
||||
if not archivos_xml:
|
||||
continue
|
||||
|
||||
correlativo = carpeta.split("_")[0]
|
||||
|
||||
documentos = []
|
||||
for xml_path in archivos_xml:
|
||||
print(f"📄 {carpeta}/{os.path.basename(xml_path)}")
|
||||
data = extraer_datos_xml(xml_path)
|
||||
if data:
|
||||
ruta_xml = renombrar_xml(xml_path, data, ruta_carpeta)
|
||||
data["ruta_xml"] = ruta_xml or xml_path
|
||||
data["ruta_pdf"] = None
|
||||
if data.get("es_cdr"):
|
||||
print(f" ├── ℹ️ {data.get('serie','?')}-{data.get('numero','?')} (CDR) — omitido del resultado")
|
||||
continue
|
||||
documentos.append(data)
|
||||
print(f" ├── ✅ {data.get('serie','?')}-{data.get('numero','?')} ({data.get('tipo_descripcion','?')})")
|
||||
else:
|
||||
print(f" ├── ℹ️ No es factura 01 ni guía 31")
|
||||
|
||||
if documentos:
|
||||
# Post-procesamiento
|
||||
# 1. Distribuir documentos_referencia (tipo 09 y 31) a cada línea
|
||||
for doc in documentos:
|
||||
if doc.get("tipo") != "01":
|
||||
continue
|
||||
refs_09 = [r for r in doc.get("documentos_referencia", []) if r.get("tipo") == "09"]
|
||||
refs_31 = [r for r in doc.get("documentos_referencia", []) if r.get("tipo") == "31"]
|
||||
for i, linea in enumerate(doc.get("lineas", [])):
|
||||
if linea.get("guias_remision"):
|
||||
continue
|
||||
if i < len(refs_09):
|
||||
es_ultima = i == len(doc["lineas"]) - 1
|
||||
if es_ultima:
|
||||
linea["guias_remision"] = [dict(r) for r in refs_09[i:]]
|
||||
else:
|
||||
linea["guias_remision"] = [dict(refs_09[i])]
|
||||
if not linea.get("guia_transportista") and i < len(refs_31):
|
||||
linea["guia_transportista"] = dict(refs_31[i])
|
||||
|
||||
# 2. Resolver GREs desde guías de transporte (tipo 31 → tipo 09)
|
||||
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:
|
||||
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)
|
||||
|
||||
ruta_json = os.path.join(ruta_carpeta, f"xml_resultado_{correlativo}.json")
|
||||
with open(ruta_json, "w", encoding="utf-8") as f:
|
||||
json.dump(documentos, f, ensure_ascii=False, indent=2)
|
||||
print(f" └── 💾 Guardado: xml_resultado_{correlativo}.json ({len(documentos)} doc(s))")
|
||||
total += len(documentos)
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f"📊 Total documentos XML procesados: {total}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
procesar_carpetas()
|
||||
Reference in new issue
Block a user