Add notification service using Google ADK
This commit is contained in:
232
src/va_agent/notifications.py
Normal file
232
src/va_agent/notifications.py
Normal file
@@ -0,0 +1,232 @@
|
||||
"""Notification management for VAia agent."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import time
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from google.cloud.firestore_v1.async_client import AsyncClient
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class NotificationService:
|
||||
"""Service for fetching and managing user notifications from Firestore."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
db: AsyncClient,
|
||||
collection_path: str,
|
||||
max_to_notify: int = 5,
|
||||
) -> None:
|
||||
"""Initialize NotificationService.
|
||||
|
||||
Args:
|
||||
db: Firestore async client
|
||||
collection_path: Path to notifications collection
|
||||
max_to_notify: Maximum number of notifications to return
|
||||
|
||||
"""
|
||||
self._db = db
|
||||
self._collection_path = collection_path
|
||||
self._max_to_notify = max_to_notify
|
||||
|
||||
async def get_pending_notifications(
|
||||
self, phone_number: str
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Get pending notifications for a user.
|
||||
|
||||
Retrieves notifications that have not been notified by the agent yet,
|
||||
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 notification dictionaries with structure:
|
||||
{
|
||||
"id_notificacion": str,
|
||||
"texto": str,
|
||||
"status": str,
|
||||
"timestamp_creacion": timestamp,
|
||||
"parametros": {...}
|
||||
}
|
||||
|
||||
"""
|
||||
try:
|
||||
# Query Firestore document by phone number
|
||||
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 {}
|
||||
all_notifications = data.get("notificaciones", [])
|
||||
|
||||
if not all_notifications:
|
||||
logger.info("No notifications in array for phone: %s", phone_number)
|
||||
return []
|
||||
|
||||
# Filter notifications that have NOT been notified by the agent
|
||||
pending = [
|
||||
n
|
||||
for n in all_notifications
|
||||
if not n.get("notified_by_agent", False)
|
||||
]
|
||||
|
||||
if not pending:
|
||||
logger.info(
|
||||
"All notifications already notified for phone: %s", phone_number
|
||||
)
|
||||
return []
|
||||
|
||||
# Sort by timestamp_creacion (most recent first)
|
||||
pending.sort(
|
||||
key=lambda n: n.get("timestamp_creacion", 0), reverse=True
|
||||
)
|
||||
|
||||
# Return top N most recent
|
||||
result = pending[: self._max_to_notify]
|
||||
|
||||
logger.info(
|
||||
"Found %d pending notifications for phone: %s (returning top %d)",
|
||||
len(pending),
|
||||
phone_number,
|
||||
len(result),
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"Failed to fetch notifications for phone: %s", phone_number
|
||||
)
|
||||
return []
|
||||
|
||||
async def mark_as_notified(
|
||||
self, phone_number: str, notification_ids: list[str]
|
||||
) -> bool:
|
||||
"""Mark notifications as notified by the agent.
|
||||
|
||||
Updates the notifications in Firestore by adding:
|
||||
- notified_by_agent: true
|
||||
- notified_at: current timestamp
|
||||
|
||||
Args:
|
||||
phone_number: User's phone number (document ID)
|
||||
notification_ids: List of id_notificacion values to mark
|
||||
|
||||
Returns:
|
||||
True if update was successful, False otherwise
|
||||
|
||||
"""
|
||||
if not notification_ids:
|
||||
return True
|
||||
|
||||
try:
|
||||
doc_ref = self._db.collection(self._collection_path).document(
|
||||
phone_number
|
||||
)
|
||||
doc = await doc_ref.get()
|
||||
|
||||
if not doc.exists:
|
||||
logger.warning(
|
||||
"Cannot mark notifications as notified: document not found for %s",
|
||||
phone_number,
|
||||
)
|
||||
return False
|
||||
|
||||
data = doc.to_dict() or {}
|
||||
notificaciones = data.get("notificaciones", [])
|
||||
|
||||
if not notificaciones:
|
||||
logger.warning(
|
||||
"Cannot mark notifications: empty array for %s", phone_number
|
||||
)
|
||||
return False
|
||||
|
||||
# Update matching notifications
|
||||
now = time.time()
|
||||
updated_count = 0
|
||||
|
||||
for notif in notificaciones:
|
||||
if notif.get("id_notificacion") in notification_ids:
|
||||
notif["notified_by_agent"] = True
|
||||
notif["notified_at"] = now
|
||||
updated_count += 1
|
||||
|
||||
if updated_count == 0:
|
||||
logger.warning(
|
||||
"No notifications matched IDs for phone: %s", phone_number
|
||||
)
|
||||
return False
|
||||
|
||||
# Save back to Firestore
|
||||
await doc_ref.update(
|
||||
{
|
||||
"notificaciones": notificaciones,
|
||||
"ultima_actualizacion": now,
|
||||
}
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"Marked %d notification(s) as notified for phone: %s",
|
||||
updated_count,
|
||||
phone_number,
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"Failed to mark notifications as notified for phone: %s",
|
||||
phone_number,
|
||||
)
|
||||
return False
|
||||
|
||||
def format_notification_summary(
|
||||
self, notifications: list[dict[str, Any]]
|
||||
) -> str:
|
||||
"""Format notifications into a human-readable summary.
|
||||
|
||||
Args:
|
||||
notifications: List of notification dictionaries
|
||||
|
||||
Returns:
|
||||
Formatted string summarizing the notifications
|
||||
|
||||
"""
|
||||
if not notifications:
|
||||
return ""
|
||||
|
||||
count = len(notifications)
|
||||
summary_lines = [
|
||||
f"El usuario tiene {count} notificación(es) pendiente(s):"
|
||||
]
|
||||
|
||||
for i, notif in enumerate(notifications, 1):
|
||||
texto = notif.get("texto", "Sin texto")
|
||||
params = notif.get("parametros", {})
|
||||
|
||||
# Extract key parameters if available
|
||||
amount = params.get("notification_po_amount")
|
||||
tx_id = params.get("notification_po_transaction_id")
|
||||
|
||||
line = f"{i}. {texto}"
|
||||
if amount:
|
||||
line += f" (monto: ${amount})"
|
||||
if tx_id:
|
||||
line += f" [ID: {tx_id}]"
|
||||
|
||||
summary_lines.append(line)
|
||||
|
||||
return "\n".join(summary_lines)
|
||||
Reference in New Issue
Block a user