georgeandClaude Sonnet 5
15e3b5079c
- hana_manager.get_delivery ahora busca en ODLN y, si no encuentra, en OWTR - sapb1_manager resuelve StockTransfers -> Orders (via U_IDC_NUMOV) para validar precios de flete cuando la guia es un traslado de inventario - se corrige el orden de parametros tipo/serie/numero (sin cruces) en get_delivery, get_delivery_notes y _get_delivery_entries - se agrega sapb1_manager.obtener_condicion_pago (BusinessPartners + PaymentTermsTypes) y pipeline._calcular_fecha_vencimiento para que DocDueDate del draft sea fecha de contabilizacion + dias de la condicion de pago del proveedor - se agrega variante de regex para "VALOR REFER" truncado en sunat_schema.py - se actualiza documentacion (docs/) reflejando estos cambios Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
115 lines
4.0 KiB
Python
115 lines
4.0 KiB
Python
"""
|
|
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_manager
|
|
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:
|
|
# Ruta absoluta: no depende del directorio de trabajo desde el que se lance el proceso
|
|
load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), ".env"))
|
|
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")
|
|
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)}")
|
|
|
|
logger_manager.setup_logging()
|
|
|
|
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: {logger_manager.LOG_FILE}")
|
|
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) |