""" poller.py — Loop de polling de notificaciones (multi-tenant). Responsabilidades: - Revisar el PJ de cada tenant activo cada N minutos en horario hábil - Calcular diff vs snapshot → solo notificar las nuevas - Manejar errores de conexión sin crashear ni detener otros tenants """ from __future__ import annotations import asyncio from collections.abc import Callable from datetime import datetime from pathlib import Path from zoneinfo import ZoneInfo import structlog import audit import vault from config import Settings from csj_client import ( CSJAuthError, CSJClient, CSJConnectionError, CSJPasswordTemporalError, ) from database import delete_tenant, get_tenant, list_active_tenants, save_tenant from storage import Snapshot logger = structlog.get_logger(__name__) PY_TZ = ZoneInfo("America/Asuncion") _MAX_ERRORES_CONSECUTIVOS = 5 class Poller: """ Revisa el PJ periódicamente para todos los tenants activos. """ def __init__( self, settings: Settings, bot: object, # PedritoBot — evitamos import circular on_poll_done: Callable[[int], None] | None = None, ) -> None: self._settings = settings self._bot = bot self._on_poll_done = on_poll_done self._running = False self._snapshots: dict[int, Snapshot] = {} self._errores_consecutivos: dict[int, int] = {} # Cola de revisiones manuales (chat_id) disparadas desde /notif self._manual_queue: asyncio.Queue[int] = asyncio.Queue() def _en_horario_habitual(self) -> bool: ahora = datetime.now(PY_TZ) if ahora.weekday() >= 5: # sábado o domingo return False return self._settings.poll_hora_inicio <= ahora.hour < self._settings.poll_hora_fin def _get_snapshot(self, chat_id: int) -> Snapshot: if chat_id not in self._snapshots: path: Path = self._settings.data_dir / f"snapshot_{chat_id}.json" self._snapshots[chat_id] = Snapshot(path) return self._snapshots[chat_id] # ────────────────────────────────────────────────────────────────────────── # Lógica de revisión por tenant # ────────────────────────────────────────────────────────────────────────── async def _revisar_tenant(self, tenant: dict, es_manual: bool = False) -> None: """Revisa notificaciones para UN tenant específico.""" chat_id: int = tenant["chat_id"] snapshot = self._get_snapshot(chat_id) credentials_enc = tenant.get("credentials_enc") if not credentials_enc: logger.warning("tenant_sin_credenciales_skip", chat_id=chat_id) return try: usuario, password = vault.descifrar_credenciales( chat_id, credentials_enc, self._settings.fernet_key ) except ValueError as e: logger.error("error_descifrar_credenciales_tenant", chat_id=chat_id, error=str(e)) return logger.info("iniciando_revision_tenant", chat_id=chat_id, es_manual=es_manual) await audit.log("polling_inicio", chat_id=chat_id, detalle={"es_manual": es_manual}, tenant=tenant) try: async with CSJClient(self._settings, usuario=usuario, password=password, chat_id=chat_id) as client: try: cantidad = await client.contar_notificaciones() logger.info("cantidad_notificaciones_tenant", chat_id=chat_id, cantidad=cantidad) if cantidad == 0: snapshot.registrar_check() if es_manual: await self._bot.enviar_sin_novedades(tenant, snapshot) # type: ignore[attr-defined] self._errores_consecutivos[chat_id] = 0 await audit.log("polling_tenant_ok", chat_id=chat_id, detalle={"nuevas": 0, "total": 0}, tenant=tenant) return notificaciones = await client.listar_notificaciones() nuevas = [n for n in notificaciones if not snapshot.ya_vista(n.cod_notificacion_origen)] logger.info( "diff_calculado_tenant", chat_id=chat_id, total=len(notificaciones), nuevas=len(nuevas), ) if nuevas: for notif in nuevas: await self._bot.enviar_notificacion(notif, tenant) # type: ignore[attr-defined] snapshot.marcar_vista(notif.cod_notificacion_origen) elif es_manual: await self._bot.enviar_sin_novedades(tenant, snapshot) # type: ignore[attr-defined] snapshot.registrar_check() self._errores_consecutivos[chat_id] = 0 await audit.log("polling_tenant_ok", chat_id=chat_id, detalle={"nuevas": len(nuevas), "total": len(notificaciones)}, tenant=tenant) except CSJPasswordTemporalError: logger.warning("password_temporal_tenant", chat_id=chat_id) await audit.log("polling_tenant_error", chat_id=chat_id, resultado="error", detalle={"tipo": "password_temporal"}, tenant=tenant) await self._bot.enviar_password_temporal(tenant) # type: ignore[attr-defined] await save_tenant(chat_id, estado="suspendido") self._snapshots.pop(chat_id, None) except CSJAuthError as e: logger.error("error_auth_tenant", chat_id=chat_id, error=str(e)) await audit.log("polling_tenant_error", chat_id=chat_id, resultado="error", detalle={"tipo": "auth"}, tenant=tenant) await self._bot.enviar_error_auth(tenant) # type: ignore[attr-defined] await save_tenant(chat_id, estado="suspendido") self._snapshots.pop(chat_id, None) except CSJConnectionError as e: n = self._errores_consecutivos.get(chat_id, 0) + 1 self._errores_consecutivos[chat_id] = n logger.warning( "error_conexion_transitorio_tenant", chat_id=chat_id, error=str(e), consecutivos=n, ) await audit.log("pj_error_conexion", chat_id=chat_id, resultado="error", detalle={"consecutivos": n}, tenant=tenant) if n >= _MAX_ERRORES_CONSECUTIVOS: await self._bot.enviar_error_conexion(str(e), tenant) # type: ignore[attr-defined] self._errores_consecutivos[chat_id] = 0 except Exception as e: logger.exception("error_inesperado_tenant", chat_id=chat_id, error=str(e)) await audit.log("polling_tenant_error", chat_id=chat_id, resultado="error", detalle={"tipo": "inesperado"}, tenant=tenant) self._errores_consecutivos[chat_id] = self._errores_consecutivos.get(chat_id, 0) + 1 finally: del usuario, password # ────────────────────────────────────────────────────────────────────────── # API pública # ────────────────────────────────────────────────────────────────────────── async def revisar_ahora(self, chat_id: int | None = None, es_manual: bool = False) -> int: """ Si chat_id: revisar solo ese tenant (para /notif manual). Si None: revisar todos los tenants activos. Retorna la cantidad de tenants revisados. """ if chat_id is not None: tenant = await get_tenant(chat_id) if tenant and tenant.get("estado") == "activo": await self._revisar_tenant(tenant, es_manual=es_manual) return 1 elif es_manual and tenant: logger.warning("revisar_ahora_tenant_no_activo", chat_id=chat_id, estado=tenant.get("estado")) return 0 else: tenants = await list_active_tenants() if not tenants: logger.info("sin_tenants_activos_skip") return 0 await asyncio.gather(*(self._revisar_tenant(t) for t in tenants)) return len(tenants) def disparar_revision_manual(self, chat_id: int) -> None: """Llamado desde el handler de /notif para disparar revisión inmediata.""" self._manual_queue.put_nowait(chat_id) def detener(self) -> None: self._running = False async def run_forever(self) -> None: """Loop principal. Revisa todos los tenants cada poll_interval_minutes en horario hábil.""" self._running = True logger.info( "poller_iniciado", interval_minutes=self._settings.poll_interval_minutes, hora_inicio=self._settings.poll_hora_inicio, hora_fin=self._settings.poll_hora_fin, ) if self._en_horario_habitual(): logger.info("revision_inicial_al_arrancar") count = await self.revisar_ahora() if self._on_poll_done: self._on_poll_done(count) else: logger.info("revision_inicial_skipped_fuera_de_horario") while self._running: try: chat_id = await asyncio.wait_for( self._manual_queue.get(), timeout=self._settings.poll_interval_minutes * 60, ) await self.revisar_ahora(chat_id=chat_id, es_manual=True) except asyncio.TimeoutError: if self._en_horario_habitual(): count = await self.revisar_ahora() if self._on_poll_done: self._on_poll_done(count) else: logger.debug("fuera_de_horario_skip")