diff --git a/pyproject.toml b/pyproject.toml index f66b064..b454aba 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,6 +14,7 @@ dependencies = [ "pydantic-settings[yaml]>=2.13.1", "google-auth>=2.34.0", "google-genai>=1.64.0", + "redis>=5.0", ] [build-system] diff --git a/src/va_agent/agent.py b/src/va_agent/agent.py index 07aa959..6369737 100644 --- a/src/va_agent/agent.py +++ b/src/va_agent/agent.py @@ -39,6 +39,7 @@ notification_service = FirestoreNotificationBackend( db=firestore_db, collection_path=settings.notifications_collection_path, max_to_notify=settings.notifications_max_to_notify, + window_hours=settings.notifications_window_hours, ) # Agent with static and dynamic instructions diff --git a/src/va_agent/config.py b/src/va_agent/config.py index ae33d2d..a4e202e 100644 --- a/src/va_agent/config.py +++ b/src/va_agent/config.py @@ -31,6 +31,7 @@ class AgentSettings(BaseSettings): "artifacts/bnt-orquestador-cognitivo-dev/notifications" ) notifications_max_to_notify: int = 5 + notifications_window_hours: float = 48 # MCP configuration mcp_audience: str diff --git a/src/va_agent/notifications.py b/src/va_agent/notifications.py index 958281a..e076f7f 100644 --- a/src/va_agent/notifications.py +++ b/src/va_agent/notifications.py @@ -30,7 +30,12 @@ class NotificationBackend(Protocol): class FirestoreNotificationBackend: - """Firestore-backed notification backend.""" + """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, @@ -38,18 +43,20 @@ class FirestoreNotificationBackend: 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_pending_notifications( self, phone_number: str ) -> list[dict[str, Any]]: - """Get pending notifications for a user. + """Get recent notifications for a user. - Retrieves notifications that have not been notified by the agent yet, + Retrieves notifications created within the configured time window, ordered by timestamp (most recent first), limited to max_to_notify. Args: @@ -83,26 +90,31 @@ class FirestoreNotificationBackend: 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) - ] + cutoff = time.time() - (self._window_hours * 3600) - if not pending: + def _ts(n: dict[str, Any]) -> Any: + return n.get( + "timestamp_creacion", + n.get("timestampCreacion", 0), + ) + + recent = [n for n in all_notifications if _ts(n) >= cutoff] + + if not recent: logger.info( - "All notifications already notified for phone: %s", phone_number + "No notifications within the last %.0fh for phone: %s", + self._window_hours, + phone_number, ) return [] - # Sort by timestamp_creacion (most recent first) - pending.sort(key=lambda n: n.get("timestamp_creacion", 0), reverse=True) + recent.sort(key=_ts, reverse=True) - # Return top N most recent - result = pending[: self._max_to_notify] + result = recent[: self._max_to_notify] logger.info( - "Found %d pending notifications for phone: %s (returning top %d)", - len(pending), + "Found %d recent notifications for phone: %s (returning top %d)", + len(recent), phone_number, len(result), ) @@ -116,80 +128,110 @@ class FirestoreNotificationBackend: return result async def mark_as_notified( - self, phone_number: str, notification_ids: list[str] + self, phone_number: str, notification_ids: list[str] # noqa: ARG002 ) -> bool: - """Mark notifications as notified by the agent. + """No-op — the agent is not the delivery mechanism.""" + return True - 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 +class RedisNotificationBackend: + """Redis-backed notification backend (read-only).""" - Returns: - True if update was successful, False otherwise + 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_pending_notifications( + self, phone_number: str + ) -> list[dict[str, Any]]: + """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*. """ - if not notification_ids: - return True + import json # noqa: PLC0415 try: - doc_ref = self._db.collection(self._collection_path).document(phone_number) - doc = await doc_ref.get() + raw = await self._client.get(f"notification:{phone_number}") - if not doc.exists: - logger.warning( - "Cannot mark notifications as notified: document not found for %s", + if not raw: + logger.info( + "No notification data in Redis for phone: %s", phone_number, ) - return False + return [] - 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, - } + data = json.loads(raw) + all_notifications: list[dict[str, Any]] = data.get( + "notificaciones", [] ) + if not all_notifications: + logger.info( + "No notifications in array for phone: %s", + phone_number, + ) + return [] + + cutoff = time.time() - (self._window_hours * 3600) + + def _ts(n: dict[str, Any]) -> Any: + return n.get( + "timestamp_creacion", + n.get("timestampCreacion", 0), + ) + + recent = [n for n in all_notifications if _ts(n) >= cutoff] + + if not recent: + logger.info( + "No notifications within the last %.0fh for phone: %s", + self._window_hours, + phone_number, + ) + return [] + + recent.sort(key=_ts, reverse=True) + + result = recent[: self._max_to_notify] + logger.info( - "Marked %d notification(s) as notified for phone: %s", - updated_count, + "Found %d recent notifications for phone: %s " + "(returning top %d)", + len(recent), phone_number, + len(result), ) except Exception: logger.exception( - "Failed to mark notifications as notified for phone: %s", + "Failed to fetch notifications from Redis for phone: %s", phone_number, ) - return False + return [] else: - return True + return result + + async def mark_as_notified( + self, phone_number: str, notification_ids: list[str] # noqa: ARG002 + ) -> bool: + """No-op — the agent is not the delivery mechanism.""" + return True diff --git a/utils/check_notifications.py b/utils/check_notifications.py new file mode 100644 index 0000000..b2d0629 --- /dev/null +++ b/utils/check_notifications.py @@ -0,0 +1,108 @@ +# /// 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 + REDIS_HOST=10.33.22.4 uv run utils/check_notifications.py --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]} [--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() diff --git a/utils/register_notification.py b/utils/register_notification.py new file mode 100644 index 0000000..d5df458 --- /dev/null +++ b/utils/register_notification.py @@ -0,0 +1,159 @@ +# /// script +# requires-python = ">=3.12" +# dependencies = ["redis>=5.0"] +# /// +"""Register a new notification in Redis for a given phone number. + +Usage: + REDIS_HOST=10.33.22.4 uv run utils/register_notification.py + +The notification content is randomly picked from a predefined set based on +existing entries in Memorystore. +""" + +import json +import os +import random +import sys +import uuid +from datetime import UTC, datetime + +import redis + +HOST = os.environ.get("REDIS_HOST", "127.0.0.1") +PORT = int(os.environ.get("REDIS_PORT", "6379")) +TTL_SECONDS = 18 * 24 * 3600 # ~18 days, matching existing keys + +NOTIFICATION_TEMPLATES = [ + { + "texto": ( + "Se detectó un cargo de $1,500 en tu cuenta" + ), + "parametros": { + "notification_po_transaction_id": "TXN15367", + "notification_po_amount": 5814, + }, + }, + { + "texto": ( + "💡 Recuerda que puedes obtener tu Adelanto de Nómina en cualquier" + " momento, sólo tienes que seleccionar Solicitud adelanto de Nómina" + " en tu app." + ), + "parametros": { + "notification_po_Categoria": "Adelanto de Nómina solicitud", + "notification_po_caption": "Adelanto de Nómina", + "notification_po_CTA": "Realiza la solicitud desde tu app", + "notification_po_Descripcion": ( + "Notificación para incentivar la solicitud de Adelanto de" + " Nómina desde la APP" + ), + "notification_po_link": ( + "https://public-media.yalochat.com/banorte/" + "1764025754-10e06fb8-b4e6-484c-ad0b-7f677429380e-03-ADN-Toque-1.jpg" + ), + "notification_po_Beneficios": ( + "Tasa de interés de 0%: Solicita tu Adelanto sin preocuparte" + " por los intereses, así de fácil. No requiere garantías o aval." + ), + "notification_po_Requisitos": ( + "Tener Cuenta Digital o Cuenta Digital Ilimitada con dispersión" + " de Nómina No tener otro Adelanto vigente Ingreso neto mensual" + " mayor a $2,000" + ), + }, + }, + { + "texto": ( + "Estás a un clic de Programa de Lealtad, entra a tu app y finaliza" + " Tu contratación en instantes. ⏱ 🤳" + ), + "parametros": { + "notification_po_Categoria": "Tarjeta de Crédito Contratación", + "notification_po_caption": "Tarjeta de Crédito", + "notification_po_CTA": "Entra a tu app y contrata en instantes", + "notification_po_Descripcion": ( + "Notificación para terminar el proceso de contratación de la" + " Tarjeta de Crédito, desde la app" + ), + "notification_po_link": ( + "https://public-media.yalochat.com/banorte/" + "1764363798-05dadc23-6e47-447c-8e38-0346f25e31c0-15-TDC-Toque-1.jpg" + ), + "notification_po_Beneficios": ( + "Acceso al Programa de Lealtad: Cada compra suma, gana" + " experiencias exclusivas" + ), + "notification_po_Requisitos": ( + "Ser persona física o física con actividad empresarial." + " Ingresos mínimos de $2,000 pesos mensuales. Sin historial de" + " crédito o con buró positivo" + ), + }, + }, + { + "texto": ( + "🚀 ¿Listo para obtener tu Cápsula Plus? Continúa en tu app y" + " termina al instante. Conoce más en: va.app" + ), + "parametros": {}, + }, + { + "texto": ( + "🚀 ¿Listo para obtener tu Cuenta Digital ilimitada? Continúa en" + " tu app y termina al instante. Conoce más en: va.app" + ), + "parametros": {}, + }, +] + + +def main() -> None: + if len(sys.argv) < 2: + print(f"Usage: {sys.argv[0]} ") + sys.exit(1) + + phone = sys.argv[1] + r = redis.Redis(host=HOST, port=PORT, decode_responses=True, socket_connect_timeout=5) + + now = datetime.now(UTC).isoformat() + template = random.choice(NOTIFICATION_TEMPLATES) + notification = { + "id_notificacion": str(uuid.uuid4()), + "telefono": phone, + "timestamp_creacion": now, + "texto": template["texto"], + "nombre_evento_dialogflow": "notificacion", + "codigo_idioma_dialogflow": "es", + "parametros": template["parametros"], + "status": "active", + } + + session_key = f"notification:{phone}" + existing = r.get(session_key) + + if existing: + session = json.loads(existing) + session["ultima_actualizacion"] = now + session["notificaciones"].append(notification) + else: + session = { + "session_id": phone, + "telefono": phone, + "fecha_creacion": now, + "ultima_actualizacion": now, + "notificaciones": [notification], + } + + r.set(session_key, json.dumps(session, ensure_ascii=False), ex=TTL_SECONDS) + r.set(f"notification:phone_to_notification:{phone}", phone, ex=TTL_SECONDS) + + total = len(session["notificaciones"]) + print(f"✅ Registered notification for {phone}") + print(f" ID: {notification['id_notificacion']}") + print(f" Text: {template['texto'][:80]}...") + print(f" Total notifications for this phone: {total}") + + +if __name__ == "__main__": + main() diff --git a/uv.lock b/uv.lock index 5598062..952f9fa 100644 --- a/uv.lock +++ b/uv.lock @@ -871,6 +871,7 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ea/ab/1608e5a7578e62113506740b88066bf09888322a311cff602105e619bd87/greenlet-3.3.2-cp312-cp312-macosx_11_0_universal2.whl", hash = "sha256:ac8d61d4343b799d1e526db579833d72f23759c71e07181c2d2944e429eb09cd", size = 280358, upload-time = "2026-02-20T20:17:43.971Z" }, { url = "https://files.pythonhosted.org/packages/a5/23/0eae412a4ade4e6623ff7626e38998cb9b11e9ff1ebacaa021e4e108ec15/greenlet-3.3.2-cp312-cp312-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:3ceec72030dae6ac0c8ed7591b96b70410a8be370b6a477b1dbc072856ad02bd", size = 601217, upload-time = "2026-02-20T20:47:31.462Z" }, { url = "https://files.pythonhosted.org/packages/f8/16/5b1678a9c07098ecb9ab2dd159fafaf12e963293e61ee8d10ecb55273e5e/greenlet-3.3.2-cp312-cp312-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:a2a5be83a45ce6188c045bcc44b0ee037d6a518978de9a5d97438548b953a1ac", size = 611792, upload-time = "2026-02-20T20:55:58.423Z" }, + { url = "https://files.pythonhosted.org/packages/5c/c5/cc09412a29e43406eba18d61c70baa936e299bc27e074e2be3806ed29098/greenlet-3.3.2-cp312-cp312-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:ae9e21c84035c490506c17002f5c8ab25f980205c3e61ddb3a2a2a2e6c411fcb", size = 626250, upload-time = "2026-02-20T21:02:46.596Z" }, { url = "https://files.pythonhosted.org/packages/50/1f/5155f55bd71cabd03765a4aac9ac446be129895271f73872c36ebd4b04b6/greenlet-3.3.2-cp312-cp312-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:43e99d1749147ac21dde49b99c9abffcbc1e2d55c67501465ef0930d6e78e070", size = 613875, upload-time = "2026-02-20T20:21:01.102Z" }, { url = "https://files.pythonhosted.org/packages/fc/dd/845f249c3fcd69e32df80cdab059b4be8b766ef5830a3d0aa9d6cad55beb/greenlet-3.3.2-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:4c956a19350e2c37f2c48b336a3afb4bff120b36076d9d7fb68cb44e05d95b79", size = 1571467, upload-time = "2026-02-20T20:49:33.495Z" }, { url = "https://files.pythonhosted.org/packages/2a/50/2649fe21fcc2b56659a452868e695634722a6655ba245d9f77f5656010bf/greenlet-3.3.2-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:6c6f8ba97d17a1e7d664151284cb3315fc5f8353e75221ed4324f84eb162b395", size = 1640001, upload-time = "2026-02-20T20:21:09.154Z" }, @@ -1625,6 +1626,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/1a/08/67bd04656199bbb51dbed1439b7f27601dfb576fb864099c7ef0c3e55531/pyyaml-6.0.3-cp312-cp312-win_arm64.whl", hash = "sha256:64386e5e707d03a7e172c0701abfb7e10f0fb753ee1d773128192742712a98fd", size = 140344, upload-time = "2025-09-25T21:32:22.617Z" }, ] +[[package]] +name = "redis" +version = "7.2.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/e9/31/1476f206482dd9bc53fdbbe9f6fbd5e05d153f18e54667ce839df331f2e6/redis-7.2.1.tar.gz", hash = "sha256:6163c1a47ee2d9d01221d8456bc1c75ab953cbda18cfbc15e7140e9ba16ca3a5", size = 4906735, upload-time = "2026-02-25T20:05:18.171Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ca/98/1dd1a5c060916cf21d15e67b7d6a7078e26e2605d5c37cbc9f4f5454c478/redis-7.2.1-py3-none-any.whl", hash = "sha256:49e231fbc8df2001436ae5252b3f0f3dc930430239bfeb6da4c7ee92b16e5d33", size = 396057, upload-time = "2026-02-25T20:05:16.533Z" }, +] + [[package]] name = "referencing" version = "0.37.0" @@ -1926,6 +1936,7 @@ dependencies = [ { name = "google-cloud-firestore" }, { name = "google-genai" }, { name = "pydantic-settings", extra = ["yaml"] }, + { name = "redis" }, ] [package.dev-dependencies] @@ -1944,6 +1955,7 @@ requires-dist = [ { name = "google-cloud-firestore", specifier = ">=2.23.0" }, { name = "google-genai", specifier = ">=1.64.0" }, { name = "pydantic-settings", extras = ["yaml"], specifier = ">=2.13.1" }, + { name = "redis", specifier = ">=5.0" }, ] [package.metadata.requires-dev]