146 lines
5.0 KiB
Python
146 lines
5.0 KiB
Python
 
|
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(__file__), "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()
|