Anibal Angulo
ac27d12ed3
All checks were successful
CI / ci (push) Successful in 21s
Co-authored-by: Anibal Angulo <a8065384@banorte.com> Reviewed-on: #31
279 lines
8.5 KiB
Python
279 lines
8.5 KiB
Python
"""Notification management for VAia agent."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from datetime import datetime
|
|
from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
|
|
|
|
from pydantic import AliasChoices, BaseModel, Field, field_validator
|
|
|
|
if TYPE_CHECKING:
|
|
from google.cloud.firestore_v1.async_client import AsyncClient
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Notification(BaseModel):
|
|
"""A single notification, normalised from either schema.
|
|
|
|
Handles snake_case (``id_notificacion``), camelCase
|
|
(``idNotificacion``), and English short names (``notificationId``)
|
|
transparently via ``AliasChoices``.
|
|
"""
|
|
|
|
id_notificacion: str = Field(
|
|
validation_alias=AliasChoices(
|
|
"id_notificacion", "idNotificacion", "notificationId"
|
|
),
|
|
)
|
|
texto: str = Field(
|
|
default="Sin texto",
|
|
validation_alias=AliasChoices("texto", "text"),
|
|
)
|
|
nombre_evento: str = Field(
|
|
default="notificacion",
|
|
validation_alias=AliasChoices(
|
|
"nombre_evento_dialogflow", "nombreEventoDialogflow", "event"
|
|
),
|
|
)
|
|
timestamp_creacion: float = Field(
|
|
default=0.0,
|
|
validation_alias=AliasChoices("timestamp_creacion", "timestampCreacion"),
|
|
)
|
|
status: str = "active"
|
|
parametros: dict[str, Any] = Field(
|
|
default_factory=dict,
|
|
validation_alias=AliasChoices("parametros", "parameters"),
|
|
)
|
|
|
|
@field_validator("timestamp_creacion", mode="before")
|
|
@classmethod
|
|
def _coerce_timestamp(cls, v: Any) -> float:
|
|
"""Normalise Firestore timestamps (float, str, datetime) to float."""
|
|
if isinstance(v, (int, float)):
|
|
return float(v)
|
|
if isinstance(v, datetime):
|
|
return v.timestamp()
|
|
if isinstance(v, str):
|
|
try:
|
|
return float(v)
|
|
except ValueError:
|
|
return 0.0
|
|
return 0.0
|
|
|
|
|
|
class NotificationDocument(BaseModel):
|
|
"""Top-level Firestore / Redis document that wraps a list of notifications.
|
|
|
|
Mirrors the schema used by ``utils/check_notifications.py``
|
|
(``NotificationSession``) but keeps only what the agent needs.
|
|
"""
|
|
|
|
notificaciones: list[Notification] = Field(default_factory=list)
|
|
|
|
|
|
@runtime_checkable
|
|
class NotificationBackend(Protocol):
|
|
"""Backend-agnostic interface for notification storage."""
|
|
|
|
async def get_recent_notifications(self, phone_number: str) -> list[Notification]:
|
|
"""Return recent notifications for *phone_number*."""
|
|
...
|
|
|
|
async def mark_as_notified(
|
|
self, phone_number: str, notification_ids: list[str]
|
|
) -> bool:
|
|
"""Mark the given notification IDs as notified. Return success."""
|
|
...
|
|
|
|
|
|
class FirestoreNotificationBackend:
|
|
"""Firestore-backed notification backend (read-only).
|
|
|
|
Reads notifications from a Firestore document keyed by phone number.
|
|
Filters by a configurable time window instead of tracking read/unread
|
|
state — the agent is awareness-only; delivery happens in the app.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
db: AsyncClient,
|
|
collection_path: str,
|
|
max_to_notify: int = 5,
|
|
window_hours: float = 48,
|
|
) -> None:
|
|
"""Initialize with Firestore client and collection path."""
|
|
self._db = db
|
|
self._collection_path = collection_path
|
|
self._max_to_notify = max_to_notify
|
|
self._window_hours = window_hours
|
|
|
|
async def get_recent_notifications(self, phone_number: str) -> list[Notification]:
|
|
"""Get recent notifications for a user.
|
|
|
|
Retrieves notifications created within the configured time window,
|
|
ordered by timestamp (most recent first), limited to max_to_notify.
|
|
|
|
Args:
|
|
phone_number: User's phone number (used as document ID)
|
|
|
|
Returns:
|
|
List of validated :class:`Notification` instances.
|
|
|
|
"""
|
|
try:
|
|
doc_ref = self._db.collection(self._collection_path).document(phone_number)
|
|
doc = await doc_ref.get()
|
|
|
|
if not doc.exists:
|
|
logger.info(
|
|
"No notification document found for phone: %s", phone_number
|
|
)
|
|
return []
|
|
|
|
data = doc.to_dict() or {}
|
|
document = NotificationDocument.model_validate(data)
|
|
|
|
if not document.notificaciones:
|
|
logger.info("No notifications in array for phone: %s", phone_number)
|
|
return []
|
|
|
|
cutoff = time.time() - (self._window_hours * 3600)
|
|
|
|
parsed = [
|
|
n for n in document.notificaciones if n.timestamp_creacion >= cutoff
|
|
]
|
|
|
|
if not parsed:
|
|
logger.info(
|
|
"No notifications within the last %.0fh for phone: %s",
|
|
self._window_hours,
|
|
phone_number,
|
|
)
|
|
return []
|
|
|
|
parsed.sort(key=lambda n: n.timestamp_creacion, reverse=True)
|
|
|
|
result = parsed[: self._max_to_notify]
|
|
|
|
logger.info(
|
|
"Found %d recent notifications for phone: %s (returning top %d)",
|
|
len(parsed),
|
|
phone_number,
|
|
len(result),
|
|
)
|
|
|
|
except Exception:
|
|
logger.exception(
|
|
"Failed to fetch notifications for phone: %s", phone_number
|
|
)
|
|
return []
|
|
else:
|
|
return result
|
|
|
|
async def mark_as_notified(
|
|
self,
|
|
phone_number: str, # noqa: ARG002
|
|
notification_ids: list[str], # noqa: ARG002
|
|
) -> bool:
|
|
"""No-op — the agent is not the delivery mechanism."""
|
|
return True
|
|
|
|
|
|
class RedisNotificationBackend:
|
|
"""Redis-backed notification backend (read-only)."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
host: str = "127.0.0.1",
|
|
port: int = 6379,
|
|
max_to_notify: int = 5,
|
|
window_hours: float = 48,
|
|
) -> None:
|
|
"""Initialize with Redis connection parameters."""
|
|
import redis.asyncio as aioredis # noqa: PLC0415
|
|
|
|
self._client = aioredis.Redis(
|
|
host=host,
|
|
port=port,
|
|
decode_responses=True,
|
|
socket_connect_timeout=5,
|
|
)
|
|
self._max_to_notify = max_to_notify
|
|
self._window_hours = window_hours
|
|
|
|
async def get_recent_notifications(self, phone_number: str) -> list[Notification]:
|
|
"""Get recent notifications for a user from Redis.
|
|
|
|
Reads from the ``notification:{phone}`` key, parses the JSON
|
|
payload, and returns notifications created within the configured
|
|
time window, sorted by creation timestamp (most recent first),
|
|
limited to *max_to_notify*.
|
|
"""
|
|
import json # noqa: PLC0415
|
|
|
|
try:
|
|
raw = await self._client.get(f"notification:{phone_number}")
|
|
|
|
if not raw:
|
|
logger.info(
|
|
"No notification data in Redis for phone: %s",
|
|
phone_number,
|
|
)
|
|
return []
|
|
|
|
document = NotificationDocument.model_validate(json.loads(raw))
|
|
|
|
if not document.notificaciones:
|
|
logger.info(
|
|
"No notifications in array for phone: %s",
|
|
phone_number,
|
|
)
|
|
return []
|
|
|
|
cutoff = time.time() - (self._window_hours * 3600)
|
|
|
|
parsed = [
|
|
n for n in document.notificaciones if n.timestamp_creacion >= cutoff
|
|
]
|
|
|
|
if not parsed:
|
|
logger.info(
|
|
"No notifications within the last %.0fh for phone: %s",
|
|
self._window_hours,
|
|
phone_number,
|
|
)
|
|
return []
|
|
|
|
parsed.sort(key=lambda n: n.timestamp_creacion, reverse=True)
|
|
|
|
result = parsed[: self._max_to_notify]
|
|
|
|
logger.info(
|
|
"Found %d recent notifications for phone: %s (returning top %d)",
|
|
len(parsed),
|
|
phone_number,
|
|
len(result),
|
|
)
|
|
|
|
except Exception:
|
|
logger.exception(
|
|
"Failed to fetch notifications from Redis for phone: %s",
|
|
phone_number,
|
|
)
|
|
return []
|
|
else:
|
|
return result
|
|
|
|
async def mark_as_notified(
|
|
self,
|
|
phone_number: str, # noqa: ARG002
|
|
notification_ids: list[str], # noqa: ARG002
|
|
) -> bool:
|
|
"""No-op — the agent is not the delivery mechanism."""
|
|
return True
|