9f4edcb04a
Some checks failed
CI / ci (pull_request) Failing after 1m40s
- Refactor FirestoreNotificationBackend to use time-gated window (window_hours) instead of notified_by_agent status filtering; mark_as_notified is now a no-op (agent is awareness-only). - Update agent.py to instantiate FirestoreNotificationBackend using the shared firestore_db client instead of RedisNotificationBackend. - Remove redis_host/redis_port settings from config.py; add notifications_collection_path, max_to_notify, and window_hours. - Move redis/json imports inside RedisNotificationBackend methods so redis is only required if that backend is explicitly instantiated. - Add utility scripts for checking and registering notifications. - Add google-cloud-firestore dependency to pyproject.toml.
109 lines
3.2 KiB
Python
109 lines
3.2 KiB
Python
# /// script
|
|
# requires-python = ">=3.12"
|
|
# dependencies = ["redis>=5.0", "pydantic>=2.0"]
|
|
# ///
|
|
"""Check pending notifications for a phone number.
|
|
|
|
Usage:
|
|
REDIS_HOST=10.33.22.4 uv run utils/check_notifications.py <phone>
|
|
REDIS_HOST=10.33.22.4 uv run utils/check_notifications.py <phone> --since 2026-01-01
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import UTC, datetime
|
|
|
|
import redis
|
|
from pydantic import AliasChoices, BaseModel, Field, ValidationError
|
|
|
|
|
|
class Notification(BaseModel):
|
|
id_notificacion: str = Field(
|
|
validation_alias=AliasChoices("id_notificacion", "idNotificacion"),
|
|
)
|
|
telefono: str
|
|
timestamp_creacion: datetime = Field(
|
|
validation_alias=AliasChoices("timestamp_creacion", "timestampCreacion"),
|
|
)
|
|
texto: str
|
|
nombre_evento_dialogflow: str = Field(
|
|
validation_alias=AliasChoices(
|
|
"nombre_evento_dialogflow", "nombreEventoDialogflow"
|
|
),
|
|
)
|
|
codigo_idioma_dialogflow: str = Field(
|
|
default="es",
|
|
validation_alias=AliasChoices(
|
|
"codigo_idioma_dialogflow", "codigoIdiomaDialogflow"
|
|
),
|
|
)
|
|
parametros: dict = Field(default_factory=dict)
|
|
status: str
|
|
|
|
|
|
class NotificationSession(BaseModel):
|
|
session_id: str = Field(
|
|
validation_alias=AliasChoices("session_id", "sessionId"),
|
|
)
|
|
telefono: str
|
|
fecha_creacion: datetime = Field(
|
|
validation_alias=AliasChoices("fecha_creacion", "fechaCreacion"),
|
|
)
|
|
ultima_actualizacion: datetime = Field(
|
|
validation_alias=AliasChoices("ultima_actualizacion", "ultimaActualizacion"),
|
|
)
|
|
notificaciones: list[Notification]
|
|
|
|
|
|
HOST = os.environ.get("REDIS_HOST", "127.0.0.1")
|
|
PORT = int(os.environ.get("REDIS_PORT", "6379"))
|
|
|
|
|
|
def main() -> None:
|
|
if len(sys.argv) < 2:
|
|
print(f"Usage: {sys.argv[0]} <phone> [--since YYYY-MM-DD]")
|
|
sys.exit(1)
|
|
|
|
phone = sys.argv[1]
|
|
since = None
|
|
if "--since" in sys.argv:
|
|
idx = sys.argv.index("--since")
|
|
since = datetime.fromisoformat(sys.argv[idx + 1]).replace(tzinfo=UTC)
|
|
|
|
r = redis.Redis(host=HOST, port=PORT, decode_responses=True, socket_connect_timeout=5)
|
|
raw = r.get(f"notification:{phone}")
|
|
|
|
if not raw:
|
|
print(f"📭 No notifications found for {phone}")
|
|
sys.exit(0)
|
|
|
|
try:
|
|
session = NotificationSession.model_validate(json.loads(raw))
|
|
except ValidationError as e:
|
|
print(f"❌ Invalid notification data for {phone}:\n{e}")
|
|
sys.exit(1)
|
|
|
|
active = [n for n in session.notificaciones if n.status == "active"]
|
|
|
|
if since:
|
|
active = [n for n in active if n.timestamp_creacion >= since]
|
|
|
|
if not active:
|
|
print(f"📭 No {'new ' if since else ''}active notifications for {phone}")
|
|
sys.exit(0)
|
|
|
|
print(f"🔔 {len(active)} active notification(s) for {phone}\n")
|
|
for i, n in enumerate(active, 1):
|
|
categoria = n.parametros.get("notification_po_Categoria", "")
|
|
print(f" [{i}] {n.timestamp_creacion.isoformat()}")
|
|
print(f" ID: {n.id_notificacion}")
|
|
if categoria:
|
|
print(f" Category: {categoria}")
|
|
print(f" {n.texto[:120]}{'…' if len(n.texto) > 120 else ''}")
|
|
print()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|