7743919695
Bot de Telegram para monitoreo de notificaciones judiciales del PJ Paraguay. Multi-tenant con onboarding conversacional, audit log, y polling automático. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
77 lines
2.5 KiB
Python
77 lines
2.5 KiB
Python
"""
|
|
storage.py — Persistencia del snapshot de notificaciones.
|
|
|
|
Guarda en JSON local en data/snapshot.json.
|
|
Simple a propósito: en v1 esto se reemplaza por Postgres con el mismo interface.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import structlog
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
|
|
class Snapshot:
|
|
"""
|
|
Snapshot del estado de notificaciones.
|
|
Guarda los IDs de las notificaciones ya notificadas al abogado.
|
|
"""
|
|
|
|
def __init__(self, path: Path) -> None:
|
|
self._path = path
|
|
self._notificaciones_vistas: set[int] = set()
|
|
self._last_check: datetime | None = None
|
|
self._load()
|
|
|
|
def _load(self) -> None:
|
|
if not self._path.exists():
|
|
logger.info("snapshot_no_existe_iniciando_vacio", path=str(self._path))
|
|
return
|
|
|
|
try:
|
|
data = json.loads(self._path.read_text(encoding="utf-8"))
|
|
self._notificaciones_vistas = set(data.get("notificaciones_vistas", []))
|
|
last_check_str = data.get("last_check")
|
|
if last_check_str:
|
|
self._last_check = datetime.fromisoformat(last_check_str)
|
|
logger.info(
|
|
"snapshot_cargado",
|
|
notificaciones_vistas=len(self._notificaciones_vistas),
|
|
last_check=last_check_str,
|
|
)
|
|
except (json.JSONDecodeError, ValueError) as e:
|
|
logger.warning("snapshot_corrupto_iniciando_vacio", error=str(e))
|
|
|
|
def _save(self) -> None:
|
|
data = {
|
|
"last_check": self._last_check.isoformat() if self._last_check else None,
|
|
"notificaciones_vistas": sorted(self._notificaciones_vistas),
|
|
}
|
|
self._path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
def ya_vista(self, cod_notificacion: int) -> bool:
|
|
return cod_notificacion in self._notificaciones_vistas
|
|
|
|
def marcar_vista(self, cod_notificacion: int) -> None:
|
|
self._notificaciones_vistas.add(cod_notificacion)
|
|
|
|
def marcar_varias_vistas(self, codigos: list[int]) -> None:
|
|
self._notificaciones_vistas.update(codigos)
|
|
|
|
def registrar_check(self) -> None:
|
|
self._last_check = datetime.now(timezone.utc)
|
|
self._save()
|
|
|
|
@property
|
|
def last_check(self) -> datetime | None:
|
|
return self._last_check
|
|
|
|
@property
|
|
def total_vistas(self) -> int:
|
|
return len(self._notificaciones_vistas)
|