""" 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 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 ) -> None: self._settings = settings self._bot = bot 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) -> None: """ Si chat_id: revisar solo ese tenant (para /notif manual). Si None: revisar todos los tenants activos. """ 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) elif es_manual and tenant: logger.warning("revisar_ahora_tenant_no_activo", chat_id=chat_id, estado=tenant.get("estado")) else: tenants = await list_active_tenants() if not tenants: logger.info("sin_tenants_activos_skip") return await asyncio.gather(*(self._revisar_tenant(t) for t in 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, ) logger.info("revision_inicial_al_arrancar") await self.revisar_ahora() 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(): await self.revisar_ahora() else: logger.debug("fuera_de_horario_skip")