136 lines
4.0 KiB
Python
136 lines
4.0 KiB
Python
import os
|
|
import time
|
|
import argparse
|
|
from datetime import datetime, timedelta
|
|
from dotenv import load_dotenv
|
|
|
|
import logger_utilities
|
|
from pipeline import run_pipeline
|
|
|
|
# ============================================================================
|
|
|
|
HORARIOS = [
|
|
(6, 30, 13, 15, 300, "rapido"), # 06:30 - 13:15 -> cada 5 minutos
|
|
(13, 15, 14, 0, 1800, "full"), # 13:15 - 14:00 -> cada 30 minutos
|
|
(14, 0, 18, 30, 300, "rapido"), # 14:00 - 17:30 -> cada 5 minutos
|
|
(18, 30, 21, 30, 3600, "full"), # 17:30 - 21:30 -> cada 60 minutos
|
|
(0, 0, 5, 0, 60, "rapido"), # 00:00 - 05:00 -> cada 5 minutos
|
|
]
|
|
# ============================================================================
|
|
def obtener_intervalo(dt: datetime) -> tuple[int, str] | None:
|
|
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: datetime) -> int:
|
|
inicio = dt.replace(hour=6, minute=30, second=0, microsecond=0)
|
|
|
|
if dt >= inicio:
|
|
inicio += timedelta(days=1)
|
|
|
|
return int((inicio - dt).total_seconds())
|
|
|
|
# ============================================================================
|
|
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 main() -> None:
|
|
|
|
parser = argparse.ArgumentParser(
|
|
description="Obtener correos electrónicos de Office365 de hoy y los últimos X días."
|
|
)
|
|
|
|
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)."
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
dias = args.dias
|
|
|
|
load_dotenv(dotenv_path=".env")
|
|
|
|
EMAIL = os.getenv("EMAIL", "")
|
|
DOWNLOAD_FOLDER = os.getenv("DOWNLOAD_FOLDER", "")
|
|
OUTPUT_FOLDER = os.getenv("OUTPUT_FOLDER", "")
|
|
|
|
print("🚀 Demonio iniciado")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
while True:
|
|
|
|
inicio_ciclo = datetime.now()
|
|
|
|
resultado = obtener_intervalo(inicio_ciclo)
|
|
|
|
if resultado is not None:
|
|
|
|
intervalo, modo = resultado
|
|
|
|
print(
|
|
f"\n[{inicio_ciclo:%Y-%m-%d %H:%M:%S}] "
|
|
f"Iniciando ciclo [{modo}] (cada {intervalo}s)"
|
|
)
|
|
|
|
run_pipeline(dias, EMAIL, DOWNLOAD_FOLDER, OUTPUT_FOLDER)
|
|
|
|
duracion = (datetime.now() - inicio_ciclo).total_seconds()
|
|
|
|
if duracion > intervalo:
|
|
print(
|
|
f"⚠️ El ciclo demoró {duracion:.1f}s "
|
|
f"(intervalo configurado {intervalo}s)"
|
|
)
|
|
|
|
dormir = max(0, intervalo - duracion)
|
|
|
|
print(f"💤 Próximo ciclo [{modo}] en {dormir:.0f}s...")
|
|
|
|
else:
|
|
|
|
dormir = segundos_hasta_proximo_inicio(inicio_ciclo)
|
|
|
|
despertar = inicio_ciclo + timedelta(seconds=dormir)
|
|
|
|
print(
|
|
f"\n[{inicio_ciclo:%Y-%m-%d %H:%M:%S}] "
|
|
f"Fuera del horario.\n"
|
|
f"Próxima ejecución: {despertar:%Y-%m-%d %H:%M:%S}"
|
|
)
|
|
|
|
time.sleep(dormir)
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n\n🛑 Demonio detenido por el usuario.")
|
|
|
|
# ============================================================================
|
|
|
|
if __name__ == "__main__":
|
|
log_path = logger_utilities.setup_logger()
|
|
logger_utilities.limpiar_logs_antiguos(dias=30)
|
|
|
|
print("=" * 60)
|
|
print(f"🚀 Demonio iniciado: {datetime.now():%Y-%m-%d %H:%M:%S}")
|
|
print(f"📄 Log guardado en: {log_path}")
|
|
print("=" * 60)
|
|
|
|
main()
|