295 lines
11 KiB
Python
295 lines
11 KiB
Python
import sys
|
||
import json
|
||
import os
|
||
import time
|
||
import signal
|
||
from datetime import datetime, timedelta
|
||
|
||
import outlook_utilities
|
||
import database_utilities
|
||
import email_services
|
||
import document_processor
|
||
import file_manager
|
||
import llm_processor
|
||
from smb_manager import sincronizar_a_smb
|
||
from logger_utilities import setup_logger, limpiar_logs_antiguos, rotar_log
|
||
from config_loader import CONFIG
|
||
|
||
|
||
# ============================================================================
|
||
# CONFIGURACIÓN DE HORARIOS (cargado desde scheduler.json)
|
||
# ============================================================================
|
||
_SCHEDULER_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "scheduler.json")
|
||
with open(_SCHEDULER_PATH, "r", encoding="utf-8") as _f:
|
||
_sched = json.load(_f)
|
||
|
||
HORARIOS = [tuple(h) for h in _sched["HORARIOS"]]
|
||
DIAS_ACTIVOS = _sched["DIAS_ACTIVOS"]
|
||
|
||
_running = True
|
||
|
||
# ─── Colores ────────────────────────────────────────────
|
||
color_verde = "\033[92m"
|
||
color_amarillo = "\033[93m"
|
||
color_rojo = "\033[91m"
|
||
color_fin = "\033[0m"
|
||
|
||
# ========================================================================
|
||
def ic(ok, texto_ok, texto_fail):
|
||
icono = "✅" if ok else "⚠️"
|
||
color = color_verde if ok else color_amarillo
|
||
return f"{icono} {color}{texto_ok if ok else texto_fail}{color_fin}"
|
||
|
||
# ========================================================================
|
||
def obtener_intervalo(dt):
|
||
minutos = dt.hour * 60 + dt.minute
|
||
for hi, mi, hf, mf, intervalo, modo in HORARIOS:
|
||
inicio = hi * 60 + mi
|
||
fin = hf * 60 + mf
|
||
if inicio <= minutos < fin:
|
||
return (intervalo, modo)
|
||
return None
|
||
|
||
# ========================================================================
|
||
def segundos_hasta_proximo_inicio(dt):
|
||
target = dt.replace(hour=6, minute=30, second=0, microsecond=0)
|
||
if dt >= target:
|
||
target += timedelta(days=1)
|
||
return int((target - dt).total_seconds())
|
||
|
||
# ========================================================================
|
||
def segundos_hasta_proximo_dia_activo(dt):
|
||
for dias in range(1, 8):
|
||
prox = dt + timedelta(days=dias)
|
||
if prox.weekday() in DIAS_ACTIVOS:
|
||
target = prox.replace(hour=6, minute=30, second=0, microsecond=0)
|
||
return int((target - dt).total_seconds())
|
||
return 86400
|
||
|
||
# ========================================================================
|
||
def obtener_token(outlook_token, fecha_obtencion):
|
||
"""Reusa el token si fue obtenido hace menos de 50 minutos, si no renueva."""
|
||
if outlook_token and fecha_obtencion:
|
||
delta = datetime.now() - fecha_obtencion
|
||
if delta.total_seconds() < 3000: # ~50 min (token MSAL dura 60 min)
|
||
return outlook_token
|
||
|
||
print("\n🔑 Obteniendo nuevo token de acceso...")
|
||
nuevo = outlook_utilities.get_access_token()
|
||
print(f" {ic(True, 'Token obtenido.', '')}")
|
||
return nuevo
|
||
|
||
# ========================================================================
|
||
def procesar_cuenta(mail, outlook_token, fecha_inicio, fecha_fin):
|
||
"""Procesa todos los correos nuevos de una cuenta."""
|
||
print(f"\n{'='*60}")
|
||
print(f"📧 [{mail.email}]")
|
||
|
||
emails = outlook_utilities.get_emails(mail.email, outlook_token, fecha_inicio, fecha_fin)
|
||
print(f" 📬 {len(emails)} correos en el rango.")
|
||
|
||
correos_nuevos = email_services.filtrar_correos_nuevos(emails)
|
||
if not correos_nuevos:
|
||
print(f" {ic(True, 'Sin correos nuevos.', '')}")
|
||
return []
|
||
|
||
print(f" 🆕 {len(correos_nuevos)} nuevos (no procesados antes).")
|
||
procesados = []
|
||
|
||
for e in correos_nuevos:
|
||
if not _running:
|
||
if llm_processor.CONTADOR_LLM > 0:
|
||
print(f" 🤖 LLM invocado {llm_processor.CONTADOR_LLM} vez/veces (modelo: {llm_processor.MODEL})")
|
||
return procesados
|
||
if not e['has_attach']:
|
||
continue
|
||
|
||
try:
|
||
archivos_orig = outlook_utilities.download_attachments(
|
||
mail.email, e['id'], outlook_token, mail.ruta_descarga
|
||
)
|
||
if not archivos_orig:
|
||
continue
|
||
|
||
print(f"\n ├── 📬 [{e['fecha']}] {e['subject'][:80]}...")
|
||
|
||
archivos_validos = email_services.procesar_adjuntos_email(
|
||
e, archivos_orig.copy(), mail.ruta_descarga
|
||
)
|
||
if not archivos_validos:
|
||
print(f" │ {ic(False, '', 'Descartado: sin PDF/XML.')}")
|
||
file_manager.limpiar_carpeta_descarga(mail.ruta_descarga)
|
||
continue
|
||
|
||
metadatos = document_processor.procesar_adjuntos_enriquecidos(e, mail.ruta_descarga)
|
||
|
||
for a in archivos_orig:
|
||
if a.lower().endswith('.zip'):
|
||
print(f" │ 📦 Descomprimido: {a}")
|
||
|
||
# Clasificar archivos por completitud de metadata
|
||
completos = [m for m in metadatos if m.get('archivo_propuesto') is not None]
|
||
parciales = [m for m in metadatos if m.get('archivo_propuesto') is None and file_manager.tiene_datos_parciales(m)]
|
||
vacios = [m for m in metadatos if m.get('archivo_propuesto') is None and not file_manager.tiene_datos_parciales(m)]
|
||
|
||
if not completos and not parciales:
|
||
print(f" │ {ic(False, '', 'Descartado: ningun archivo tiene datos utiles.')}")
|
||
file_manager.limpiar_carpeta_descarga(mail.ruta_descarga)
|
||
continue
|
||
|
||
# Renombrar y mover completos a resultado
|
||
if completos:
|
||
file_manager.renombrar_archivos_por_metadata(completos, mail.ruta_descarga)
|
||
file_manager.mover_validos_a_resultado(completos, mail.ruta_descarga, mail.ruta_final)
|
||
|
||
# Mover parciales a evaluar para revisión manual
|
||
if parciales:
|
||
ruta_evaluar = os.path.join(mail.ruta_descarga, "evaluar")
|
||
file_manager.mover_a_evaluar(parciales, mail.ruta_descarga, ruta_evaluar)
|
||
|
||
# Limpiar lo que queda (vacíos + sobrantes)
|
||
file_manager.limpiar_carpeta_descarga(mail.ruta_descarga)
|
||
|
||
for m in metadatos:
|
||
if m.get('archivo_propuesto') is not None:
|
||
if m.get('llm_usado'):
|
||
llm_resp = m.get('llm_respuesta') or {}
|
||
print(f" │ 🤖 {m['archivo']} -> [{m['tipo']}] RUC:{m['ruc']} S/N:{m['serie_numero']} | LLM: {llm_resp}")
|
||
else:
|
||
icono = "📕" if m['extension'] == ".pdf" else "🌐"
|
||
print(f" │ {icono} {m['archivo']} -> [{m['tipo']}] RUC:{m['ruc']} S/N:{m['serie_numero']}")
|
||
elif file_manager.tiene_datos_parciales(m):
|
||
print(f" │ 🔍 {m['archivo']} (metadata parcial -> evaluar/)")
|
||
else:
|
||
print(f" │ ❌ {m['archivo']} (sin datos utiles, eliminado)")
|
||
|
||
e['metadatos_adjuntos'] = metadatos
|
||
procesados.append(e)
|
||
|
||
except Exception as ex:
|
||
print(f" │ {ic(False, '', f'Error: {ex}')}")
|
||
file_manager.limpiar_carpeta_descarga(mail.ruta_descarga)
|
||
|
||
if llm_processor.CONTADOR_LLM > 0:
|
||
print(f" 🤖 LLM invocado {llm_processor.CONTADOR_LLM} vez/veces en esta ejecucion (modelo: {llm_processor.MODEL})")
|
||
|
||
return procesados
|
||
|
||
# ========================================================================
|
||
def guardar_json(todos_procesados):
|
||
if not todos_procesados:
|
||
return
|
||
|
||
output_dir = "output_json"
|
||
if not os.path.exists(output_dir):
|
||
os.makedirs(output_dir)
|
||
|
||
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
|
||
json_path = os.path.join(output_dir, f"{timestamp}.json")
|
||
|
||
with open(json_path, "w", encoding="utf-8") as f:
|
||
json.dump(todos_procesados, f, ensure_ascii=False, indent=4)
|
||
|
||
print(f"\n💾 Resultados exportados a: {json_path}")
|
||
|
||
# ========================================================================
|
||
def signal_handler(signum, frame):
|
||
global _running
|
||
print("\n\n⏹️ Senal de parada recibida. Deteniendo...")
|
||
_running = False
|
||
|
||
# ========================================================================
|
||
def main():
|
||
global _running
|
||
signal.signal(signal.SIGINT, signal_handler)
|
||
signal.signal(signal.SIGTERM, signal_handler)
|
||
|
||
log_path = setup_logger()
|
||
limpiar_logs_antiguos(dias=30)
|
||
|
||
print("=" * 60)
|
||
print("🚀 DAEMON DESCARGA ADJUNTOS CORREOS CPE")
|
||
print("=" * 60)
|
||
print(f"\n📅 Dias activos: {DIAS_ACTIVOS} (0=lun, 6=dom)")
|
||
print(f"⏰ Horarios:")
|
||
for hi, mi, hf, mf, intervalo, modo in HORARIOS:
|
||
print(f" {hi:02d}:{mi:02d} - {hf:02d}:{mf:02d} | cada {intervalo}s | {modo}")
|
||
print(f"\n📄 Log: {log_path}")
|
||
|
||
database_utilities.init_db()
|
||
|
||
hoy = datetime.now()
|
||
ayer = hoy - timedelta(days=5)
|
||
fecha_inicio = ayer.strftime("%Y-%m-%d")
|
||
fecha_fin = hoy.strftime("%Y-%m-%d")
|
||
|
||
outlook_token = None
|
||
fecha_obtencion_token = None
|
||
|
||
while _running:
|
||
rotar_log()
|
||
ahora = datetime.now()
|
||
|
||
if ahora.weekday() not in DIAS_ACTIVOS:
|
||
espera = segundos_hasta_proximo_dia_activo(ahora)
|
||
prox = ahora + timedelta(seconds=espera)
|
||
print(f"\n⏸️ Dia no activo. Proximo inicio: {prox.strftime('%a %Y-%m-%d %H:%M:%S')}")
|
||
for _ in range(int(espera)):
|
||
if not _running:
|
||
break
|
||
time.sleep(1)
|
||
continue
|
||
|
||
intervalo_info = obtener_intervalo(ahora)
|
||
if not intervalo_info:
|
||
espera = segundos_hasta_proximo_inicio(ahora)
|
||
prox = ahora + timedelta(seconds=espera)
|
||
print(f"\n⏸️ Fuera de horario. Proximo inicio: {prox.strftime('%H:%M:%S')}")
|
||
for _ in range(int(espera)):
|
||
if not _running:
|
||
break
|
||
time.sleep(1)
|
||
continue
|
||
|
||
intervalo, modo = intervalo_info
|
||
|
||
print(f"\n{'#'*60}")
|
||
print(f"# 🔄 CICLO: {ahora.strftime('%Y-%m-%d %H:%M:%S')} | modo={modo} | intervalo={intervalo}s")
|
||
print(f"{'#'*60}")
|
||
|
||
try:
|
||
outlook_token = obtener_token(outlook_token, fecha_obtencion_token)
|
||
if not fecha_obtencion_token:
|
||
fecha_obtencion_token = datetime.now()
|
||
except Exception as e:
|
||
print(f" {ic(False, '', f'Error obteniendo token: {e}')}")
|
||
time.sleep(60)
|
||
continue
|
||
|
||
todos_procesados = []
|
||
|
||
for mail in CONFIG.mails:
|
||
if not _running:
|
||
break
|
||
procesados = procesar_cuenta(mail, outlook_token, fecha_inicio, fecha_fin)
|
||
if procesados:
|
||
sincronizar_a_smb(mail.ruta_final, mail.smb_folder)
|
||
print(f"\n✅ {mail.email}: {len(procesados)} correos procesados.")
|
||
else:
|
||
print(f"\nℹ️ {mail.email}: sin correos para procesar.")
|
||
todos_procesados.extend(procesados)
|
||
|
||
guardar_json(todos_procesados)
|
||
|
||
print(f"\n⏳ Esperando {intervalo}s hasta proximo ciclo...")
|
||
for _ in range(intervalo):
|
||
if not _running:
|
||
break
|
||
time.sleep(1)
|
||
|
||
print("\n👋 Daemon detenido.")
|
||
|
||
# ========================================================================
|
||
if __name__ == "__main__":
|
||
main()
|