95 lines
2.8 KiB
Python
95 lines
2.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""vistos.py — Estado persistente de "ya visto" por fuente (RF-8, RF-23).
|
|
|
|
Guarda, por cada fuente, el conjunto de identificadores ya procesados para no
|
|
volver a contarlos ni a avisar por ellos en el siguiente ciclo. El estado se
|
|
persiste en `estado/monitor/visto/<clave>.json` y sobrevive a reinicios.
|
|
|
|
Los identificadores pueden ser enteros (Telegram: id de mensaje) o cadenas
|
|
(correo: id de mensaje de Gmail; Jira: clave del ticket). Internamente se
|
|
normalizan a texto.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Callable, Iterable
|
|
|
|
from . import config
|
|
|
|
VISTO_DIR: Path = config.MONITOR_DIR / "visto"
|
|
|
|
# Limite de ids guardados por fuente (ventana movil; evita ficheros infinitos).
|
|
MAX_IDS = 2000
|
|
|
|
|
|
def _nombre_fichero(clave: str) -> str:
|
|
"""Nombre de fichero seguro para la clave de la fuente."""
|
|
seguro = re.sub(r"[^A-Za-z0-9_.-]", "_", clave)
|
|
return f"{seguro}.json"
|
|
|
|
|
|
def _ruta(clave: str) -> Path:
|
|
return VISTO_DIR / _nombre_fichero(clave)
|
|
|
|
|
|
def _cargar(clave: str) -> dict:
|
|
"""Carga el estado de la fuente (lista de ids vistos)."""
|
|
ruta = _ruta(clave)
|
|
if ruta.exists():
|
|
try:
|
|
with open(ruta, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
if isinstance(data, dict):
|
|
return data
|
|
except Exception:
|
|
pass
|
|
return {"ids_vistos": []}
|
|
|
|
|
|
def _guardar(clave: str, data: dict) -> None:
|
|
"""Persiste el estado de la fuente (recortando a MAX_IDS)."""
|
|
VISTO_DIR.mkdir(parents=True, exist_ok=True)
|
|
ids = data.get("ids_vistos", [])
|
|
data["ids_vistos"] = ids[-MAX_IDS:]
|
|
with open(_ruta(clave), "w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=2, ensure_ascii=False)
|
|
|
|
|
|
def _norm(x) -> str:
|
|
return str(x)
|
|
|
|
|
|
def marcar(clave: str, ids: Iterable) -> None:
|
|
"""Añade ids al estado de la fuente y persiste."""
|
|
data = _cargar(clave)
|
|
vistos = set(data.get("ids_vistos", []))
|
|
nuevos = {_norm(i) for i in ids} - vistos
|
|
if not nuevos:
|
|
return
|
|
data["ids_vistos"] = list(vistos | nuevos)
|
|
_guardar(clave, data)
|
|
|
|
|
|
def filtrar_nuevos(
|
|
clave: str,
|
|
elementos: Iterable,
|
|
id_fn: Callable = lambda e: e,
|
|
marcar_automatico: bool = True,
|
|
) -> list:
|
|
"""Devuelve los elementos cuyo id no estaba visto.
|
|
|
|
Si `marcar_automatico` es True (por defecto), los ids nuevos quedan marcados
|
|
en el mismo paso, de modo que el siguiente ciclo no los repita.
|
|
"""
|
|
elementos = list(elementos)
|
|
vistos = set(_cargar(clave).get("ids_vistos", []))
|
|
nuevos = [e for e in elementos if _norm(id_fn(e)) not in vistos]
|
|
if marcar_automatico and nuevos:
|
|
data = _cargar(clave)
|
|
data["ids_vistos"] = list(vistos | {_norm(id_fn(e)) for e in nuevos})
|
|
_guardar(clave, data)
|
|
return nuevos
|