257 lines
12 KiB
Python
257 lines
12 KiB
Python
"""
|
|
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,
|
|
CSJDocumentoNoDisponibleError,
|
|
CSJPasswordTemporalError,
|
|
)
|
|
from pdf_extractor import extraer_extracto
|
|
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]
|
|
|
|
async def _intentar_extracto(
|
|
self,
|
|
client: CSJClient,
|
|
notif: object,
|
|
) -> str | None:
|
|
"""Descarga el PDF de la notificación y extrae el texto. Degradación elegante: None si falla."""
|
|
from csj_client import NotificacionPendiente # evita import circular en type hint
|
|
assert isinstance(notif, NotificacionPendiente)
|
|
if not (notif.cod_caso_judicial and notif.cod_actuacion_caso and notif.origen is not None):
|
|
return None
|
|
try:
|
|
pdf_bytes = await client.descargar_documento_principal(
|
|
cod_caso_judicial=notif.cod_caso_judicial,
|
|
origen_caso=notif.origen,
|
|
cod_actuacion_caso=notif.cod_actuacion_caso,
|
|
origen_actuacion=notif.origen,
|
|
)
|
|
return extraer_extracto(pdf_bytes)
|
|
except (CSJDocumentoNoDisponibleError, CSJConnectionError, CSJAuthError):
|
|
return None
|
|
except Exception:
|
|
return None
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────
|
|
# 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:
|
|
extracto = await self._intentar_extracto(client, notif)
|
|
await self._bot.enviar_notificacion(notif, tenant, extracto=extracto) # 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")
|