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:
1 parent
d4c1e0a264
commit
9d295d6485
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()
|
||||
Reference in new issue
Block a user