diff --git a/.gitignore b/.gitignore
index bb61760..4304f8b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -7,3 +7,7 @@ venv/
.idea/
logs/
adjuntos/
+resultado/
+emails.db
+.env.dev
+procesos/
diff --git a/config.json b/config.json
deleted file mode 100644
index ecdefa9..0000000
--- a/config.json
+++ /dev/null
@@ -1,24 +0,0 @@
-{
- "hana": {
- "server": "192.168.9.13",
- "port": 30015,
- "user": "SYSTEM",
- "password": ""
- },
- "outlook": {
- "client_id": "671f6154-715f-4e52-aef9-94ad108dc1d4",
- "client_secret": "oG58Q~yufghPKmZuh0oVqX6rcLedLP1HTwIC-bq5",
- "tenant_id": "5ac66bf8-fbbd-4615-80c0-8e3d521dc5e6",
- "download_folder": "./adjuntos",
- "verify_ssl": false,
- "ca_bundle": null
- },
- "client": {
- "email": "contabilidad1@aberio.com.pe",
- "ruta_descarga": "./adjuntos",
- "ruta_final": "./resultado",
- "base_datos": "SBO_ABERIO_PRODUCCION",
- "usuario": "manager",
- "password": "s1st3m4s"
- }
-}
\ No newline at end of file
diff --git a/config_loader.py b/config_loader.py
deleted file mode 100644
index b9641bb..0000000
--- a/config_loader.py
+++ /dev/null
@@ -1,22 +0,0 @@
-import json
-from pathlib import Path
-from config_model import Config, HanaConfig, OutlookConfig, ClientConfig
-
-########################################################
-CONFIG_PATH = Path(__file__).resolve().parent / "config.json"
-
-with open(CONFIG_PATH, 'r', encoding='utf-8') as f:
- data = json.load(f)
-
-# Instantiate dataclasses
-hana = HanaConfig(**data.get('hana', {}))
-outlook = OutlookConfig(**data.get('outlook', {}))
-client = ClientConfig(**data.get('client', {}))
-
-CONFIG = Config(
- hana=hana,
- outlook=outlook,
- client=client
-)
-
-########################################################
\ No newline at end of file
diff --git a/config_model.py b/config_model.py
deleted file mode 100644
index a103786..0000000
--- a/config_model.py
+++ /dev/null
@@ -1,33 +0,0 @@
-from dataclasses import dataclass
-from typing import Optional
-
-@dataclass
-class HanaConfig:
- server: str
- port: int
- user: str
- password: str
-
-@dataclass
-class OutlookConfig:
- client_id: str
- client_secret: str
- tenant_id: str
- download_folder: str
- verify_ssl: bool
- ca_bundle: Optional[str]
-
-@dataclass
-class ClientConfig:
- email: str
- ruta_descarga: str
- ruta_final: str
- base_datos: str
- usuario: str
- password: str
-
-@dataclass
-class Config:
- hana: HanaConfig
- outlook: OutlookConfig
- client: ClientConfig
\ No newline at end of file
diff --git a/consulta_rapida.py b/consulta_rapida.py
new file mode 100755
index 0000000..4f20915
--- /dev/null
+++ b/consulta_rapida.py
@@ -0,0 +1,115 @@
+"""
+Script para procesar correos específicos por ID desde un archivo JSON.
+No modifica el código existente del pipeline.
+"""
+import sys
+import os
+import json
+import argparse
+import requests
+from datetime import datetime, timedelta
+from dotenv import load_dotenv
+
+import database_utilities
+import logger_utilities
+from pipeline import _procesar_un_correo, step_authenticate
+
+
+# ============================================================================
+# OBTENER CORREO POR ID
+# ============================================================================
+def get_email_by_id(user_email: str, token: str, message_id: str) -> dict | None:
+ """Obtiene un correo desde Microsoft Graph por su ID de mensaje."""
+ url = f"https://graph.microsoft.com/v1.0/users/{user_email}/messages/{message_id}"
+ headers = {"Authorization": f"Bearer {token}"}
+ resp = requests.get(url, headers=headers, verify=False)
+ if resp.status_code != 200:
+ print(f" ❌ Error al obtener mensaje {message_id}: HTTP {resp.status_code}")
+ return None
+ msg = resp.json()
+ utc_str = msg.get("receivedDateTime", "")
+ dt_utc = datetime.fromisoformat(utc_str.replace("Z", "+00:00"))
+ dt_local = dt_utc - timedelta(hours=5)
+ from_obj = msg.get("from", {})
+ sender = from_obj.get("emailAddress", {}).get("address", "Desconocido")
+ return {
+ "id": msg.get("id"),
+ "fecha": dt_local.strftime("%Y-%m-%d %H:%M:%S"),
+ "sender": sender,
+ "subject": msg.get("subject", ""),
+ "has_attach": msg.get("hasAttachments"),
+ }
+
+
+# ============================================================================
+# PIPELINE PARA IDS
+# ============================================================================
+def run_pipeline_for_ids(ids: list[str]) -> None:
+ load_dotenv(dotenv_path=".env.dev")
+ email_addr = os.getenv("EMAIL", "")
+ download_folder = os.getenv("DOWNLOAD_FOLDER", "adjuntos")
+ output_folder = os.getenv("OUTPUT_FOLDER", "resultado")
+
+ if not email_addr:
+ print("❌ EMAIL no configurado en .env.dev")
+ return
+
+ database_utilities.init_db()
+ token = step_authenticate()
+ if not token:
+ return
+
+ for msg_id in ids:
+ email = get_email_by_id(email_addr, token, msg_id.strip())
+ if not email:
+ continue
+ _procesar_un_correo(email, token, email_addr, download_folder, output_folder)
+
+
+# ============================================================================
+# MAIN
+# ============================================================================
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser(
+ description="Procesar correos específicos por set de IDs desde tests/ids_prueba.json"
+ )
+ parser.add_argument(
+ "--set", "-s", required=True,
+ help="Nombre del set en tests/ids_prueba.json"
+ )
+ args = parser.parse_args()
+
+ json_path = os.path.join(os.path.dirname(__file__), "test", "ids_prueba.json")
+ if not os.path.exists(json_path):
+ print(f"❌ No se encuentra {json_path}")
+ sys.exit(1)
+
+ with open(json_path) as f:
+ sets = json.load(f)
+
+ if args.set not in sets:
+ print(f"❌ Set '{args.set}' no encontrado. Disponibles: {', '.join(sets.keys())}")
+ sys.exit(1)
+
+ ids = sets[args.set]["ids"]
+ descripcion = sets[args.set].get("descripcion", "")
+ print(f"\n📋 Set: {args.set} — {descripcion}")
+ print(f" IDs a procesar: {len(ids)}")
+
+ log_path = logger_utilities.setup_logger()
+ logger_utilities.limpiar_logs_antiguos(dias=30)
+
+ inicio = datetime.now()
+ print("\n" + "=" * 60)
+ print(f"🚀 INICIO DE EJECUCIÓN: {inicio.strftime('%Y-%m-%d %H:%M:%S')}")
+ print(f"📄 Log guardado en: {log_path}")
+ print("=" * 60)
+
+ run_pipeline_for_ids(ids)
+
+ fin = datetime.now()
+ duracion = fin - inicio
+ print("\n" + "=" * 60)
+ print(f"🏁 FIN DE EJECUCIÓN: {fin.strftime('%Y-%m-%d %H:%M:%S')}")
+ print(f"⏱️ Duración total: {str(duracion).split('.')[0]}")
+ print("=" * 60)
\ No newline at end of file
diff --git a/core.py b/core.py
deleted file mode 100644
index 0f085e9..0000000
--- a/core.py
+++ /dev/null
@@ -1,3 +0,0 @@
-
-def procesar_correo():
- pass
\ No newline at end of file
diff --git a/database_utilities.py b/database_utilities.py
index 7d10c44..666e795 100644
--- a/database_utilities.py
+++ b/database_utilities.py
@@ -1,9 +1,13 @@
import sqlite3
+# ============================================================================
+# CONFIGURACIÓN
+# ============================================================================
+
DB_NAME = "emails.db"
-########################################################
-def init_db():
+# ============================================================================
+def init_db() -> None:
"""
Inicializa la base de datos SQLite y crea la tabla de correos si no existe.
"""
@@ -25,8 +29,8 @@ def init_db():
conn.commit()
conn.close()
-########################################################
-def save_emails(emails_list):
+# ============================================================================
+def save_emails(emails_list: list[dict]) -> int:
"""
Guarda una lista de correos en la base de datos.
Retorna la cantidad de registros nuevos insertados.
@@ -65,8 +69,8 @@ def save_emails(emails_list):
conn.close()
return rows_affected
-########################################################
-def get_existing_ids(ids_list):
+# ============================================================================
+def get_existing_ids(ids_list: list[str]) -> list[str]:
"""
Recibe una lista de IDs y devuelve una lista con aquellos que ya existen en la BD.
"""
@@ -88,8 +92,8 @@ def get_existing_ids(ids_list):
conn.close()
return existing_ids
-########################################################
-def delete_email_by_id(email_id):
+# ============================================================================
+def delete_email_by_id(email_id: str) -> bool:
"""
Elimina un correo de la base de datos por su ID.
Retorna True si se eliminó el registro, False si no se encontró.
@@ -107,4 +111,4 @@ def delete_email_by_id(email_id):
conn.close()
return rows_affected > 0
-########################################################
\ No newline at end of file
+# ============================================================================
\ No newline at end of file
diff --git a/docs/construir_draft.mkd b/docs/construir_draft.mkd
new file mode 100644
index 0000000..1f5e261
--- /dev/null
+++ b/docs/construir_draft.mkd
@@ -0,0 +1,33 @@
+# Construir Draft SAP B1
+
+```mermaid
+flowchart TD
+ A["factura + resultados_val"] --> B["Determinar ItemCode
C103002-A (mes anterior)
C103002 (mes actual)"]
+ B --> C["Calcular cuotas
DocumentInstallments"]
+
+ C --> D{"¿Tiene detracción?"}
+ D -->|Sí| E["Cuota 1: monto detracción
U_VS_TIPOCUOTA = D"]
+ E --> F["Cuota 2: neto = total - detracción
U_VS_TIPOCUOTA = N"]
+ D -->|No| G["Cuota única por el total
U_VS_TIPOCUOTA = N"]
+
+ F --> H["Por cada línea de factura"]
+ G --> H
+
+ H --> I["Obtener guias_lt[]
desde resultado_validacion"]
+ I --> J["zona_guia = guias_lt[0]['zona']"]
+ J --> K["Por cada item en guias_lt[]"]
+
+ K --> L["PriceAfterVAT:
suma pav FL* si use_sap_prices
line_total si no"]
+ L --> M["Qty = item['quantity']"]
+ M --> N["Crear DocumentLine:
ItemCode, Quantity,
PriceAfterVAT, WarehouseCode,
U_IDC_GRFLETE"]
+
+ N --> O["Acumular total_weight
+= item['weight1']"]
+ O --> K
+
+ K --> P["U_IDC_VALREF =
round((total_weight / 1000) *
CUADRO_VALOR_REFERENCIAL[zona], 2)"]
+
+ P --> Q["Armar payload completo"]
+ Q --> R["Cabecera:
DocType, CardCode, NumAtCard,
DocDate, DocDueDate,
U_IDC_NUMGR, Comments"]
+ R --> S["DocumentLines[] +
DocumentInstallments[]"]
+ S --> T["← Retornar dict para
POST /Drafts"]
+```
diff --git a/docs/pipeline_completo.mkd b/docs/pipeline_completo.mkd
new file mode 100644
index 0000000..e750269
--- /dev/null
+++ b/docs/pipeline_completo.mkd
@@ -0,0 +1,29 @@
+# Pipeline Completo
+
+```mermaid
+flowchart TD
+ A["📬 Correo recibido"] --> B{"1-2. BD check
¿Ya procesado?"}
+ B -->|"Sí"| C["⏭️ Saltar correo"]
+ B -->|"No"| D["3. Descargar adjuntos
Microsoft Graph"]
+ D --> E{"¿Hay adjuntos?"}
+ E -->|No| C
+ E -->|Sí| F["4. Descomprimir ZIPs"]
+ F --> G["5. Filtrar solo PDF/XML"]
+ G --> H{"¿Quedan válidos?"}
+ H -->|No| C
+ H -->|Sí| I["6. Procesar PDFs
extraer datos + renombrar"]
+ I --> J["7. Procesar XMLs
cruzar con PDF cache"]
+ J --> K["8. Guardar resultado JSON"]
+ K --> L["9. Filtrar facturas tipo 01"]
+ L --> M{"¿Hay facturas?"}
+ M -->|No| N["⏭️ Sin facturas para SAP"]
+ M -->|Sí| O["10. Validar precios
_validate_prices()"]
+ O --> P["11. Construir draft SAP
_build_sap_draft()"]
+ P --> Q["12. Guardar draft JSON"]
+ Q --> R{"13. ¿Precios coinciden?"}
+ R -->|Sí| S["Crear Draft en SAP B1
POST /Drafts"]
+ R -->|No| T["Notificar: validación falló"]
+ S --> U["14. Enviar email notificación"]
+ T --> U
+ N --> U
+```
diff --git a/docs/validar_precios.mkd b/docs/validar_precios.mkd
new file mode 100644
index 0000000..cbb6158
--- /dev/null
+++ b/docs/validar_precios.mkd
@@ -0,0 +1,37 @@
+# Validar Precios
+
+```mermaid
+flowchart TD
+ A["factura con lineas[]"] --> B["Obtener maestro zonas
SAP U_IDC_ZONAS"]
+ B --> C["Por cada línea de factura"]
+
+ C --> D{"cantidad > 0?"}
+ D -->|No| E["Resultado: cantidad cero, sin validar"]
+ D -->|Sí| F{"¿Hay guías de remisión?"}
+ F -->|No| G["Resultado: sin guía de referencia"]
+
+ F -->|Sí| H["Por cada guía en guias[]"]
+ H --> I["Consultar SAP B1
get_delivery_notes()"]
+ I --> J["Por cada DeliveryNote"]
+
+ J --> K["Obtener zona desde
U_IDC_DESTINO"]
+ K --> L["Extraer WarehouseCode
de DocumentLines[0]"]
+ L --> M["Acumular Weight1
de DocumentLines[0]"]
+ M --> N["Por cada línea FL* en
DocumentLines"]
+
+
+ N --> O["Sumar LineTotal"]
+ O --> P["Máx Quantity"]
+ P --> Q["Sumar PriceAfterVAT"]
+ Q --> R["Construir guias_lt[]
para cada guía:
line_total, quantity,
price_after_vat_sum,
warehouse_code, zona, weight1"]
+
+ R --> S["Comparar total_linea (XML)
vs sap_total (FL*)"]
+ S --> T{"abs(diff) < 0.01?"}
+ T -->|Sí| U["ok = True
use_sap_prices si qty==1
y total_sap_qty≠1"]
+ T -->|No| V["ok = False"]
+
+ U --> W["Resultado por línea:
ok, draft_price, sap_price,
draft_cantidad, guias_lt[]"]
+ V --> W
+ W --> C
+ C --> X["todas_coinciden = all(ok)"]
+```
diff --git a/email_utilities.py b/email_utilities.py
index a05fafa..603d04c 100644
--- a/email_utilities.py
+++ b/email_utilities.py
@@ -2,8 +2,8 @@ import os
import zipfile
import database_utilities
-########################################################
-def quitar_correos_ya_registrados_en_base_de_datos(emails):
+# ============================================================================
+def quitar_correos_ya_registrados_en_base_de_datos(emails: list[dict]) -> list[dict]:
"""
Verifica cuáles correos son nuevos (no están en la BD), los guarda
y retorna la lista filtrada de los que tienen adjuntos.
@@ -25,8 +25,8 @@ def quitar_correos_ya_registrados_en_base_de_datos(emails):
return correos_no_registrados
-########################################################
-def descomprimir_adjuntos(emails, ruta):
+# ============================================================================
+def descomprimir_adjuntos(emails: list[dict], ruta: str) -> list[dict]:
"""
Busca adjuntos ZIP, los descomprime y agrega los archivos
extraídos a la lista de adjuntos del correo.
@@ -63,8 +63,8 @@ def descomprimir_adjuntos(emails, ruta):
print(f"❌ Error general: {ex}")
return emails
-########################################################
-def filtrar_archivos_validos(archivos):
+# ============================================================================
+def filtrar_archivos_validos(archivos: list[str]) -> list[str]:
"""
Retorna una lista solo con archivos .pdf y .xml (ignora mayúsculas/minúsculas).
@@ -72,8 +72,8 @@ def filtrar_archivos_validos(archivos):
extensiones_validas = ('.pdf', '.xml')
return [a for a in archivos if a.lower().endswith(extensiones_validas)]
-########################################################
-def procesar_adjuntos_email(e, archivos, ruta_descarga):
+# ============================================================================
+def procesar_adjuntos_email(e: dict, archivos: list[str], ruta_descarga: str) -> list[str]:
"""
Descomprime y filtra los archivos de un correo.
Retorna la lista de archivos válidos si existen, de lo contrario lista vacía.
@@ -89,5 +89,5 @@ def procesar_adjuntos_email(e, archivos, ruta_descarga):
return archivos_validos
return []
-########################################################
+# ============================================================================
diff --git a/logger_utilities.py b/logger_utilities.py
index 6d05b9e..b7cf4b3 100644
--- a/logger_utilities.py
+++ b/logger_utilities.py
@@ -1,26 +1,27 @@
import sys
import os
from datetime import datetime, timedelta
+from typing import TextIO
-########################################################
+# ============================================================================
class MultiStream:
"""
Clase para redirigir la salida a múltiples flujos (consola y archivo).
"""
- def __init__(self, *streams):
- self.streams = streams
+ def __init__(self, *streams: TextIO) -> None:
+ self.streams: tuple[TextIO, ...] = streams
- def write(self, data):
+ def write(self, data: str) -> None:
for stream in self.streams:
stream.write(data)
stream.flush()
- def flush(self):
+ def flush(self) -> None:
for stream in self.streams:
stream.flush()
-########################################################
-def setup_logger():
+# ============================================================================
+def setup_logger() -> str:
"""
Configura la redirección de stdout y stderr a un archivo de log y a la consola.
Retorna la ruta del archivo de log creado.
@@ -43,8 +44,8 @@ def setup_logger():
return log_path
-########################################################
-def limpiar_logs_antiguos(dias=7):
+# ============================================================================
+def limpiar_logs_antiguos(dias: int = 7) -> None:
"""
Elimina archivos de log con más de 'n' días de antigüedad.
"""
@@ -68,3 +69,4 @@ def limpiar_logs_antiguos(dias=7):
if count > 0:
print(f"🧹 Se eliminaron {count} archivos de log antiguos (más de {dias} días).")
+# ============================================================================
\ No newline at end of file
diff --git a/main.py b/main.py
index 98791d6..a37922b 100644
--- a/main.py
+++ b/main.py
@@ -1,207 +1,68 @@
import os
-import sys
-import json
-import email_utilities
-import outlook_utilities
-import database_utilities
+import argparse
+from datetime import datetime
+from dotenv import load_dotenv
+
import logger_utilities
-import sunat_schema
-from datetime import datetime, timedelta
-from config_loader import CONFIG
+from pipeline import run_pipeline
-########################################################
-def obtener_token():
- try:
- print("Intentando obtener token de acceso...")
- outlook_token = outlook_utilities.get_access_token()
- print("✅ Token obtenido exitosamente.")
- print(f"Token prefix: {outlook_token[:25]}...")
- return outlook_token
- except Exception as e:
- print(f"❌ Error: {e}")
- sys.exit(1)
+# ============================================================================
+def non_negative_int(value: str) -> int:
+ ivalue = int(value)
+ if ivalue < 0:
+ raise argparse.ArgumentTypeError(
+ f"{value} es un valor inválido. --dias debe ser 0 o mayor."
+ )
+ return ivalue
-########################################################
-def descargar_correos(outlook_token, fecha_inicio, fecha_fin):
+# ============================================================================
+def main() -> None:
- try:
- print("\n" + "="*60)
- print(f"[MAIL] {CONFIG.client.email}")
+ parser = argparse.ArgumentParser(
+ description="Obtener correos electrónicos de Office365 de hoy y los últimos X días."
+ )
- emails = outlook_utilities.get_emails(CONFIG.client.email, outlook_token, fecha_inicio, fecha_fin)
- print(f"Se encontraron {len(emails)} correos.")
+ parser.add_argument(
+ "--dias", "-d",
+ type=non_negative_int,
+ default=2,
+ help="Número de días respecto a la fecha actual para la consulta (default: 2)."
+ )
- return emails
-
- except Exception as e:
- print(f"❌ Error: {e}")
- sys.exit(1)
-
- return []
+ args = parser.parse_args()
-########################################################
-def descargar_adjuntos_de_correos(outlook_token, emails, ruta_descarga):
- try:
- emails_con_adjuntos = []
- for email in emails:
- # Guardar lista original para el log
- adjuntos = outlook_utilities.download_attachments(
- CONFIG.client.email,
- email['id'],
- outlook_token,
- ruta_descarga)
- if adjuntos:
- email['adjuntos'] = adjuntos
- emails_con_adjuntos.append(email)
-
- return emails_con_adjuntos
-
- except Exception as ex:
- print(f" ❌ Error procesando {email['id'][:10]}: {ex}")
+ dias = args.dias
- return []
+ if dias == 0:
+ print("📋 Obteniendo correos electrónicos de Office365 hoy.")
+ else:
+ print(f"📋 Obteniendo correos electrónicos de Office365 desde hace {dias} días hasta hoy.")
-########################################################
-def dejar_mail_si_tiene_adjunto_pdf_o_xml(emails):
- """
- Recorre los adjuntos de cada email y elimina los que no tengan extensión pdf o xml
- actualiza la lista de adjuntos del correo
- si el mail no tiene pdf ni xml también se elimina
- """
- emails_con_pdf_o_xml = []
- for email in emails:
- adjuntos_validos = [adjunto for adjunto in email['adjuntos'] if adjunto.lower().endswith(('.pdf', '.xml'))]
- if adjuntos_validos:
- email['adjuntos'] = adjuntos_validos
- emails_con_pdf_o_xml.append(email)
- return emails_con_pdf_o_xml
+ load_dotenv(dotenv_path=".env.dev")
-########################################################
-def leer_xml_obtener_datos_validos(emails):
- json_final = []
- for e in emails:
- print(f"\n📬 [{e['fecha']}] {e['subject'][:80]}...")
- #adjuntos originales
- print(f" ├── 📂 Adjuntos originales:")
+ EMAIL = os.getenv("EMAIL", "")
+ DOWNLOAD_FOLDER = os.getenv("DOWNLOAD_FOLDER", "")
+ OUTPUT_FOLDER = os.getenv("OUTPUT_FOLDER", "")
- for idx, a in enumerate(e['adjuntos']):
- prefijo = " │ └── " if idx == len(e['adjuntos'])-1 else " │ ├── "
- print(f"{prefijo}{a}")
- if a.lower().endswith(".xml"):
- # procesar archivo xml de SUNAT
- carpeta_adjuntos = os.path.join(os.path.dirname(__file__), CONFIG.client.ruta_descarga)
- archivo_xml = os.path.join(carpeta_adjuntos, a)
- print(f" │ 📄 Procesando XML: {a}")
- data = sunat_schema.extraer_datos_xml(archivo_xml)
- if data:
- print(f" │ ✅ Factura extraída: {data.get('serie')}-{data.get('numero')}")
- json_final.append(data)
- else:
- print(" │ ℹ️ XML sin datos (no es factura tipo 01)")
+ run_pipeline(dias, EMAIL, DOWNLOAD_FOLDER, OUTPUT_FOLDER)
- # opcional: guardar JSON junto al XML
- archivo_json = f"ejecucion_{datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}.json"
- with open(archivo_json, "w", encoding="utf-8") as f:
- json.dump(json_final, f, ensure_ascii=False, indent=4)
+# ============================================================================
- return json_final
-
-########################################################
-def anidar_guias_en_facturas(json_final):
- # obtener solo facturas
- facturas = [item for item in json_final if item.get("tipo") == "01"]
-
- # obtener solo guias
- guias = [item for item in json_final if item.get("tipo") == "31"]
-
- # recorrer cada factura y buscar guias relacionadas de campo "documento_relacionado" que tenga el mismo numero de serie y numero
- for factura in facturas:
- # obtener la guia de tipo ='31' de los documemtos_relacionados
- guias_transporte = [item for item in factura.get("documentos_referencia", []) if item.get("tipo") == "31"]
-
- # buscar en la lista de guias si existe una guia que tenga el mismo numero de serie y numero que la guia de transporte
- for guia in guias_transporte:
- guia_encontrada = next((g for g in guias if g.get("serie")== guia.get("serie")
- and g.get("numero")== guia.get("numero")), None)
- if guia_encontrada:
- # si se encuentra, anidar la guia dentro de la factura en un nuevo campo "guia_trasporte"
- factura["guia_transporte"] = guia_encontrada
- print(f" ├── 📦 Guía {guia_encontrada.get('serie')}-{guia_encontrada.get('numero')} anidada en factura {factura.get('serie')}-{factura.get('numero')}")
- else:
- print(f" ├── ⚠️ Guía {guia.get('serie')}-{guia.get('numero')} no encontrada para factura {factura.get('serie')}-{factura.get('numero')}")
-
- # opcional: guardar JSON junto al XML
- archivo_json = f"anidado_{datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}.json"
- with open(archivo_json, "w", encoding="utf-8") as f:
- json.dump(facturas, f, ensure_ascii=False, indent=4)
-
-
- return facturas
-########################################################
-def main():
-
- # Inicializar Base de Datos
- database_utilities.init_db()
-
- # Calcular fechas
- hoy = datetime.now()
- ayer = hoy - timedelta(days=30)
-
- fecha_inicio = ayer.strftime("%Y-%m-%d")
- fecha_fin = hoy.strftime("%Y-%m-%d")
-
- print(f"Rango de búsqueda: {fecha_inicio} al {fecha_fin}")
-
- outlook_token = obtener_token()
- # se descarga todos los correos en el rango de fecha ingresado
- # tengan o no adjuntos
- emails = descargar_correos(outlook_token, fecha_inicio, fecha_fin)
-
- # se verifica si estos correos ya estan en la base de datos
- # si ya esta registrado, se quita del listado
- # si no esta se inserta en la base de datos
- emails = email_utilities.quitar_correos_ya_registrados_en_base_de_datos(emails)
-
- # descargar adjuntos de cada correo y agregar
- emails = descargar_adjuntos_de_correos(outlook_token, emails, CONFIG.client.ruta_descarga)
-
- # with open("correos_con_adjuntos.json", "w", encoding="utf-8") as f:
- # json.dump(emails, f, ensure_ascii=False, indent=4)
-
-
- # si el correo tiene archivos zip, se descomprimen y se agregan a
- # la lista de archivos adjuntos
- emails = email_utilities.descomprimir_adjuntos(emails, CONFIG.client.ruta_descarga)
-
- # se eliminan mails que no tengan adjunto pdf o xml
- # también se eliminar los adjuntos no que sean xml y pdf
- emails = dejar_mail_si_tiene_adjunto_pdf_o_xml(emails)
-
- # obtener datos de los archivos xml y generar JSON final
- json_final = leer_xml_obtener_datos_validos(emails)
-
- anidar_guias_en_facturas(json_final)
-
-
-
-########################################################
if __name__ == "__main__":
-
- # Configurar Logger (Redirige stdout a archivo y consola)
log_path = logger_utilities.setup_logger()
logger_utilities.limpiar_logs_antiguos(dias=30)
- inicio_ejecucion = datetime.now()
- print("="*60)
- print(f"🚀 INICIO DE EJECUCIÓN: {inicio_ejecucion.strftime('%Y-%m-%d %H:%M:%S')}")
+ inicio = datetime.now()
+ print("=" * 60)
+ print(f"🚀 INICIO DE EJECUCIÓN: {inicio.strftime('%Y-%m-%d %H:%M:%S')}")
print(f"📄 Log guardado en: {log_path}")
- print("="*60)
+ print("=" * 60)
main()
- fin_ejecucion = datetime.now()
- duracion = fin_ejecucion - inicio_ejecucion
- print("\n" + "="*60)
- print(f"🏁 FIN DE EJECUCIÓN: {fin_ejecucion.strftime('%Y-%m-%d %H:%M:%S')}")
- print(f"⏱️ Duración total: {str(duracion).split('.')[0]}") # Sin microsegundos
- print("="*60)
+ fin = datetime.now()
+ duracion = fin - inicio
+ print("\n" + "=" * 60)
+ print(f"🏁 FIN DE EJECUCIÓN: {fin.strftime('%Y-%m-%d %H:%M:%S')}")
+ print(f"⏱️ Duración total: {str(duracion).split('.')[0]}")
+ print("=" * 60)
diff --git a/outlook_utilities.py b/outlook_utilities.py
index d2f90a4..08b8c3e 100644
--- a/outlook_utilities.py
+++ b/outlook_utilities.py
@@ -4,41 +4,52 @@ import warnings
import base64
import os
from datetime import datetime, timedelta
-from config_loader import CONFIG
-
+from dotenv import load_dotenv
warnings.filterwarnings("ignore")
-########################################################
-def get_access_token():
+# ============================================================================
+# CONFIGURACIÓN
+# ============================================================================
+
+# Cargar variables de entorno desde .env
+load_dotenv(dotenv_path=".env.dev")
+
+# Configuración de la API
+CLIENT_ID = os.getenv("CLIENT_ID", "")
+CLIENT_SECRET = os.getenv("CLIENT_SECRET", "")
+TENANT_ID = os.getenv("TENANT_ID", "")
+DOWNLOAD_FOLDER = os.getenv("DOWNLOAD_FOLDER","")
+VERIFY_SSL = os.getenv("VERIFI_SSL", "False").lower() == "true"
+
+# ============================================================================
+def get_access_token() -> str:
"""
Obtiene un token de acceso de Microsoft Graph usando las credenciales
configuradas en config.json (Client ID, Client Secret y Tenant ID).
"""
- if not CONFIG or not CONFIG.outlook:
- raise ValueError("La configuración de Outlook no está cargada correctamente.")
-
- authority = f"https://login.microsoftonline.com/{CONFIG.outlook.tenant_id}"
+ authority = f"https://login.microsoftonline.com/{TENANT_ID}"
app = msal.ConfidentialClientApplication(
- CONFIG.outlook.client_id,
+ CLIENT_ID,
authority=authority,
- client_credential=CONFIG.outlook.client_secret,
- verify=CONFIG.outlook.verify_ssl
+ client_credential=CLIENT_SECRET,
+ verify=VERIFY_SSL
)
scopes = ["https://graph.microsoft.com/.default"]
result = app.acquire_token_for_client(scopes=scopes)
- if "access_token" in result:
- return result["access_token"]
- else:
- error = result.get("error")
- error_description = result.get("error_description")
- correlation_id = result.get("correlation_id")
- raise Exception(f"Error al obtener token: {error} - {error_description} (ID: {correlation_id})")
+ if result:
+ if "access_token" in result:
+ return result["access_token"]
+ else:
+ error = result.get("error")
+ error_description = result.get("error_description")
+ correlation_id = result.get("correlation_id")
+ raise Exception(f"Error al obtener token: {error} - {error_description} (ID: {correlation_id})")
-########################################################
-def get_emails(user_email, token, fecha_inicio, fecha_fin):
+# ============================================================================
+def get_emails(user_email: str, token: str, fecha_inicio: str, fecha_fin: str) -> list[dict]:
"""
Obtiene los correos de una cuenta específica en un rango de fechas.
Retorna una lista de JSON con: id, fecha (ajustada -5h), sender, subject y has_attach.
@@ -62,7 +73,7 @@ def get_emails(user_email, token, fecha_inicio, fecha_fin):
"$top": 50
}
- url = f"https://graph.microsoft.com/v1.0/users/{user_email}/messages"
+ url = f"https://graph.microsoft.com/v1.0/users/{user_email}/mailFolders/Inbox/messages"
all_mails = []
while url:
@@ -70,7 +81,7 @@ def get_emails(user_email, token, fecha_inicio, fecha_fin):
url,
headers=headers,
params=params,
- verify=CONFIG.outlook.verify_ssl
+ verify=VERIFY_SSL
)
if response.status_code != 200:
@@ -100,8 +111,8 @@ def get_emails(user_email, token, fecha_inicio, fecha_fin):
return all_mails
-########################################################
-def download_attachments(user_email, message_id, token, download_path):
+# ============================================================================
+def download_attachments(user_email: str, message_id: str, token: str, download_path: str) -> list[str]:
"""
Descarga los adjuntos de un mensaje específico y los guarda en la ruta indicada.
Retorna una lista con los nombres de los archivos descargados.
@@ -116,7 +127,7 @@ def download_attachments(user_email, message_id, token, download_path):
url = f"https://graph.microsoft.com/v1.0/users/{user_email}/messages/{message_id}/attachments"
- response = requests.get(url, headers=headers, verify=CONFIG.outlook.verify_ssl)
+ response = requests.get(url, headers=headers, verify=VERIFY_SSL)
if response.status_code != 200:
raise Exception(f"Error al obtener adjuntos ({message_id}): {response.status_code} - {response.text}")
@@ -137,4 +148,81 @@ def download_attachments(user_email, message_id, token, download_path):
return downloaded_files
-########################################################
+# ============================================================================
+def send_email(access_token: str, from_email: str, to_recipients: list[str], subject: str, html_body: str, attachment_paths: list[str] | None = None) -> bool:
+ """
+ Envía un correo electrónico con cuerpo HTML y múltiples adjuntos PDF.
+
+ :param access_token: Token de acceso de Microsoft Graph API.
+ :param from_email: Correo electrónico del remitente (cuenta de Office 365).
+ :param to_recipients: Lista de correos electrónicos de los destinatarios.
+ :param subject: Asunto del correo.
+ :param html_body: Contenido del correo en formato HTML.
+ :param attachment_paths: Lista de rutas a los archivos PDF que se adjuntarán.
+ :return: True si se envió correctamente, False en caso contrario.
+ """
+ url = f"https://graph.microsoft.com/v1.0/users/{from_email}/sendMail"
+ headers = {
+ 'Authorization': f'Bearer {access_token}',
+ 'Content-Type': 'application/json'
+ }
+
+ verify_ssl = False #True if os.getenv('OUTLOOK_VERIFY_SSL').lower() == 'true' else False
+
+ # Construir lista de destinatarios para Graph API
+ recipients_payload = []
+ for email in to_recipients:
+ recipients_payload.append({
+ "emailAddress": {
+ "address": email
+ }
+ })
+
+ # Construir lista de adjuntos
+ attachments_payload = []
+ if attachment_paths:
+ for path in attachment_paths:
+ if os.path.exists(path):
+ with open(path, "rb") as f:
+ content_bytes = f.read()
+
+ # Codificar el contenido en Base64
+ encoded_content = base64.b64encode(content_bytes).decode('utf-8')
+ filename = os.path.basename(path)
+
+ attachments_payload.append({
+ "@odata.type": "#microsoft.graph.fileAttachment",
+ "name": filename,
+ "contentType": "application/pdf",
+ "contentBytes": encoded_content
+ })
+ else:
+ print(f"Advertencia: El archivo no existe y no se adjuntará: {path}")
+
+ # Estructura del mensaje para la API
+ email_data = {
+ "message": {
+ "subject": subject,
+ "body": {
+ "contentType": "HTML",
+ "content": html_body
+ },
+ "toRecipients": recipients_payload,
+ "attachments": attachments_payload
+ },
+ "saveToSentItems": "true"
+ }
+
+ try:
+ response = requests.post(url, headers=headers, json=email_data, verify=verify_ssl)
+ if response.status_code == 202:
+ print(f" ├── SUCCESS: Correo enviado exitosamente a: {', '.join(to_recipients)}")
+ return True
+ else:
+ print(f" ├── ERROR: Al enviar correo: {response.status_code} - {response.text}")
+ return False
+ except Exception as e:
+ print(f" ├── ERROR: Excepcion al enviar correo: {e}")
+ return False
+
+# ============================================================================
diff --git a/pdf_utilities.py b/pdf_utilities.py
new file mode 100644
index 0000000..808b23d
--- /dev/null
+++ b/pdf_utilities.py
@@ -0,0 +1,450 @@
+import pdfplumber
+import os
+import re
+
+
+TIPO_MAP = {
+ "FACTURA": "01",
+ "BOLETA": "03",
+ "GUIA DE REMISION": "09",
+ "GUIA DE REMISION TRANSPORTISTA": "31",
+}
+
+
+# ============================================================================
+# HELPERS INTERNOS (desde leer_pdf.py)
+# ============================================================================
+
+def _extract_name_ruc(text_input: str) -> tuple[str | None, str | None]:
+ if not text_input: return None, None
+ full_s = " ".join(text_input.split()).strip()
+ name, ruc = full_s, None
+ if " - " in full_s:
+ name = full_s.split(" - ")[0].strip()
+ r_match = re.search(r"(\d{11})", full_s)
+ if r_match:
+ ruc = r_match.group(1)
+ return name, ruc
+
+# ============================================================================
+
+def _extract_base_data(text: str, pdf_path: str) -> dict:
+ basename = os.path.basename(pdf_path)
+ data = {
+ "archivo": basename,
+ "ruc_emisor": None,
+ "tipo_documento": None,
+ "serie_numero": None,
+ "fecha_emision": None,
+ "ruc_receptor": None,
+ "nombre_receptor": None,
+ "monto_total": None,
+ }
+ ruc_match = re.search(r"R\.U\.C\.?\s*(\d{11})", text, re.IGNORECASE)
+ if ruc_match:
+ data["ruc_emisor"] = ruc_match.group(1)
+ else:
+ if "-" in basename:
+ parts = basename.split('-')
+ if len(parts[0]) == 11 and parts[0].isdigit():
+ data["ruc_emisor"] = parts[0]
+ if not data["ruc_emisor"]:
+ standalone_ruc = re.search(r"(\d{11})", text)
+ if standalone_ruc:
+ data["ruc_emisor"] = standalone_ruc.group(1)
+
+ if "FACTURA ELECTRÓNICA" in text.upper():
+ data["tipo_documento"] = "FACTURA"
+ elif "BOLETA DE VENTA" in text.upper():
+ data["tipo_documento"] = "BOLETA"
+ elif "TRANSPORTISTA" in text.upper() and "GUÍA DE REMISIÓN" in text.upper():
+ data["tipo_documento"] = "GUIA DE REMISION TRANSPORTISTA"
+ elif "GUÍA DE REMISIÓN" in text.upper() or "GRRE:" in text.upper():
+ data["tipo_documento"] = "GUIA DE REMISION"
+
+ serie_numero_match = re.search(r"(?:N°\s*)?([A-Z0-9]{4})\s*-\s*(\d+)", text)
+ if serie_numero_match:
+ data["serie_numero"] = f"{serie_numero_match.group(1)}-{serie_numero_match.group(2)}"
+ elif "GRRE:" in text:
+ grre_match = re.search(r"GRRE:\s*([A-Z0-9-]{4,})", text)
+ if grre_match:
+ data["serie_numero"] = grre_match.group(1)
+
+ date_match = re.search(r"(\d{2})\s*[/-]\s*(\d{2})\s*[/-]\s*(\d{4})", text)
+ if date_match:
+ data["fecha_emision"] = f"{date_match.group(1)}/{date_match.group(2)}/{date_match.group(3)}"
+
+ ruc_label_match = re.search(r"(?:Ruc|REGISTRO ÚNICO DE CONTRIBUYENTES N°)\s*:?\s*(\d{11})", text, re.IGNORECASE)
+ if ruc_label_match and ruc_label_match.group(1) != data["ruc_emisor"]:
+ data["ruc_receptor"] = ruc_label_match.group(1)
+ else:
+ all_rucs = re.findall(r"(? 2:
+ data["nombre_receptor"] = name
+ break
+ return data
+
+# ============================================================================
+
+def _extract_factura_details(text: str, data: dict) -> None:
+ amount_regex = r"((?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2})"
+ total_patterns = [
+ rf"TOTAL\s*(?:S/|S/\.)?[:\s]*{amount_regex}",
+ rf"IMPORTE TOTAL\s*(?:S/|S/\.)?[:\s]*{amount_regex}",
+ rf"NETO A PAGAR\s*(?:S/|S/\.)?[:\s]*{amount_regex}",
+ ]
+ total_found = False
+ for pattern in total_patterns:
+ match = re.search(pattern, text, re.IGNORECASE)
+ if match:
+ data["monto_total"] = match.group(1).replace(",", "")
+ total_found = True
+ break
+ if not total_found:
+ amounts = re.findall(amount_regex, text)
+ if amounts:
+ data["monto_total"] = amounts[-1].replace(",", "")
+
+ data.update({
+ "items": [],
+ "detraccion": {"tiene": False, "porcentaje": None, "monto": None},
+ })
+
+ lines = text.split('\n')
+ table_start_index = -1
+ for i, line in enumerate(lines):
+ line_up = line.upper()
+ headers_found = sum(1 for h in ["CANT", "DESCRIP", "PRECIO", "TOTAL", "UNIT", "VALOR", "CÓDIGO", "ITEM"] if h in line_up)
+ if headers_found >= 3:
+ table_start_index = i
+ break
+
+ if table_start_index != -1:
+ items_lines = lines[table_start_index + 1:]
+ current_item = None
+ for line in items_lines:
+ line = line.strip()
+ if not line: continue
+ if re.search(r"^(?:SUB|TOTAL|IGV|SON:|OP\.|RES|CUO|DET|BANCO|Representación|Podrá|Autorizado|FORMA|CUOTA|GRAVADO|I\.G\.V\.|DESEMBOLSO|OBSERVACION)", line, re.IGNORECASE):
+ break
+ amounts = re.findall(r"(?= 2:
+ if current_item: data["items"].append(current_item)
+ qty = amounts[0]
+ total = amounts[-1]
+ p_unit = amounts[-2]
+ if len(amounts) >= 4:
+ p_unit = amounts[1]
+ desc_line = line
+ for amt in [qty, p_unit, total]:
+ desc_line = desc_line.replace(amt, "", 1)
+ desc_parts = [p.strip() for p in desc_line.split() if p.strip()]
+ codigo = None
+ for i in range(min(3, len(desc_parts))):
+ if re.search(r"[A-Z].*\d|\d.*[A-Z]|[A-Z]{2,}-\d+", desc_parts[i]):
+ codigo = desc_parts[i]
+ desc_parts.pop(i)
+ break
+ if desc_parts and desc_parts[0].isdigit() and len(desc_parts[0]) <= 2:
+ desc_parts.pop(0)
+ descripcion = " ".join(desc_parts)
+ current_item = {
+ "codigo": codigo,
+ "cantidad": qty,
+ "descripcion": descripcion,
+ "precio_unitario": p_unit.replace(",", ""),
+ "total": total.replace(",", ""),
+ "guias_remision": [],
+ "guia_transportista": None,
+ "bolsas": None,
+ "peso": None,
+ "unidad_peso": None,
+ }
+ elif current_item and len(line) > 5:
+ current_item["descripcion"] += " " + line
+ if current_item: data["items"].append(current_item)
+
+ for item in data["items"]:
+ desc = item["descripcion"]
+ guias_remision_matches = re.findall(r"\b(?:GRR\w*|GR\b|G\.R\.R\.?|GUIA REMISI.N)\s*[:/\s]*((?:[TVE][A-Z0-9]{3})-\d+)", desc, re.IGNORECASE)
+ item["guias_remision"] = [m.strip() for m in guias_remision_matches] if guias_remision_matches else []
+ guia_transportista_match = re.search(r"\b(?:GRT\w*|GR\b|G\.R\.T\.?|GUIA TRANSPORTISTA)\s*[:/\s]*((?:[TVE][A-Z0-9]{3})-\d+)", desc, re.IGNORECASE)
+ if guia_transportista_match:
+ item["guia_transportista"] = guia_transportista_match.group(1).strip()
+ bolsas_match = re.search(r"(\d+)\s*(?:BOLSAS|BLS|BOLS|SACOS)\b", desc, re.IGNORECASE)
+ if bolsas_match:
+ item["bolsas"] = bolsas_match.group(1)
+ peso_match = re.search(r"(?:PESO(?: TOTAL)?|CARGA)\s*(?:(TN|KG|KILOS?))?\.?\s*[:\s]\s*([\d,.]+)\s*(TN|KG|KILOS?)?", desc, re.IGNORECASE)
+ if peso_match:
+ item["peso"] = peso_match.group(2).strip()
+ item["unidad_peso"] = (peso_match.group(1) or peso_match.group(3) or "").upper()
+
+ det = data["detraccion"]
+ if re.search(r"\bDETRACC(?:I[OÓ]N|\.)", text, re.IGNORECASE):
+ det["tiene"] = True
+ pct_match = re.search(r"PORCENTAJE\s*[:/]\s*([\d.]+)\s*%", text, re.IGNORECASE)
+ if not pct_match:
+ pct_match = re.search(r"DETRACC[^\n]*?([\d]+(?:\.\d+)?)\s*%", text, re.IGNORECASE)
+ if pct_match:
+ det["porcentaje"] = pct_match.group(1)
+ monto_match = re.search(
+ r"(?:MONTO\s*(?:DETRACC\.?|DE\s*DETRACCI[OÓ]N)|Monto\s*Detracc\.?)\s*[:/]?\s*S/\.?\s*([\d,.]+)",
+ text, re.IGNORECASE,
+ )
+ if not monto_match:
+ monto_match = re.search(r"MONTO\s*DETRACC\w*[:\s]+S?/?\.?\s*([\d,.]+)", text, re.IGNORECASE)
+ if not monto_match:
+ monto_match = re.search(r"DETRACCION\s+\d+%\s+([\d.]+)", text, re.IGNORECASE)
+ if monto_match:
+ raw = monto_match.group(1).replace(",", "")
+ try:
+ det["monto"] = f"{float(raw):.2f}"
+ except ValueError:
+ det["monto"] = raw
+
+# ============================================================================
+
+def _extract_guia_remision_details(text: str, data: dict) -> None:
+ data.update({
+ "peso_total": None, "punto_partida": None, "punto_llegada": None, "placa": None,
+ "marca": None, "chofer": None, "brevete": None, "configuracion_vehicular": None, "items": [],
+ })
+ peso_match = re.search(r"Peso Total (?:TN|KG)?:\s*([\d,.]+)", text, re.IGNORECASE)
+ if not peso_match:
+ peso_match = re.search(r"Peso Total:\s*([\d,.]+)", text, re.IGNORECASE)
+ if peso_match:
+ data["peso_total"] = peso_match.group(1)
+ conf = re.search(r"(?:Configuración vehicular|C\. Vehc\.?):\s*([A-Z0-9/]+)", text, re.IGNORECASE)
+ if conf: data["configuracion_vehicular"] = conf.group(1).strip()
+ for field, pattern in {
+ "placa": r"Placa:\s*([A-Z0-9\s/]+?)(?=\s*(?:C\.|Marca|Peso|$))",
+ "marca": r"Marca:\s*([A-Z0-9\s-]+?)(?=\s*(?:Placa|Chofer|Peso|$))",
+ "chofer": r"Chofer:\s*([A-Z\s.-]+?)(?=\s*(?:Brevete|Placa|Peso|$))",
+ "brevete": r"Brevete:\s*([A-Z0-9]+)",
+ }.items():
+ match = re.search(pattern, text, re.IGNORECASE)
+ if match: data[field] = match.group(1).strip()
+ p_match = re.search(r"Punto de Partida:\s*(.*?)\s*Punto de Llegada:", text, re.DOTALL | re.IGNORECASE)
+ if p_match: data["punto_partida"] = " ".join(p_match.group(1).split()).strip()
+ l_match = re.search(r"Punto de Llegada:\s*(.*?)\s*(?:Fecha de inicio|T[0-9]{3}-|$)", text, re.DOTALL | re.IGNORECASE)
+ if l_match: data["punto_llegada"] = " ".join(l_match.group(1).split()).strip()
+ header = re.search(r"(?:CANT|CANTIDAD)\s+(?:UN|UNIDAD)\s+(?:DESCRIPCIÓN|DESCRIPCION|DETALLE)", text, re.IGNORECASE)
+ if header:
+ items_text = text[header.end():].split("Peso Aprox")[0]
+ current_item = None
+ for line in items_text.split('\n'):
+ line = line.strip()
+ if not line or re.search(r"^(?:Peso Aprox|Punto de|Fecha de|Representación|Placa|Chofer|C\. Vehc)", line, re.IGNORECASE):
+ continue
+ match = re.match(r"^(\d+(?:\.\d+)?)\s+([A-Z]{2,3})\s+(.+)", line)
+ if match:
+ if current_item: data["items"].append(current_item)
+ current_item = {"cantidad": match.group(1), "unidad": match.group(2), "descripcion": match.group(3).strip()}
+ elif current_item:
+ current_item["descripcion"] += " " + line
+ if current_item: data["items"].append(current_item)
+
+# ============================================================================
+
+def _extract_guia_transportista_details(text: str, data: dict) -> None:
+ data.update({
+ "punto_partida": None, "punto_llegada": None, "nombre_remitente": None, "ruc_remitente": None,
+ "nombre_destinatario": None, "ruc_destinatario": None, "documentos_relacionados": [],
+ "unidad_medida_peso_bruto": None, "peso_bruto_total_carga": None, "indicador_pagador_flete": None,
+ "nombre_pagador": None, "ruc_pagador": None, "datos_vehiculos": [], "datos_conductores": [], "observaciones": None,
+ })
+ p_match = re.search(r"Punto de Partida:\s*(.*?)\s*Punto de llegada:", text, re.DOTALL | re.IGNORECASE)
+ if p_match: data["punto_partida"] = " ".join(p_match.group(1).split()).strip()
+ l_match = re.search(r"Punto de llegada:\s*(.*?)\s*Datos del remitente:", text, re.DOTALL | re.IGNORECASE)
+ if l_match: data["punto_llegada"] = " ".join(l_match.group(1).split()).strip()
+ rem_match = re.search(r"Datos del remitente:\s*(.*?)\s*Datos del destinatario:", text, re.DOTALL | re.IGNORECASE)
+ if rem_match: data["nombre_remitente"], data["ruc_remitente"] = _extract_name_ruc(rem_match.group(1))
+ dest_match = re.search(r"Datos del destinatario:\s*(.*?)\s*Documentos Relacionados:", text, re.DOTALL | re.IGNORECASE)
+ if dest_match: data["nombre_destinatario"], data["ruc_destinatario"] = _extract_name_ruc(dest_match.group(1))
+ pag_match = re.search(r"Datos de pagador de flete:\s*(.*?)\s*(?:Observaciones|Esta es|$)", text, re.DOTALL | re.IGNORECASE)
+ if pag_match: data["nombre_pagador"], data["ruc_pagador"] = _extract_name_ruc(pag_match.group(1))
+ rel_match = re.search(r"Documentos Relacionados:\s*(.*?)\s*(?:Bienes por Transportar|Datos de|$)", text, re.DOTALL | re.IGNORECASE)
+ if rel_match:
+ for line in rel_match.group(1).split('\n'):
+ line = line.strip()
+ if not line: continue
+ doc = re.search(r"(?:N°\s*)([A-Z0-9\s-]+)- RUC N°\s*(\d{11})", line, re.IGNORECASE)
+ data["documentos_relacionados"].append({"documento": doc.group(1).strip(), "ruc_relacionado": doc.group(2)} if doc else line)
+ u_match = re.search(r"Unidad de Medida del Peso Bruto:\s*(\w+)", text, re.IGNORECASE)
+ if u_match: data["unidad_medida_peso_bruto"] = u_match.group(1)
+ p_match = re.search(r"Peso Bruto total de la carga:\s*([\d,.]+)", text, re.IGNORECASE)
+ if p_match: data["peso_bruto_total_carga"] = p_match.group(1)
+ ind_match = re.search(r"Indicador del pagador del flete:\s*(\w+)", text, re.IGNORECASE)
+ if ind_match: data["indicador_pagador_flete"] = ind_match.group(1)
+ v_match = re.search(r"Datos de los vehículos:\s*(.*?)\s*Datos de los conductores:", text, re.DOTALL | re.IGNORECASE)
+ if v_match: data["datos_vehiculos"] = [{"placa": p} for p in re.findall(r"Número de placa:\s*([A-Z0-9]+)", v_match.group(1), re.IGNORECASE)]
+ c_match = re.search(r"Datos de los conductores:\s*(.*?)\s*(?:Datos de pagador|Observaciones|$)", text, re.DOTALL | re.IGNORECASE)
+ if c_match:
+ c_text = c_match.group(1)
+ data["datos_conductores"] = [{"nombre": n.strip()} for n in re.findall(r"Principal:\s*([A-Z\s-]+)(?:\s*-|$)", c_text, re.IGNORECASE)]
+ for i, b in enumerate(re.findall(r"Número de lincencia de conducir:\s*([A-Z0-9]+)", c_text, re.IGNORECASE)):
+ if i < len(data["datos_conductores"]): data["datos_conductores"][i]["brevete"] = b
+ obs = re.search(r"Observaciones\s*:\s*(.*?)(?=\s*(?:Esta es|$))", text, re.IGNORECASE | re.DOTALL)
+ if obs: data["observaciones"] = obs.group(1).strip()
+
+# ============================================================================
+
+def _extract_detraccion_spatial(pdf, data: dict) -> None:
+ det = data.get("detraccion", {})
+ if not det.get("tiene"):
+ return
+ if det.get("porcentaje") and det.get("monto"):
+ return
+ for page in pdf.pages:
+ words = page.extract_words()
+ if not words:
+ continue
+ det_words = [w for w in words if re.search(r"Detrac", w["text"], re.IGNORECASE)]
+ if not det_words:
+ continue
+ region_text = None
+ for rect in (page.rects or []):
+ for dw in det_words:
+ if (rect["x0"] <= dw["x0"] <= rect["x1"] and
+ rect["top"] <= dw["top"] <= rect["bottom"]):
+ cropped = page.crop((rect["x0"], rect["top"], rect["x1"], rect["bottom"]))
+ region_text = cropped.extract_text() or ""
+ break
+ if region_text:
+ break
+ if not region_text and det_words:
+ dw = det_words[0]
+ cropped = page.crop((0, max(0, dw["top"] - 30), page.width, min(page.height, dw["bottom"] + 80)))
+ region_text = cropped.extract_text() or ""
+ if not region_text:
+ continue
+ if not det.get("porcentaje"):
+ pct = re.search(r"(\d+(?:\.\d+)?)\s*%\s*Detrac", region_text, re.IGNORECASE)
+ if not pct:
+ pct = re.search(r"PORCENTAJE\s*[:/]\s*(\d+(?:\.\d+)?)\s*%", region_text, re.IGNORECASE)
+ if pct:
+ try:
+ det["porcentaje"] = f"{float(pct.group(1)):.2f}"
+ except ValueError:
+ det["porcentaje"] = pct.group(1)
+ if not det.get("monto"):
+ monto = re.search(r"Detrac[^:]*:\s*([\d,.]+)", region_text, re.IGNORECASE)
+ if not monto:
+ monto = re.search(r"MONTO[^:]*:\s*S?/?\.?\s*([\d,.]+)", region_text, re.IGNORECASE)
+ if monto:
+ raw = monto.group(1).replace(",", "")
+ try:
+ det["monto"] = f"{float(raw):.2f}"
+ except ValueError:
+ det["monto"] = raw
+ break
+
+
+# ============================================================================
+# FUNCIONES PÚBLICAS
+# ============================================================================
+
+def extraer_datos_pdf(pdf_path: str) -> dict:
+ try:
+ with pdfplumber.open(pdf_path) as pdf:
+ text = "".join(page.extract_text() or "" for page in pdf.pages)
+ data = _extract_base_data(text, pdf_path)
+ if data["tipo_documento"] == "FACTURA":
+ _extract_factura_details(text, data)
+ _extract_detraccion_spatial(pdf, data)
+ elif data["tipo_documento"] == "GUIA DE REMISION":
+ _extract_guia_remision_details(text, data)
+ elif data["tipo_documento"] == "GUIA DE REMISION TRANSPORTISTA":
+ _extract_guia_transportista_details(text, data)
+ return data
+ except Exception as e:
+ return {"archivo": os.path.basename(pdf_path), "error": str(e)}
+
+# ============================================================================
+
+def generar_nombre_pdf(data: dict) -> str | None:
+ ruc = data.get("ruc_emisor")
+ tipo_doc = data.get("tipo_documento")
+ serie_numero = data.get("serie_numero")
+ if not ruc or not tipo_doc or not serie_numero:
+ return None
+ abrev = TIPO_MAP.get(tipo_doc.upper())
+ if not abrev:
+ return None
+ if "-" in serie_numero:
+ serie, numero = serie_numero.split("-", 1)
+ numero = numero.lstrip("0") or "0"
+ else:
+ serie = serie_numero
+ numero = ""
+ return f"{ruc}-{abrev}-{serie}-{numero}.pdf"
+
+# ============================================================================
+
+def renombrar_pdf(pdf_path: str, data: dict) -> str:
+ dirname = os.path.dirname(pdf_path)
+ nuevo_nombre = generar_nombre_pdf(data)
+ if not nuevo_nombre:
+ return pdf_path
+ nuevo_path = os.path.join(dirname, nuevo_nombre)
+ if pdf_path == nuevo_path:
+ return pdf_path
+ try:
+ os.rename(pdf_path, nuevo_path)
+ return nuevo_path
+ except OSError:
+ return pdf_path
+
+# ============================================================================
+
+def _extraer_serie_numero(texto: str) -> tuple[str, str] | None:
+ m = re.search(r'\b([TVE][A-Z0-9]{3})\s*-\s*(\d+)\b', texto)
+ if m:
+ return m.group(1).upper(), m.group(2)
+ return None
+
+
+def extraer_referencias(items: list) -> list:
+ referencias = []
+ vistos = set()
+ for item in items:
+ for gr_str in item.get("guias_remision", []):
+ parsed = _extraer_serie_numero(gr_str)
+ if not parsed:
+ partes = gr_str.split("-", 1)
+ if len(partes) == 2:
+ parsed = (partes[0].strip(), partes[1].strip())
+ else:
+ continue
+ serie, numero = parsed
+ clave = f"09|{serie}|{numero}"
+ if clave not in vistos:
+ vistos.add(clave)
+ referencias.append({"tipo": "09", "serie": serie, "numero": numero})
+ gt_str = item.get("guia_transportista")
+ if gt_str:
+ parsed = _extraer_serie_numero(gt_str)
+ if not parsed:
+ partes = gt_str.split("-", 1)
+ if len(partes) == 2:
+ parsed = (partes[0].strip(), partes[1].strip())
+ else:
+ continue
+ serie, numero = parsed
+ clave = f"31|{serie}|{numero}"
+ if clave not in vistos:
+ vistos.add(clave)
+ referencias.append({"tipo": "31", "serie": serie, "numero": numero})
+ return referencias
diff --git a/pipeline.py b/pipeline.py
new file mode 100644
index 0000000..1725b33
--- /dev/null
+++ b/pipeline.py
@@ -0,0 +1,925 @@
+"""
+Pipeline de procesamiento de facturas de transporte.
+
+Cada correo se procesa individualmente en un bucle:
+ BD check → descargar adjuntos → descomprimir ZIPs → filtrar PDF/XML
+ → procesar PDFs (lookup) → procesar XMLs (cruzar con PDF)
+ → guardar resultado → construir draft SAP → validar precios
+ → crear draft en SAP (si válido) → notificar → guardar draft JSON
+"""
+
+import os
+import re
+import json
+import math
+from datetime import datetime, timedelta
+
+import email_utilities
+import outlook_utilities
+import database_utilities
+import sunat_schema
+import sapb1_manager
+import pdf_utilities
+
+CUADRO_VALOR_REFERENCIAL = {
+ "CHINCHA": 89.09,
+ "PISCO": 101.7,
+ "CAÑETE": 79.81,
+ "TARMA": 106.54,
+ "SAN CLEMENTE": 94.27,
+}
+
+
+# ============================================================================
+# HELPERS
+# ============================================================================
+
+def _ruta_carpeta(subcarpeta: str) -> str:
+ return os.path.join(os.path.dirname(__file__), subcarpeta)
+
+
+# ============================================================================
+# PASO 1: Autenticación Microsoft Graph
+# ============================================================================
+def step_authenticate() -> str | None:
+ try:
+ print("Intentando obtener token de acceso...")
+ token = outlook_utilities.get_access_token()
+ if token:
+ print("✅ Token obtenido exitosamente.")
+ print(f"Token prefix: {token[:25]}...")
+ return token
+ return None
+ except Exception as e:
+ print(f"❌ Error: {e}")
+ return None
+
+
+# ============================================================================
+# PASO 2: Descarga de correos
+# ============================================================================
+def step_fetch_emails(token: str, email: str, fecha_inicio: str, fecha_fin: str) -> list[dict]:
+ print("\n" + "=" * 60)
+ print(f"[MAIL] {email}")
+ try:
+ emails = outlook_utilities.get_emails(email, token, fecha_inicio, fecha_fin)
+ print(f"Se encontraron {len(emails)} correos.")
+ return emails
+ except Exception as e:
+ print(f"❌ Error: {e}")
+ return []
+
+
+# ============================================================================
+# PROCESAR UN CORREO
+# ============================================================================
+def _procesar_un_correo(email: dict, token: str, email_addr: str, download_folder: str, output_folder: str) -> None:
+ """
+ Procesa un correo completo de principio a fin.
+
+ Flujo (pasos 1-14):
+ 1-2. Verifica en BD si el correo ya fue procesado; si no, lo guarda.
+ 3. Descarga adjuntos desde Microsoft Graph.
+ 4. Descomprime archivos ZIP.
+ 5. Filtra solo archivos PDF/XML.
+ 6. Procesa PDFs (extrae datos, renombra, construye cache).
+ 7. Procesa XMLs (cruza con PDF, asigna guías de remisión).
+ 8. Guarda resultado JSON con datos extraídos.
+ 9. Filtra solo facturas tipo 01 para SAP.
+ 10. Valida precios contra DeliveryNotes en SAP.
+ 11. Construye payload del draft SAP B1.
+ 12. Guarda draft JSON para depuración.
+ 13. Crea el draft en SAP (si los precios coinciden) o notifica error.
+ 14. Envía email de notificación con el resultado.
+
+ Args:
+ email: Diccionario del correo (debe tener ``id``, ``fecha``,
+ ``subject``).
+ token: Token de acceso Microsoft Graph.
+ email_addr: Dirección de correo de la bandeja de entrada.
+ download_folder: Carpeta relativa para descargar adjuntos.
+ output_folder: Carpeta relativa para guardar resultados.
+
+ Returns:
+ None. Los resultados se persisten a disco (JSON) y se envían
+ notificaciones por email.
+ """
+
+ def _descargar_adjuntos() -> list[str] | None:
+ """
+ Descarga todos los adjuntos del correo usando Microsoft Graph.
+
+ Almacena la lista de nombres de archivo en ``email["adjuntos"]``.
+
+ Returns:
+ Lista de nombres de archivo descargados, o None si no hay adjuntos
+ o hubo un error.
+
+ Example:
+ Si el correo tiene ``factura.xml`` y ``guia.pdf``::
+
+ adjuntos = _descargar_adjuntos()
+ # → ["factura.xml", "guia.pdf"]
+ # email["adjuntos"] = ["factura.xml", "guia.pdf"]
+ """
+ try:
+ adjuntos = outlook_utilities.download_attachments(
+ email_addr, email["id"], token, download_folder
+ )
+ if adjuntos:
+ email["adjuntos"] = adjuntos
+ return adjuntos
+ return None
+ except Exception as ex:
+ print(f" ❌ Error descargando adjuntos: {ex}")
+ return None
+
+ def _descomprimir() -> None:
+ """
+ Descomprime archivos ZIP entre los adjuntos del correo.
+
+ Los archivos extraídos se colocan en la misma carpeta de descarga.
+ Los ZIP originales no se eliminan automáticamente.
+
+ Example:
+ Si hay ``adjuntos.zip`` con ``factura.pdf`` y ``guia.xml``::
+
+ _descomprimir()
+ # → Aparecen factura.pdf y guia.xml en download_folder
+ """
+ email_utilities.descomprimir_adjuntos([email], download_folder)
+
+ def _filtrar_y_limpiar() -> bool:
+ """
+ Filtra los adjuntos conservando solo archivos PDF y XML.
+
+ - Los archivos que no son PDF ni XML se eliminan del disco.
+ - Si no queda ningún PDF/XML válido, se eliminan todos los archivos
+ descargados y se retorna False.
+ - Actualiza ``email["adjuntos"]`` con la lista filtrada.
+
+ Returns:
+ True si hay al menos un PDF o XML, False si no queda ninguno.
+
+ Example:
+ Adjuntos originales: ``[img.png, factura.xml, guia.pdf]``::
+
+ ok = _filtrar_y_limpiar()
+ # → img.png eliminado
+ # → ok = True
+ # → email["adjuntos"] = ["factura.xml", "guia.pdf"]
+ """
+ todos = email.get("adjuntos", [])
+ validos = [a for a in todos if a.lower().endswith((".pdf", ".xml"))]
+ no_validos = [a for a in todos if a not in validos]
+
+ carpeta = _ruta_carpeta(download_folder)
+
+ if not validos:
+ for a in todos:
+ ruta = os.path.join(carpeta, a)
+ if os.path.exists(ruta):
+ os.remove(ruta)
+ print(f" ├── 🗑️ Correo sin PDF/XML válidos, adjuntos eliminados")
+ return False
+
+ for a in no_validos:
+ ruta = os.path.join(carpeta, a)
+ if os.path.exists(ruta):
+ os.remove(ruta)
+
+ email["adjuntos"] = validos
+ return True
+
+ def _procesar_pdfs() -> 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()
+ # → {"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
+ print(f" │ 📄 PDF procesado: {canonical} ({data.get('tipo_documento', '?')})")
+ else:
+ print(f" │ ⚠️ PDF no procesado: {adjunto} - {data.get('error', 'error desconocido')}")
+
+ if datos_pdf:
+ email["datos_pdf"] = datos_pdf
+
+ return pdf_cache
+
+ def _procesar_xmls(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.
+
+ 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(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 = []
+
+ for idx, a in enumerate(email["adjuntos"]):
+ prefijo = " │ └── " if idx == len(email["adjuntos"]) - 1 else " │ ├── "
+ print(f"{prefijo}{a}")
+
+ if not a.lower().endswith(".xml"):
+ continue
+
+ xml_path = os.path.join(carpeta, a)
+ print(f" │ 📄 Procesando XML: {a}")
+ data = sunat_schema.extraer_datos_xml(xml_path)
+
+ if not data:
+ print(f" │ ℹ️ XML sin datos (no es factura tipo 01 o guía 31)")
+ continue
+
+ data["ruta_xml"] = xml_path
+
+ # ruta_pdf: construir nombre canónico {ruc}-{tipo}-{serie}-{numero}.pdf
+ # El número ya viene con lstrip("0") desde _split_serie_numero en sunat_schema
+ # y generar_nombre_pdf ahora también aplica lstrip("0") para que coincidan
+ 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
+
+ # 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
+ print(f" │ 🔄 documentos_referencia completados desde PDF ({len(refs)} refs)")
+ 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
+ print(f" │ 🔄 documentos_referencia completados desde PDF (fallback, {len(refs)} refs)")
+ break
+
+ # Asignar guias_remision[] y guia_transportista a cada línea
+ # Primero: todas las refs tipo 09 y 31 desde 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"]
+ # Segundo: si hay datos del PDF, usar agrupación por ítem (respeta multi-guía)
+ pdf_items = (pdf_cache.get(pdf_canonical) or {}).get("items", [])
+ for i, linea in enumerate(data.get("lineas", [])):
+ if i < len(pdf_items):
+ pi = pdf_items[i]
+ linea["guias_remision"] = []
+ for g_str in pi.get("guias_remision", []):
+ m = re.search(r"\b([TVE][A-Z0-9]{3})\s*-\s*(\d+)\b", g_str)
+ if m:
+ linea["guias_remision"].append({
+ "tipo": "09", "serie": m.group(1).upper(), "numero": m.group(2),
+ })
+ else:
+ partes = g_str.split("-", 1)
+ if len(partes) == 2:
+ linea["guias_remision"].append({
+ "tipo": "09", "serie": partes[0].strip(), "numero": partes[1].strip(),
+ })
+ guia_transportista = pi.get("guia_transportista")
+ if guia_transportista:
+ m = re.search(r"\b([TVE][A-Z0-9]{3})\s*-\s*(\d+)\b", guia_transportista)
+ if m:
+ pi["guia_transportista_serie"] = m.group(1)
+ pi["guia_transportista_numero"] = m.group(2)
+ elif "-" in guia_transportista:
+ partes_list = guia_transportista.split("-", 1)
+ linea["guia_transportista"] = {"tipo": "31", "serie": partes_list[0].strip(), "numero": partes_list[1].strip()}
+ else:
+ linea["guia_transportista"] = None
+ else:
+ linea["guia_transportista"] = None
+ elif i < len(refs_09):
+ r = refs_09[i]
+ linea["guias_remision"] = [{"tipo": "09", "serie": r["serie"].strip(), "numero": r["numero"].strip()}]
+ linea["guia_transportista"] = refs_31[i] if i < len(refs_31) else None
+ else:
+ linea["guias_remision"] = []
+ linea["guia_transportista"] = None
+
+ print(f" │ ✅ Datos extraídos: {data.get('serie')}-{data.get('numero')} ({data.get('tipo_descripcion', '?')})")
+ documentos.append(data)
+
+ return documentos
+
+ def _guardar_resultado(documentos: list[dict]) -> None:
+ """
+ Guarda la lista de documentos extraídos en un archivo JSON.
+
+ El nombre del archivo sigue el patrón
+ ``resultado_{serie}-{numero}_{timestamp}.json`` y se escribe en
+ ``output_folder``.
+
+ Args:
+ documentos: Lista de diccionarios con datos extraídos de XML
+ (producida por ``_procesar_xmls``).
+
+ Example:
+ Si ``documentos[0]`` tiene ``serie="F002", numero="486"``::
+
+ _guardar_resultado(documentos)
+ # → Crea resultado_F002-486_2026-07-08_12-00-00.json
+ """
+ if not documentos:
+ return
+ if not os.path.exists(output_folder):
+ os.makedirs(output_folder, exist_ok=True)
+ ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
+ serie = documentos[0].get("serie", "unknown")
+ numero = documentos[0].get("numero", "unknown")
+ filename = f"resultado_{serie}-{numero}_{ts}.json"
+ filepath = os.path.join(output_folder, filename)
+ with open(filepath, "w", encoding="utf-8") as f:
+ json.dump(documentos, f, ensure_ascii=False, indent=4)
+ print(f" 💾 Resultado guardado: {filename}")
+
+ def _build_sap_draft(factura: dict, resultados_val: list[dict]) -> dict:
+ """
+ Construye el payload JSON para crear un Draft en SAP B1 Service Layer
+ (endpoint ``POST /Drafts``).
+
+ Flujo:
+ 1. Toma los datos de cabecera de la factura (RUCS, fechas, totales).
+ 2. Determina el ``ItemCode`` según la fecha de emisión:
+ ``C103002-A`` si es de un mes anterior, ``C103002`` si es del
+ mes actual.
+ 3. Prepara las cuotas de detracción (``DocumentInstallments``):
+ si la factura tiene detracción, genera dos cuotas (detracción y
+ neto); si no, una sola cuota por el total.
+ 4. Por cada línea de factura, itera sobre las guías validadas
+ (``guias_lt[]`` de ``resultados_val``) y genera una línea de
+ documento (``DocumentLine``) por cada guía con:
+ - ``ItemCode``, ``Quantity``, ``PriceAfterVAT``,
+ - ``WarehouseCode``, ``U_IDC_GRFLETE`` y campos fijos.
+ Cuando la validación es ``ok=True`` usa los valores de SAP
+ (``draft_quantity``, ``draft_price``); cuando no, usa los del XML.
+ 5. Calcula ``U_IDC_VALREF`` como:
+ ``total_weight × CUADRO_VALOR_REFERENCIAL[zona]``
+ donde ``total_weight`` es la suma de ``Weight1`` de todos los
+ DeliveryNotes de la guía, y ``zona`` se obtiene del campo ``U_IDC_DESTINO`` del primer
+ DeliveryNote de la primera guía, mapeado contra el maestro de
+ zonas SAP (``U_IDC_ZONAS``).
+
+ Args:
+ factura: Diccionario con los datos de la factura extraídos del
+ XML (claves: ``serie``, ``numero``, ``tipo``, ``lineas[]``,
+ ``guias_remision[]``, ``tiene_detraccion``, etc.).
+ resultados_val: Lista de resultados de ``_validate_prices``,
+ uno por línea de factura. Cada elemento contiene
+ ``guias_lt[]``, ``ok``, ``use_sap_prices``, etc.
+
+ Returns:
+ Dict con la estructura completa del draft SAP B1 lista para
+ enviar a ``POST /Drafts``.
+
+ Example:
+ Factura F002-486 con guía T001-127773 y SAP devuelve
+ ``FL10007 qty=350 pav=1.3396``::
+
+ draft = _build_sap_draft(factura, resultados_val)
+ # → {
+ # "DocType": "dDocument_Items",
+ # "CardCode": "PL20537929520",
+ # "DocumentLines": [{
+ # "ItemCode": "C103002",
+ # "Quantity": 350,
+ # "PriceAfterVAT": 1.3396,
+ # "U_IDC_GRFLETE": "09T001-127773",
+ # ...
+ # }],
+ # "DocumentInstallments": [...],
+ # "U_IDC_VALREF": 31181.5,
+ # ...
+ # }
+ """
+ serie = factura.get("serie", "")
+ numero = factura.get("numero", "")
+ tipo = factura.get("tipo", "01")
+ total = factura.get("total_factura", 0) or 0
+ tiene_detraccion = factura.get("tiene_detraccion", False)
+ codigo_detraccion = factura.get("codigo_detraccion", "")
+ porcentaje_detraccion = factura.get("porcentaje_detraccion", 0) or 0
+ monto_detraccion = factura.get("monto_detraccion", 0) or 0
+ fecha_emision = factura.get("fecha_emision", "")
+ ruc_emisor = factura.get("ruc_emisor", "")
+ razon_social_emisor = factura.get("razon_social_emisor", "")
+
+ fecha_contabilizacion = datetime.now().strftime("%Y-%m-%d")
+ emision = datetime.strptime(fecha_emision, "%Y-%m-%d")
+ item_code = "C103002-A" if (emision.year, emision.month) < (datetime.now().year, datetime.now().month) else "C103002"
+
+ neto = round(total - monto_detraccion, 2)
+ if tiene_detraccion:
+ installments = [
+ {"DueDate": fecha_emision, "Total": round(monto_detraccion, 2), "InstallmentId": 1, "U_VS_TIPOCUOTA": "D", "U_VS_TIPOSERVICIO": f"{codigo_detraccion}02"},
+ {"DueDate": fecha_emision, "Total": neto, "InstallmentId": 2, "U_VS_TIPOCUOTA": "N", "U_VS_TIPOSERVICIO": ""},
+ ]
+ else:
+ installments = [
+ {"DueDate": fecha_emision, "Total": round(total, 2), "InstallmentId": 1, "U_VS_TIPOCUOTA": "N", "U_VS_TIPOSERVICIO": ""},
+ ]
+
+ doc_lines = []
+ total_weight = 0
+ zona_guia = ""
+ guia_referencia = ""
+
+ for i, linea in enumerate(factura.get("lineas", [])):
+ resultado_validacion = resultados_val[i] if i < len(resultados_val) else None
+ guias_lt = (resultado_validacion or {}).get("guias_lt", [])
+ if not guias_lt:
+ continue
+
+
+ zona_guia = guias_lt[0].get("zona","")
+
+ guia_referencia = f"{guias_lt[0]['guia']['tipo']}{guias_lt[0]['guia']['serie']}-{guias_lt[0]['guia']['numero']}"
+
+ for item in guias_lt:
+ codigo_guia = f"{item['guia']['tipo']}{item['guia']['serie']}-{item['guia']['numero']}"
+ lt_val = math.trunc(item["draft_price"] * 1_000_000) / 1_000_000
+ qty = item.get("draft_quantity", int(item.get("quantity", 1)))
+
+ total_weight += item.get("weight1", 0)
+
+ doc_lines.append({
+ "ItemCode": item_code,
+ #"ItemDescription": linea.get("descripcion", "TRANSPORTE DE CARGA MERCADERIA")[:200],
+ "Quantity": qty,
+ "PriceAfterVAT": lt_val,
+ "WarehouseCode": item.get("warehouse_code", "A-CENTRA"),
+ "CostingCode": "PRO",
+ "CostingCode2": "PRO03",
+ "TaxCode": "I18",
+ "FreeText": linea.get("descripcion", "")[:100],
+ "MeasureUnit": "UND",
+ "U_BPP_CTAD": "9231101",
+ "U_BPP_OPER": "A",
+ "U_tipoOpT12": "02",
+ "U_VS_ONEROSO": "1",
+ "U_VS_TIPAFE": "10",
+ "U_IDC_GRFLETE": codigo_guia,
+ })
+
+
+ valor_referencial = round((total_weight / 1000 )* (CUADRO_VALOR_REFERENCIAL.get(zona_guia, 0)),2)
+
+ return {
+ "DocType": "dDocument_Items",
+ "DocDate": fecha_contabilizacion,
+ "DocDueDate": fecha_emision,
+ "TaxDate": fecha_emision,
+ "CardCode": f"PL{ruc_emisor}",
+ "NumAtCard": f"{tipo}{serie}-{numero}",
+ "U_IDC_NUMGR": "" if not guia_referencia else guia_referencia,
+ "DocCurrency": "SOL",
+ "Comments": f"Factura {serie}-{numero} / {razon_social_emisor}"[:254],
+ "JournalMemo": razon_social_emisor[:254],
+ "DocObjectCode": "oPurchaseInvoices",
+ "U_BPV_SERI": f"{serie}-01",
+ "U_BPV_NCON2": numero,
+ "U_BPP_MDTD": tipo,
+ "U_BPP_MDBI": "A",
+ "U_BPP_MDCD": numero,
+ "U_BPP_MDSD": serie,
+ "U_VS_CODSER": f"{codigo_detraccion}02" if codigo_detraccion else "",
+ "U_VS_AFEDET": "Y" if tiene_detraccion else "N",
+ "U_VS_PORDET": porcentaje_detraccion,
+ "U_VS_APLANT": "Y",
+ "U_BPP_MDSD2": serie,
+ "U_IDC_FLETE": "02",
+ "U_IDC_LINEANE": "01",
+ "U_IDC_VALREF": valor_referencial,
+ "DocumentLines": doc_lines,
+ "DocumentInstallments": installments,
+ }
+
+ def _validate_prices(factura: dict) -> tuple[bool, list[dict]]:
+ """
+ Valida los precios de cada guía contra DeliveryNotes en SAP y
+ prepara los datos necesarios para construir el draft.
+
+ Flujo:
+ 1. Obtiene el maestro de zonas SAP (``U_IDC_ZONAS``) para resolver
+ el nombre de zona a partir de ``U_IDC_DESTINO``.
+ 2. Por cada línea de factura:
+ a. Itera sobre sus ``guias_remision[]``.
+ b. Para cada guía consulta SAP via
+ ``sapb1_manager.get_delivery_notes``.
+ c. Dentro de cada DeliveryNote, suma los ``LineTotal``,
+ ``PriceAfterVAT`` y obtiene la máxima ``Quantity`` de las
+ líneas cuyo ``ItemCode`` coincide con ``FL\\d``.
+ d. Acumula los resultados en ``guias_lt[]`` (uno por guía) con
+ ``line_total``, ``quantity``, ``price_after_vat_sum``,
+ ``warehouse_code`` y ``zona``.
+ 3. Determina si los precios coinciden (``ok``):
+ ``abs(total_linea - sap_total) < 0.01``.
+ 4. Decide si se deben usar precios de SAP en el draft
+ (``use_sap_prices``):
+ ``ok and int(cantidad) == 1 and int(total_sap_qty) != 1``.
+
+ Args:
+ factura: Diccionario con los datos de la factura (debe contener
+ ``lineas[]``, y cada línea debe tener ``guias_remision[]``
+ y ``total_linea``).
+
+ Returns:
+ tuple[bool, list[dict]]:
+ - ``todas_coinciden``: True si TODAS las líneas tienen
+ ``ok=True``.
+ - ``resultados``: Lista de resultados por línea, cada uno
+ con claves: ``grflete``, ``ok``, ``draft_price``,
+ ``sap_price``, ``warehouse_code``, ``draft_cantidad``,
+ ``use_sap_prices``, ``guias_lt[]``.
+
+ Example:
+ Factura con guía ``T001-127773`` y SAP devuelve
+ ``FL10007 qty=350 pav=1.3396, FL20007 qty=350 pav=0``::
+
+ ok, resultados = _validate_prices(factura)
+ # → ok = False (397.34 != 633.47)
+ # → resultados[0]["guias_lt"][0] = {
+ # "guia": {"tipo":"09","serie":"T001","numero":"127773"},
+ # "line_total": 397.34,
+ # "quantity": 350.0,
+ # "price_after_vat_sum": 1.3396,
+ # "zona": "CAÑETE",
+ # "warehouse_code": "A-CENTRA"
+ # }
+ """
+ resultados = []
+ factura_lines = factura.get("lineas", [])
+
+ zonas_raw = sapb1_manager.get_zonas()
+ zonas = {z["Code"]: z["Name"] for z in zonas_raw}
+
+ for fl in factura_lines:
+ cantidad = float(fl.get("cantidad", 0))
+ total_linea = float(fl.get("total_linea", 0) or 0)
+ guias = fl.get("guias_remision", [])
+
+ if cantidad == 0:
+ resultados.append({"grflete": "-", "ok": False, "draft_price": None, "sap_price": None, "label": "Cantidad cero, sin validar", "warehouse_code": "-", "draft_cantidad": cantidad, "guias_lt": []})
+ continue
+
+ if not guias:
+ resultados.append({"grflete": "-", "ok": False, "draft_price": None, "sap_price": None, "label": "Sin guía de referencia", "warehouse_code": "-", "draft_cantidad": cantidad, "guias_lt": []})
+ continue
+
+ guias_lt = []
+ all_wh = set()
+ all_ok = True
+
+ for guia in guias:
+ codigo_guia = f"{guia['tipo']}{guia['serie']}-{guia['numero']}"
+ try:
+ delivery_notes = sapb1_manager.get_delivery_notes(
+ tipo=guia["serie"], serie=guia["tipo"], numero=guia["numero"]
+ )
+ print(f" │ 🔍 Guía {codigo_guia}: {len(delivery_notes)} documento(s) encontrado(s) en SAP")
+
+ wh_set = set()
+ suma_line_total = 0.0
+ max_qty = 0.0
+ sum_pav = 0.0
+ weight1 = 0.0
+ zona = ""
+
+ for dn in delivery_notes:
+ zona = zonas.get(dn.get("U_IDC_DESTINO",""),"")
+
+ first_line = dn.get("DocumentLines", [None])[0]
+ if first_line:
+ wh = first_line.get("WarehouseCode", "")
+ if wh:
+ wh_set.add(wh)
+ weight1 += float(first_line.get("Weight1", 0) or 0)
+ for linea_detalle_dict in dn.get("DocumentLines", []):
+ if not re.match(r"FL\d", linea_detalle_dict.get("ItemCode", "")):
+ continue
+ line_total = float(linea_detalle_dict.get("LineTotal", 0) or 0)
+ price_after_vat = float(linea_detalle_dict.get("PriceAfterVAT", 0) or 0)
+ quantity = float(linea_detalle_dict.get("Quantity", 0) or 0)
+ suma_line_total += line_total
+ max_qty = max(max_qty, quantity)
+ sum_pav += price_after_vat
+ print(f" │ └─ {dn.get('Series', '?')}-{dn.get('DocNum', '?')} | ItemCode: {linea_detalle_dict.get('ItemCode', '')} | PriceAfterVAT: {price_after_vat:.6f} | Quantity: {quantity:.6f} | LineTotal: {line_total:.6f}")
+
+ guias_lt.append({"guia": guia, "line_total": suma_line_total, "warehouse_code": ", ".join(sorted(wh_set)) if wh_set else "A-CENTRA", "quantity": max_qty, "price_after_vat_sum": sum_pav, "zona":zona, "weight1": weight1, "draft_quantity": 0, "draft_price": 0.0})
+ all_wh.update(wh_set)
+ except Exception as ex:
+ print(f" │ ⚠️ Guía {codigo_guia}: error al consultar SAP - {ex}")
+ guias_lt.append({"guia": guia, "line_total": 0.0, "warehouse_code": "A-CENTRA", "weight1": 0.0, "quantity": 0.0, "price_after_vat_sum": 0.0, "zona": "", "draft_quantity": 0, "draft_price": 0.0})
+ all_ok = False
+
+ sap_total = sum(item["line_total"] for item in guias_lt)
+ ok = all_ok and abs(total_linea - sap_total) < 0.01
+ grflete_str = "; ".join(f"{item['guia']['tipo']}{item['guia']['serie']}-{item['guia']['numero']}" for item in guias_lt)
+ wh_str = ", ".join(sorted(all_wh)) if all_wh else "A-CENTRA"
+
+ for item in guias_lt:
+ if ok:
+ item["draft_quantity"] = int(item["quantity"])
+ item["draft_price"] = item["price_after_vat_sum"]
+ else:
+ item["draft_quantity"] = int(cantidad)
+ item["draft_price"] = float(total_linea) / max(float(cantidad), 1)
+
+ resultados.append({
+ "grflete": grflete_str,
+ "ok": ok,
+ "draft_price": total_linea,
+ "sap_price": sap_total,
+ "label": "LineTotal (FL*) vs total_linea",
+ "warehouse_code": wh_str,
+ "guias_lt": guias_lt,
+ })
+
+ print(f" │ Cantidad: {cantidad} | Total LineTotal (SAP): {sap_total:.6f} vs total_linea (XML): {total_linea:.6f}")
+ if ok:
+ print(f" │ ✅ Coincide: {total_linea:.6f} ≈ {sap_total:.6f}")
+ else:
+ print(f" │ ❌ NO coincide — XML: {total_linea:.6f} vs SAP: {sap_total:.6f}")
+
+ todas_coinciden = all(r["ok"] for r in resultados)
+ return todas_coinciden, resultados
+
+ def _build_validation_html(resultados: list[dict]) -> str:
+ """
+ Genera una tabla HTML con el resultado de la validación de precios
+ para incluir en el cuerpo del email de notificación.
+
+ Args:
+ resultados: Lista de resultados de ``_validate_prices``.
+
+ Returns:
+ str — Código HTML de la tabla con columnas:
+ Guía, Estado, Draft Price, SAP, Almacén, Resultado.
+
+ Example:
+ Con un resultado ``ok=True`` y ``grflete="09T001-127773"``::
+
+ html = _build_validation_html(resultados)
+ # → ""
+ """
+ rows = ""
+ for r in resultados:
+ icono = "✅" if r["ok"] else "❌"
+ dp = f"{r['draft_price']:.6f}" if r["draft_price"] is not None else "-"
+ sap_price_str = f"{r['sap_price']:.6f}" if r["sap_price"] is not None else "-"
+ rows += f"| {r['grflete']} | {icono} | {dp} | {sap_price_str} | {r['warehouse_code']} | {r['label']} |
"
+ return (
+ ""
+ "| Guía | Estado | Draft Price | SAP | Almacén | Resultado |
|---|
"
+ f"{rows}
"
+ )
+
+ def _build_notification_body(status: str, factura: dict, validation_table: str, **kwargs) -> str:
+ """
+ Construye el cuerpo HTML del email de notificación del resultado
+ del proceso SAP.
+
+ Args:
+ status: ``"created"`` (draft creado), ``"error"`` (error al
+ crear), o cualquier otro valor (validación falló, draft no
+ creado).
+ factura: Diccionario de la factura (se usa ``serie``,
+ ``numero``, ``razon_social_emisor``).
+ validation_table: Tabla HTML generada por
+ ``_build_validation_html``.
+ **kwargs: Argumentos adicionales:
+ - ``doc_entry`` (int): DocEntry del draft creado.
+ - ``error`` (str): Mensaje de error si ocurrió.
+
+ Returns:
+ str — Cuerpo HTML completo del email.
+
+ Example:
+ Draft creado exitosamente::
+
+ body = _build_notification_body(
+ "created", factura, tabla_html, doc_entry=2021
+ )
+ # → "✅ Draft SAP B1 creado exitosamente.
..."
+ """
+ base = (
+ f"Factura: {factura.get('serie')}-{factura.get('numero')}
"
+ f"Emisor: {factura.get('razon_social_emisor', '')}
"
+ )
+ if status == "created":
+ return f"✅ Draft SAP B1 creado exitosamente.
DocEntry: {kwargs.get('doc_entry')}
{base}Validación de precios
{validation_table}"
+ elif status == "error":
+ return f"❌ Error al crear Draft SAP B1.
{base}Error: {kwargs.get('error')}
Validación de precios
{validation_table}"
+ else:
+ return f"⏭️ Draft NO creado. Los precios de las guías no coinciden con SAP.
{base}Validación de precios
{validation_table}"
+
+ def _guardar_draft_json(draft: dict, serie: str, numero: str) -> None:
+ """
+ Guarda el payload del draft SAP B1 en un archivo JSON para
+ depuración.
+
+ El nombre sigue el patrón
+ ``draft_sapb1_{serie}-{numero}_{timestamp}.json``.
+
+ Args:
+ draft: Payload del draft SAP B1 (producido por
+ ``_build_sap_draft``).
+ serie: Serie de la factura (ej. ``"F002"``).
+ numero: Número de la factura (ej. ``"486"``).
+
+ Example:
+ Draft de factura F002-486::
+
+ _guardar_draft_json(draft, "F002", "486")
+ # → Crea draft_sapb1_F002-486_2026-07-08_12-00-00.json
+ """
+ if not os.path.exists(output_folder):
+ os.makedirs(output_folder, exist_ok=True)
+ ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
+ filename = f"draft_sapb1_{serie}-{numero}_{ts}.json"
+ with open(os.path.join(output_folder, filename), "w", encoding="utf-8") as f:
+ json.dump(draft, f, ensure_ascii=False, indent=4)
+ print(f" 💾 Draft guardado: {filename}")
+
+ # ========================================================================
+ # FLUJO DEL CORREO
+ # ========================================================================
+
+ print(f"\n📬 [{email['fecha']}] {email['subject'][:80]}...")
+ print(f" ├── 📂 Adjuntos:")
+
+ # 1-2. BD: verificar existencia y guardar
+ exists = database_utilities.get_existing_ids([email["id"]])
+ if email["id"] in exists:
+ print(f" ├── ⏭️ Correo ya procesado, saltando")
+ return
+ database_utilities.save_emails([email])
+
+ # 3. Descargar adjuntos
+ adjuntos = _descargar_adjuntos()
+ if not adjuntos:
+ print(f" ├── ⏭️ Correo sin adjuntos")
+ return
+
+ # 4. Descomprimir ZIPs
+ _descomprimir()
+
+ # 5. Filtrar PDF/XML y limpiar si no hay válidos
+ if not _filtrar_y_limpiar():
+ return
+
+ # 6. Procesar PDFs
+ pdf_cache = _procesar_pdfs()
+
+ # 7. Procesar XMLs
+ documentos = _procesar_xmls(pdf_cache)
+ if not documentos:
+ print(f" ├── ⏭️ No se extrajeron documentos XML")
+ return
+
+ # 8. Guardar resultado con todos los documentos
+ _guardar_resultado(documentos)
+
+ # 9. Filtrar solo facturas para SAP
+ facturas = [d for d in documentos if d.get("tipo") == "01"]
+ if not facturas:
+ print(f" ├── ⏭️ No hay facturas (tipo 01) para SAP")
+ return
+
+ for factura in facturas:
+ serie = factura.get("serie", "?")
+ numero = factura.get("numero", "?")
+
+ # 10. Validar precios (consulta SAP DNs, NO requiere draft)
+ todas_coinciden, resultados_val = _validate_prices(factura)
+ validation_table = _build_validation_html(resultados_val)
+
+ # 11. Construir draft (usa LineTotal de los DNs, no del XML)
+ draft = _build_sap_draft(factura, resultados_val)
+
+ adjuntos_email = [p for p in [factura.get("ruta_xml"), factura.get("ruta_pdf")] if p]
+
+ # 12. Guardar draft JSON (siempre, para depuración)
+ _guardar_draft_json(draft, serie, numero)
+
+ # 13. Crear draft en SAP o generar HTML
+ if todas_coinciden:
+ try:
+ sap_result = sapb1_manager.create_draft(draft)
+ doc_entry = sap_result.get("DocEntry")
+ print(f" ✅ Draft SAP B1 creado: {serie}-{numero} → DocEntry {doc_entry}")
+ html_body = _build_notification_body("created", factura, validation_table, doc_entry=doc_entry)
+ except Exception as ex:
+ print(f" ❌ Error creando draft {serie}-{numero}: {ex}")
+ html_body = _build_notification_body("error", factura, validation_table, error=ex)
+ else:
+ print(f" ⏭️ Draft NO creado: {serie}-{numero} — validación falló")
+ html_body = _build_notification_body("skipped", factura, validation_table)
+
+ # 14. Enviar notificación
+ if token:
+ outlook_utilities.send_email(
+ token, email_addr,
+ ["jlopez@aberio.com.pe", "gcalle@pucp.pe"],
+ f"Draft SAP B1 - {serie}-{numero}",
+ html_body, adjuntos_email,
+ )
+
+
+# ============================================================================
+# PIPELINE COMPLETO
+# ============================================================================
+def run_pipeline(dias: int, EMAIL: str, DOWNLOAD_FOLDER: str, OUTPUT_FOLDER: str) -> None:
+ """Ejecuta el pipeline completo: autentica, descarga correos y procesa cada uno."""
+
+ database_utilities.init_db()
+
+ hoy = datetime.now()
+ ayer = hoy - timedelta(days=dias)
+ fecha_inicio = ayer.strftime("%Y-%m-%d")
+ fecha_fin = hoy.strftime("%Y-%m-%d")
+ print(f"Rango de búsqueda: {fecha_inicio} al {fecha_fin}")
+
+ # 1. Autenticar
+ token = step_authenticate()
+ if not token:
+ print("❌ No se pudo autenticar. Abortando.")
+ return
+
+ # 2. Descargar correos
+ emails = step_fetch_emails(token, EMAIL, fecha_inicio, fecha_fin)
+ if not emails:
+ print("No se encontraron correos.")
+ return
+
+ # 3. Bucle por cada correo
+ for email in emails:
+ _procesar_un_correo(email, token, EMAIL, DOWNLOAD_FOLDER, OUTPUT_FOLDER)
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..bea4fa8
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,16 @@
+certifi==2026.6.17
+cffi==2.1.0
+charset-normalizer==3.4.9
+cryptography==49.0.0
+dotenv==0.9.9
+idna==3.18
+msal==1.37.0
+pdfminer.six==20260107
+pdfplumber==0.11.10
+pillow==12.3.0
+pycparser==3.0
+PyJWT==2.13.0
+pypdfium2==5.11.0
+python-dotenv==1.2.2
+requests==2.34.2
+urllib3==2.7.0
diff --git a/sapb1_manager.py b/sapb1_manager.py
new file mode 100644
index 0000000..8f378ed
--- /dev/null
+++ b/sapb1_manager.py
@@ -0,0 +1,252 @@
+
+import requests
+import urllib3
+import os
+import json
+from typing import Optional, Dict, Any
+from dotenv import load_dotenv
+
+
+# ============================================================================
+# CONFIGURACIÓN
+# ============================================================================
+
+urllib3.disable_warnings()
+
+load_dotenv(dotenv_path=".env.dev")
+
+SAPB1_URL = os.getenv("SAP_B1_URL", "")
+SAPB1_USER = os.getenv("SAP_B1_USER", "")
+SAPB1_PASSWORD = os.getenv("SAP_B1_PASSWORD", "")
+SAPB1_COMPANY = os.getenv("SAP_B1_COMPANY", "")
+SAPB1_TIMEOUT = float(os.getenv("SAP_B1_TIMEOUT", 15.0))
+SAPB1_VERIFY_SSL = os.getenv("SAP_B1_VERIFY_SSL", "False").lower() == "true"
+
+# Almacenamiento global de sesión
+_sap_session: dict = {"headers": None, "b1session": None, "routeid": None}
+
+
+# ============================================================================
+# LOGIN
+# ============================================================================
+def login() -> dict:
+ payload = {
+ "CompanyDB": SAPB1_COMPANY,
+ "UserName": SAPB1_USER,
+ "Password": SAPB1_PASSWORD,
+ }
+ print("🔐 Iniciando sesión en SAP B1 Service Layer...")
+ resp = requests.post(
+ f"{SAPB1_URL}/Login",
+ json=payload,
+ timeout=SAPB1_TIMEOUT,
+ verify=SAPB1_VERIFY_SSL,
+ )
+ if resp.status_code != 200:
+ raise ConnectionError(
+ f"Login fallido: HTTP {resp.status_code} - {resp.text}"
+ )
+
+ data = resp.json()
+ b1session = data["SessionId"]
+ routeid = resp.cookies.get("ROUTEID", "")
+
+ headers = {"Cookie": f"B1SESSION={b1session}; ROUTEID={routeid}"}
+ _sap_session.update({"headers": headers, "b1session": b1session, "routeid": routeid})
+ print(f"✅ Sesión SAP B1 establecida (B1SESSION: {b1session[:10]}...)")
+ return headers
+
+
+# ============================================================================
+# LOGOUT
+# ============================================================================
+def logout() -> None:
+ if not _sap_session["b1session"]:
+ print("⚠️ No hay sesión activa para cerrar.")
+ return
+ print("🔓 Cerrando sesión SAP B1...")
+ try:
+ resp = requests.post(
+ f"{SAPB1_URL}/Logout",
+ headers=_sap_session["headers"],
+ timeout=SAPB1_TIMEOUT,
+ verify=SAPB1_VERIFY_SSL,
+ )
+ if resp.status_code == 200:
+ print("✅ Sesión SAP B1 cerrada correctamente.")
+ else:
+ print(f"⚠️ Logout respondió HTTP {resp.status_code}")
+ except Exception as exc:
+ print(f"❌ Error al cerrar sesión: {exc}")
+ finally:
+ _sap_session.update({"headers": None, "b1session": None, "routeid": None})
+
+
+# ============================================================================
+# SESIÓN (interna - lazy login)
+# ============================================================================
+def _ensure_session() -> dict:
+ if _sap_session["headers"]:
+ return _sap_session["headers"]
+ return login()
+
+
+# ============================================================================
+# LOGGING DE RESPUESTA
+# ============================================================================
+def _log_response(method: str, endpoint: str, response: requests.Response, payload: Any = None) -> None:
+ log_line = (
+ f" ├── [{method}] {endpoint} → {response.status_code}"
+ f" ({len(response.content)} bytes)"
+ )
+ if response.status_code >= 400:
+ print("⚠️ %s - %s", log_line, response.text[:300])
+ else:
+ print(log_line)
+
+
+# ============================================================================
+# PROXY CENTRALIZADO
+# ============================================================================
+def forward_to_sapb1(
+ method: str,
+ endpoint_path: str,
+ json_data: Optional[Dict[str, Any]] = None,
+ params: Optional[Dict[str, Any]] = None,
+ extra_headers: Optional[Dict[str, str]] = None
+) -> requests.Response:
+ if not method or not endpoint_path:
+ raise ValueError("method y endpoint_path son requeridos")
+
+ method = method.upper()
+ headers = _ensure_session()
+
+ url = f"{SAPB1_URL}/{endpoint_path}"
+ req_headers = {
+ "Content-Type": "application/json",
+ "Accept": "application/json",
+ **_sap_session["headers"],
+ }
+
+ if extra_headers:
+ req_headers.update(extra_headers)
+
+ print(f" ├── 🔄 [{method}] {endpoint_path}")
+
+ try:
+ resp = requests.request(
+ method=method,
+ url=url,
+ json=json_data,
+ params=params,
+ headers=req_headers,
+ timeout=SAPB1_TIMEOUT,
+ verify=SAPB1_VERIFY_SSL,
+ )
+ except requests.exceptions.Timeout:
+ raise ConnectionError(f"Timeout ({SAPB1_TIMEOUT}s) en {endpoint_path}")
+ except requests.exceptions.ConnectionError as exc:
+ raise ConnectionError(f"No se pudo conectar con SAP B1: {exc}")
+ except requests.RequestException as exc:
+ raise ConnectionError(f"Error en solicitud HTTP: {exc}")
+
+ if resp.status_code == 401:
+ print("⚠️ Sesión expirada (401). Reautenticando e intentando una vez...")
+ req_headers["Cookie"] = login()["Cookie"]
+ resp = requests.request(
+ method=method,
+ url=url,
+ json=json_data,
+ params=params,
+ headers=req_headers,
+ timeout=SAPB1_TIMEOUT,
+ verify=SAPB1_VERIFY_SSL,
+ )
+
+ _log_response(method, endpoint_path, resp, json_data)
+
+ if resp.status_code >= 400:
+ try:
+ error_body = resp.json()
+ sap_message = (
+ error_body.get("error", {})
+ .get("message", {})
+ .get("value", resp.text)
+ )
+ except (json.JSONDecodeError, AttributeError):
+ sap_message = resp.text
+ raise ConnectionError(
+ f"Error en SAP {method} {endpoint_path}: "
+ f"HTTP {resp.status_code} - {sap_message}"
+ )
+
+ return resp
+
+
+# ============================================================================
+# CREAR DRAFT (POST /Drafts)
+# ============================================================================
+def create_draft(payload: dict) -> dict:
+ resp = forward_to_sapb1("POST", "Drafts", json_data=payload)
+ result = resp.json()
+ print(f" ├── ✅ Draft creado DocEntry={result.get('DocEntry')}")
+ return result
+
+
+# ============================================================================
+# OBTENER FACTURA DE COMPRA (GET /PurchaseInvoices(DocEntry))
+# ============================================================================
+def get_purchase_invoice(doc_entry: int) -> dict:
+ resp = forward_to_sapb1("GET", f"PurchaseInvoices({doc_entry})")
+ result = resp.json()
+ print(f" ├── ✅ PurchaseInvoices({doc_entry}) obtenida")
+ return result
+
+# ============================================================================
+# OBTENER GUIA (GET /DeliveryNotes)
+# ============================================================================
+def get_delivery_notes(tipo: str, serie: str, numero: str) -> list:
+ filter_str = (
+ f"Cancelled eq 'tNO'"
+ f" and U_BPP_MDSD eq '{tipo}'"
+ f" and U_BPP_MDTD eq '{serie}'"
+ f" and contains(U_BPP_MDCD,'{numero}')"
+ )
+ resp = forward_to_sapb1("GET", "DeliveryNotes", params={"$filter": filter_str})
+ result = resp.json()
+ value = result.get("value", [])
+ print(f" ├── ✅ DeliveryNotes obtenida ({len(value)} registros)")
+ return value
+
+# ============================================================================
+# OBTENER ZONAS (GET /U_IDC_ZONAS)
+# ============================================================================
+def get_zonas() -> list:
+
+ filter_str = (
+ f"U_Activo eq 'Y'"
+ )
+
+ sap_params = {
+ "$skip": "0",
+ "$top": "100",
+ "$inlinecount": "allpages"
+ }
+
+ sap_params["$select"] = "Code, Name"
+
+ base_headers = {"B1S-PageSize": "100"}
+
+ if filter_str:
+ sap_params["$filter"] = filter_str
+
+ resp = forward_to_sapb1(
+ method="GET"
+ , endpoint_path="U_IDC_ZONAS"
+ , params=sap_params
+ , extra_headers=base_headers
+ )
+ result = resp.json()
+ value = result.get("value", [])
+ print(f" ├── ✅ U_IDC_ZONAS obtenida ({len(value)} registros)")
+ return value
diff --git a/sunat_schema.py b/sunat_schema.py
index 084e284..2190634 100644
--- a/sunat_schema.py
+++ b/sunat_schema.py
@@ -6,12 +6,12 @@ Soporta los siguientes tipos de documento UBL 2.1:
- DespatchAdvice (Guía de Remisión Remitente): tipo 09
- DespatchAdvice (Guía de Remisión Transportista): tipo 31
"""
-
+import re
import xml.etree.ElementTree as ET
from dataclasses import dataclass, asdict, field
from typing import Optional, List
-########################################################
+# ============================================================================
# Namespaces UBL 2.1 utilizados en los XML de SUNAT
NAMESPACES = {
"inv": "urn:oasis:names:specification:ubl:schema:xsd:Invoice-2",
@@ -30,7 +30,7 @@ TIPOS_DOCUMENTO = {
"09": "Guía de Remisión Remitente",
"31": "Guía de Remisión Transportista",
}
-########################################################
+# ============================================================================
@dataclass
class DocumentoReferencia:
"""Representa un documento de referencia (p.ej. Guía de Remisión asociada)."""
@@ -46,6 +46,7 @@ class DetalleLinea:
descripcion: str
cantidad: float
precio_unitario: float
+ total_linea: float = 0.0
@dataclass
@@ -121,7 +122,7 @@ class DatosComprobante:
)
-########################################################
+# ============================================================================
def _find_text(element: ET.Element, xpath: str, namespaces: dict) -> Optional[str]:
"""Busca un elemento por XPath y devuelve su texto, o None si no existe."""
node = element.find(xpath, namespaces)
@@ -129,7 +130,7 @@ def _find_text(element: ET.Element, xpath: str, namespaces: dict) -> Optional[st
return node.text.strip()
return None
-########################################################
+# ============================================================================
def _split_serie_numero(comprobante_id: str) -> tuple[str, str]:
"""
Separa el ID del comprobante en serie y número, removiendo ceros a la izquierda del número.
@@ -144,7 +145,26 @@ def _split_serie_numero(comprobante_id: str) -> tuple[str, str]:
return partes[0], numero
return comprobante_id, ""
-########################################################
+# ============================================================================
+def _extraer_serie_numero_desde_texto(texto: str) -> tuple[str, str] | None:
+ texto = re.sub(r'[–—−]', '-', texto)
+ patron = r'\b([TVE][A-Z0-9]{3})\s*-\s*(\d+)\b'
+ m = re.search(patron, texto)
+ if m:
+ return m.group(1).upper(), m.group(2)
+ return None
+
+
+def obtener_documentos_referencia_detalle_factura(texto: str) -> list[str]:
+ texto = re.sub(r'[–—−]', '-', texto)
+
+ patron = r'\b([TVE][A-Z0-9]{3})\s*-\s*(\d+)\b'
+
+ return list(dict.fromkeys(
+ f"{serie.upper()}-{numero}"
+ for serie, numero in re.findall(patron, texto, flags=re.IGNORECASE)
+ ))
+# ============================================================================
def _parse_invoice(root: ET.Element) -> DatosComprobante:
"""Extrae datos de un documento Invoice (Factura/Boleta)."""
ns = NAMESPACES
@@ -203,8 +223,27 @@ def _parse_invoice(root: ET.Element) -> DatosComprobante:
ref_id = _find_text(ref_node, "cbc:ID", ns) or ""
ref_tipo = _find_text(ref_node, "cbc:DocumentTypeCode", ns) or ""
if ref_id or ref_tipo:
- documentos_referencia.append(DocumentoReferencia(tipo=ref_tipo, serie=_split_serie_numero(ref_id)[0], numero=_split_serie_numero(ref_id)[1]))
+ parsed = _extraer_serie_numero_desde_texto(ref_id)
+ if parsed:
+ ref_serie, ref_numero = parsed
+ else:
+ ref_serie, ref_numero = _split_serie_numero(ref_id)
+ documentos_referencia.append(DocumentoReferencia(tipo=ref_tipo, serie=ref_serie, numero=ref_numero))
+ # si los documentos de referencia non estan en los tags de SUNAT
+ # buscamos estos documentos en el cuerpo de la factura
+ if len(documentos_referencia) == 0:
+ first_invoice_line = root.find('.//cac:InvoiceLine', ns)
+ if first_invoice_line is not None:
+ item = first_invoice_line.find('cac:Item', ns)
+ if item is not None:
+ description = item.find('cbc:Description', ns)
+ if description is not None:
+ guias = obtener_documentos_referencia_detalle_factura(description.text)
+ if guias:
+ for guia in guias:
+ documentos_referencia.append(DocumentoReferencia(tipo="09" if guia[:1] in ("T", "V", "E") else "31", serie=_split_serie_numero(guia)[0], numero=_split_serie_numero(guia)[1]))
+
# Inicializar campos de detracción y líneas
tiene_detraccion = False
codigo_detraccion = None
@@ -242,11 +281,15 @@ def _parse_invoice(root: ET.Element) -> DatosComprobante:
)
precio_unitario = float(price_str) if price_str else 0.0
+ total_linea_str = _find_text(line_node, "cbc:LineExtensionAmount", ns)
+ total_linea = float(total_linea_str) if total_linea_str else 0.0
+
lineas.append(
DetalleLinea(
descripcion=descripcion,
cantidad=cantidad,
precio_unitario=precio_unitario,
+ total_linea=total_linea,
)
)
@@ -268,7 +311,7 @@ def _parse_invoice(root: ET.Element) -> DatosComprobante:
razon_social_emisor=razon_social_emisor,
)
-########################################################
+# ============================================================================
def _parse_despatch_advice(root: ET.Element) -> DatosComprobante:
"""Extrae datos de un documento DespatchAdvice (Guía de Remisión)."""
ns = NAMESPACES
@@ -354,6 +397,37 @@ def _parse_despatch_advice(root: ET.Element) -> DatosComprobante:
# Las guías de remisión no tienen monto total
total_factura = None
+
+ # Documentos de referencia
+ documentos_referencia = []
+ for ref_node in root.findall("cac:AdditionalDocumentReference", ns):
+ ref_id = _find_text(ref_node, "cbc:ID", ns) or ""
+ ref_tipo = _find_text(ref_node, "cbc:DocumentTypeCode", ns) or ""
+ if ref_id or ref_tipo:
+ parsed = _extraer_serie_numero_desde_texto(ref_id)
+ if parsed:
+ ref_serie, ref_numero = parsed
+ else:
+ ref_serie, ref_numero = _split_serie_numero(ref_id)
+ documentos_referencia.append(DocumentoReferencia(tipo=ref_tipo, serie=ref_serie, numero=ref_numero))
+
+ lineas = []
+ # Obtener cada cac:InvoiceLine
+ for line_node in root.findall("cac:DespatchLine", ns):
+ descripcion = _find_text(line_node, "cac:Item/cbc:Description", ns) or ""
+
+ qty_str = _find_text(line_node, "cbc:DeliveredQuantity", ns)
+ cantidad = float(qty_str) if qty_str else 0.0
+
+ precio_unitario = 0.0
+
+ lineas.append(
+ DetalleLinea(
+ descripcion=descripcion,
+ cantidad=cantidad,
+ precio_unitario=precio_unitario,
+ )
+ )
return DatosComprobante(
ruc_emisor=ruc_emisor,
@@ -365,6 +439,7 @@ def _parse_despatch_advice(root: ET.Element) -> DatosComprobante:
ruc_receptor=ruc_receptor,
razon_social_receptor=razon_social_receptor,
total_factura=total_factura,
+ documentos_referencia=documentos_referencia,
punto_partida=punto_partida,
punto_llegada=punto_llegada,
peso_bruto=peso_bruto,
@@ -375,10 +450,11 @@ def _parse_despatch_advice(root: ET.Element) -> DatosComprobante:
licencia_conductor=licencia_conductor,
nombre_conductor=nombre_conductor,
documento_conductor=documento_conductor,
- ubigeo_partida=ubigeo_partida
+ ubigeo_partida=ubigeo_partida,
+ lineas=lineas
)
-########################################################
+# ============================================================================
def extraer_datos_xml(ruta_xml: str) -> Optional[dict]:
"""
Recibe la ruta de un archivo XML de comprobante electrónico SUNAT.
@@ -414,11 +490,9 @@ def extraer_datos_xml(ruta_xml: str) -> Optional[dict]:
return datos.to_dict()
return None
-
-########################################################
-# ---------------------------------------------------------------------------
+# ============================================================================
# Ejecución directa: procesa todos los XML de la carpeta adjuntos como prueba
-# ---------------------------------------------------------------------------
+# ============================================================================
if __name__ == "__main__":
import os
import glob
@@ -443,3 +517,4 @@ if __name__ == "__main__":
except Exception as e:
print(f"\nError procesando {os.path.basename(archivo)}: {e}\n")
print("=" * 70)
+# ============================================================================
\ No newline at end of file
diff --git a/test/aberio/BusinessPartners.yml b/test/aberio/BusinessPartners.yml
new file mode 100644
index 0000000..dfa66d5
--- /dev/null
+++ b/test/aberio/BusinessPartners.yml
@@ -0,0 +1,15 @@
+info:
+ name: BusinessPartners
+ type: http
+ seq: 1
+
+http:
+ method: GET
+ url: https://192.168.9.3:50000/b1s/v1/BusinessPartners('PL10040532043')
+ auth: inherit
+
+settings:
+ encodeUrl: true
+ timeout: 0
+ followRedirects: true
+ maxRedirects: 5
diff --git a/test/aberio/EmployeesInfo.yml b/test/aberio/EmployeesInfo.yml
new file mode 100644
index 0000000..47b8216
--- /dev/null
+++ b/test/aberio/EmployeesInfo.yml
@@ -0,0 +1,14 @@
+info:
+ name: EmployeesInfo
+ type: http
+
+http:
+ method: GET
+ url: https://192.168.9.3:50000/b1s/v1/EmployeesInfo
+ auth: inherit
+
+settings:
+ encodeUrl: true
+ timeout: 0
+ followRedirects: true
+ maxRedirects: 5
diff --git a/test/aberio/Login.yml b/test/aberio/Login.yml
new file mode 100644
index 0000000..e128208
--- /dev/null
+++ b/test/aberio/Login.yml
@@ -0,0 +1,23 @@
+info:
+ name: Login
+ type: http
+ seq: 1
+
+http:
+ method: POST
+ url: https://192.168.9.3:50000/b1s/v1/Login
+ body:
+ type: json
+ data: |-
+ {
+ "CompanyDB" :"PRUEBA_20260609",
+ "UserName": "manager",
+ "Password":"s1st3m4s"
+ }
+ auth: inherit
+
+settings:
+ encodeUrl: true
+ timeout: 0
+ followRedirects: true
+ maxRedirects: 5
diff --git a/test/aberio/PurchaseInvoices.yml b/test/aberio/PurchaseInvoices.yml
new file mode 100644
index 0000000..dcc838e
--- /dev/null
+++ b/test/aberio/PurchaseInvoices.yml
@@ -0,0 +1,19 @@
+info:
+ name: PurchaseInvoices
+ type: http
+ seq: 1
+
+http:
+ method: GET
+ url: https://192.168.9.3:50000/b1s/v1/PurchaseInvoices?$filter=DocDate eq '2026-06-10'
+ params:
+ - name: $filter
+ value: DocDate eq '2026-06-10'
+ type: query
+ auth: inherit
+
+settings:
+ encodeUrl: true
+ timeout: 0
+ followRedirects: true
+ maxRedirects: 5
diff --git a/test/aberio/opencollection.yml b/test/aberio/opencollection.yml
new file mode 100644
index 0000000..c9e815d
--- /dev/null
+++ b/test/aberio/opencollection.yml
@@ -0,0 +1,10 @@
+opencollection: 1.0.0
+
+info:
+ name: aberio
+bundled: false
+extensions:
+ bruno:
+ ignore:
+ - node_modules
+ - .git
diff --git a/test/ids_prueba.json b/test/ids_prueba.json
new file mode 100644
index 0000000..817964c
--- /dev/null
+++ b/test/ids_prueba.json
@@ -0,0 +1,10 @@
+{
+ "coinciden_ok": {
+ "descripcion": "Facturas donde LineTotal SAP = XML, deben usar Quantity y PriceAfterVAT de SAP",
+ "ids": [
+ "AAMkADljMzNhYmQwLTQyMjgtNDY2My05ZDUwLWEzNTBiNDZhZjFlMwBGAAAAAAB386nfKhkcTbYV6pPUJ245BwBdNWALkxgeSJ3BdBRukthBAAAAAAEMAABdNWALkxgeSJ3BdBRukthBAAASMGzzAAA=",
+ "AAMkADljMzNhYmQwLTQyMjgtNDY2My05ZDUwLWEzNTBiNDZhZjFlMwBGAAAAAAB386nfKhkcTbYV6pPUJ245BwBdNWALkxgeSJ3BdBRukthBAAAAAAEMAABdNWALkxgeSJ3BdBRukthBAAASMGztAAA=",
+ "AAMkADljMzNhYmQwLTQyMjgtNDY2My05ZDUwLWEzNTBiNDZhZjFlMwBGAAAAAAB386nfKhkcTbYV6pPUJ245BwBdNWALkxgeSJ3BdBRukthBAAAAAAEMAABdNWALkxgeSJ3BdBRukthBAAASMGzxAAA="
+ ]
+ }
+}
\ No newline at end of file