"""Cliente mínimo de Noti para empresas. Un solo archivo, sin dependencias (Python 3.9 o más nuevo). Copialo a tu proyecto: no hace falta instalar nada. https://noti.gualda.ai/empresas/integrar la guía (y el prompt para tu agente) https://noti.gualda.ai/sdk/noti.py este archivo Lo que tenés que saber antes de usarlo (es el contrato de Noti, no un detalle): - **Un 202 NO es «entregado».** `POST /send` contesta siempre `202 {"id"}`, le llegue el aviso a mil personas o a ninguna (nadie vinculado, silenció, cortó, te pasaste de un límite). Por eso `send()` devuelve un `Envio` cuyo `accepted` quiere decir «Noti lo aceptó», nada más. - **No hay estado «conectado».** `vincular_persona()` te da un pase de un solo uso; Noti no te dice si la persona lo abrió ni si después se dio de baja. Tu pantalla puede decir «pedí tu enlace», nunca «tus avisos están activados». - **`send()` nunca levanta**: avisar es accesorio, nunca tiene que romper tu operación (crear el pedido, cobrar, lo que sea). Un fallo es `accepted=False` y una línea de log. `vincular_persona()` sí levanta `NotiError`, porque ahí hay una persona esperando un link. - **Reintentos seguros.** Se reintenta sólo ante fallas de red y 5xx, con esperas cortas (0,4 s, 0,8 s). Noti descarta un segundo aviso a la misma persona dentro de los 10 segundos, así que un reintento tras un timeout no se duplica. Configuración por entorno (`Noti.from_env()`): NOTI_ENABLED=true apagado por defecto: sin esto no se manda nada NOTI_URL=https://noti.gualda.ai NOTI_TOKEN=ntf_... el token de la empresa (panel → Conectar) Para tests: `FakeNoti()` valida igual que Noti y guarda lo «enviado». """ from __future__ import annotations import json import logging import os import time import urllib.error import urllib.request from dataclasses import dataclass, field from typing import Any, Callable, Optional logger = logging.getLogger("noti") # Los topes de la API (docs/api-empresa.md). Se validan ANTES de la red: Noti # contesta 400, pero un aviso que no sale por un título largo es un aviso # perdido en silencio. TITLE_MAX = 100 BODY_MAX = 1000 CTA_LABEL_MAX = 40 CTA_URL_MAX = 2048 REF_MAX = 128 GROUP_MAX = 24 CTA_SCHEMES = ("http://", "https://", "mailto:", "tel:", "sms:", "whatsapp:") # transporte(método, url, headers, cuerpo, timeout) -> (status, cuerpo) Transport = Callable[[str, str, "dict[str, str]", Optional[bytes], float], "tuple[int, bytes]"] class NotiError(Exception): """Cualquier falla de Noti. El mensaje es seguro de mostrarle a una persona: nunca trae el token ni la URL.""" class NotiUnavailable(NotiError): """Noti está apagado, sin token o no contestó.""" @dataclass(frozen=True) class Pase: """El pase de un solo uso (vence a los 7 días). `url` es la página de Noti (botón «Abrir en Noti» en el teléfono, QR en la computadora); `qr_text` es lo que lleva el QR por dentro, por si querés dibujarlo vos.""" url: str qr_text: str expires_at: str @dataclass(frozen=True) class Envio: """El resultado de `send()`. `accepted` es «Noti lo aceptó», NO «le llegó». `id` identifica el aviso: sirve para `aviso(id)` (si se vio), nada más.""" accepted: bool id: Optional[str] = None skipped: bool = False # ya se había mandado con la misma `key` en este proceso def __bool__(self) -> bool: return self.accepted def validate( *, title: str, body: str, ref: Optional[str] = None, group: Optional[str] = None, emoji: Optional[str] = None, cta_label: Optional[str] = None, cta_url: Optional[str] = None, ) -> Optional[str]: """El motivo por el que Noti rechazaría el aviso, o `None` si está bien.""" if ref is not None and group is not None: return "usá uno solo de ref o group" if ref is not None and not 0 < len(ref) <= REF_MAX: return "ref vacío o de más de %d" % REF_MAX if group is not None and not 0 < len(group) <= GROUP_MAX: return "group vacío o de más de %d" % GROUP_MAX if not title or len(title) > TITLE_MAX: return "título vacío o de más de %d" % TITLE_MAX if not body or len(body) > BODY_MAX: return "texto vacío o de más de %d" % BODY_MAX if emoji is not None and not emoji: return "emoji vacío (poné exactamente uno o dejalo sin poner)" if (cta_label is None) != (cta_url is None): return "el botón lleva rótulo y link, los dos o ninguno" if cta_label is not None and cta_url is not None: if not 0 < len(cta_label) <= CTA_LABEL_MAX: return "rótulo del botón vacío o de más de %d" % CTA_LABEL_MAX if len(cta_url) > CTA_URL_MAX or not cta_url.lower().startswith(CTA_SCHEMES): return "link del botón inválido (http, https, mailto, tel, sms o whatsapp)" return None def _urllib_transport(method: str, url: str, headers: "dict[str, str]", body: Optional[bytes], timeout: float): request = urllib.request.Request(url, data=body, headers=headers, method=method) try: with urllib.request.urlopen(request, timeout=timeout) as response: return response.status, response.read() except urllib.error.HTTPError as error: # 4xx/5xx: es una respuesta, no una falla de red return error.code, error.read() class Noti: """El cliente real. Timeout corto: Noti nunca tiene que frenar tu operación.""" def __init__( self, *, base_url: str, token: str, timeout: float = 3.0, retries: int = 2, backoff: float = 0.4, transport: Optional[Transport] = None, sleep: Callable[[float], None] = time.sleep, ) -> None: self._base = base_url.rstrip("/") self._token = token self._timeout = timeout self._retries = retries self._backoff = backoff self._transport = transport or _urllib_transport self._sleep = sleep self._keys: "dict[str, float]" = {} @classmethod def from_env(cls, env: Optional["dict[str, str]"] = None, **kwargs: Any) -> "Noti | DisabledNoti": """`NOTI_ENABLED` + `NOTI_URL` + `NOTI_TOKEN`. Apagado o sin token no levanta: devuelve un `DisabledNoti` que no hace nada.""" env = os.environ if env is None else env if env.get("NOTI_ENABLED", "").strip().lower() not in ("1", "true", "yes", "si", "sí"): return DisabledNoti("Los avisos por Noti están apagados (NOTI_ENABLED).") token = env.get("NOTI_TOKEN", "").strip() if not token: return DisabledNoti("Falta NOTI_TOKEN.") return cls(base_url=env.get("NOTI_URL", "https://noti.gualda.ai"), token=token, **kwargs) # -- HTTP -------------------------------------------------------------- def _request(self, method: str, path: str, payload: Optional[dict] = None) -> "tuple[int, Any]": """Hace el pedido con reintentos. Devuelve `(status, json)`; levanta `NotiUnavailable` si no hubo manera de hablar con Noti.""" headers = {"Authorization": "Bearer " + self._token, "Accept": "application/json"} body = None if payload is not None: headers["Content-Type"] = "application/json" body = json.dumps(payload).encode("utf-8") ultimo = "no contestó" for intento in range(self._retries + 1): if intento: self._sleep(self._backoff * (2 ** (intento - 1))) try: status, raw = self._transport(method, self._base + path, headers, body, self._timeout) except Exception as error: # noqa: BLE001 — de un error de red sólo importa el TIPO (el texto puede traer la URL) ultimo = type(error).__name__ continue if status >= 500: ultimo = "HTTP %d" % status continue try: return status, (json.loads(raw) if raw else None) except ValueError: return status, None logger.warning("Noti no contestó (%s)", ultimo) raise NotiUnavailable("Noti no contestó. Probá de nuevo en un rato.") # -- avisos ------------------------------------------------------------ def send( self, *, title: str, body: str, ref: Optional[str] = None, group: Optional[str] = None, emoji: Optional[str] = None, cta_label: Optional[str] = None, cta_url: Optional[str] = None, key: Optional[str] = None, ) -> Envio: """Manda un aviso. Sin `ref` ni `group`, a todos los suscriptos; con `group` (el código de uno de tus QR), a los de ese QR; con `ref`, a UNA persona (la que vinculaste con ese `ref`). **No levanta nunca.** `accepted=True` es «Noti lo aceptó (202)», NO que llegó. `key` evita repetir el mismo aviso dentro de este proceso; la garantía real de «un evento = un aviso» es un UNIQUE en TU base.""" problema = validate( title=title, body=body, ref=ref, group=group, emoji=emoji, cta_label=cta_label, cta_url=cta_url ) if problema is not None: logger.warning("aviso a Noti descartado antes de enviar (%s)", problema) return Envio(False) if key is not None: if key in self._keys: return Envio(False, skipped=True) self._keys[key] = time.time() payload: "dict[str, Any]" = {"title": title, "body": body} for campo, valor in (("ref", ref), ("group", group), ("emoji", emoji)): if valor is not None: payload[campo] = valor if cta_label is not None and cta_url is not None: payload["cta"] = {"label": cta_label, "url": cta_url} try: status, data = self._request("POST", "/send", payload) except NotiError: if key is not None: self._keys.pop(key, None) return Envio(False) if status != 202: logger.warning("Noti contestó %d al aviso", status) if key is not None: self._keys.pop(key, None) return Envio(False) return Envio(True, id=(data or {}).get("id") if isinstance(data, dict) else None) def vincular_persona(self, ref: str, group: Optional[str] = None) -> Pase: """Pide el pase para que UNA persona vincule su Noti con tu empresa. Mostráselo (botón a `pase.url` o su QR). Sirve una vez y vence a los 7 días; si cambia de teléfono, pedí uno nuevo con el mismo `ref`. `ref` es TU id estable de la persona (id de usuario, cliente nº): mejor que un mail o un teléfono. Si lo sacás de la sesión y no de lo que manda el navegador, nadie pide el pase de otro. Levanta `NotiError`.""" if not 0 < len(ref) <= REF_MAX: raise NotiError("ref vacío o de más de %d caracteres" % REF_MAX) payload: "dict[str, Any]" = {"ref": ref} if group is not None: payload["group"] = group status, data = self._request("POST", "/company/connect-links", payload) if status == 401: raise NotiUnavailable("Noti rechazó el token: avisale a quien administra la integración.") if status == 404 and group is not None: raise NotiError("No existe un QR con el código %r en esta empresa." % group) if status != 201 or not isinstance(data, dict): raise NotiError("Noti no pudo armar el enlace. Probá de nuevo en un rato.") try: return Pase(url=str(data["url"]), qr_text=str(data["qr_text"]), expires_at=str(data["expires_at"])) except KeyError as error: raise NotiError("Noti contestó algo que no se entiende.") from error def aviso(self, id: str) -> Optional[dict]: """Qué se sabe de un aviso que mandaste: sólo señales POSITIVAS. Para un aviso a una persona (`ref`): `seen_on` (la fecha, sin hora) si lo abrió; si no, `null` — y `null` es «no sabemos»: no lo vio, lo silenció, cortó o no se vinculó, y Noti no distingue (a propósito). Para un aviso a muchos: `seen_count` (y `click_count`) sólo cuando son suficientes para no delatar a nadie; si no, `null`. Devuelve `None` si no existe (o si Noti no contestó).""" try: status, data = self._request("GET", "/company/avisos/" + id) except NotiError: return None return data if status == 200 and isinstance(data, dict) else None class DisabledNoti: """Apagado o sin token: no hace nada y no rompe nada. `vincular_persona` levanta con un motivo que una persona puede leer.""" def __init__(self, reason: str = "Los avisos por Noti están apagados.") -> None: self.reason = reason def send(self, **kwargs: Any) -> Envio: return Envio(False) def vincular_persona(self, ref: str, group: Optional[str] = None) -> Pase: raise NotiUnavailable(self.reason) def aviso(self, id: str) -> Optional[dict]: return None @dataclass class FakeNoti: """Para dev y tests. Valida igual que Noti y guarda lo «enviado»; `fail_with` simula un Noti caído (`send` da `accepted=False`, el pase levanta).""" sent: "list[dict[str, Any]]" = field(default_factory=list) links: "list[str]" = field(default_factory=list) fail_with: Optional[str] = None _keys: "set[str]" = field(default_factory=set) def send( self, *, title: str, body: str, ref: Optional[str] = None, group: Optional[str] = None, emoji: Optional[str] = None, cta_label: Optional[str] = None, cta_url: Optional[str] = None, key: Optional[str] = None, ) -> Envio: if self.fail_with is not None: return Envio(False) if validate( title=title, body=body, ref=ref, group=group, emoji=emoji, cta_label=cta_label, cta_url=cta_url ): return Envio(False) if key is not None: if key in self._keys: return Envio(False, skipped=True) self._keys.add(key) self.sent.append( {"title": title, "body": body, "ref": ref, "group": group, "emoji": emoji, "cta_label": cta_label, "cta_url": cta_url} ) return Envio(True, id="fake-%d" % len(self.sent)) def vincular_persona(self, ref: str, group: Optional[str] = None) -> Pase: if self.fail_with is not None: raise NotiUnavailable(self.fail_with) if not 0 < len(ref) <= REF_MAX: raise NotiError("ref vacío o de más de %d caracteres" % REF_MAX) self.links.append(ref) token = "fake%d" % len(self.links) return Pase(url="https://noti.test/c/" + token, qr_text="noti://c/" + token, expires_at="2099-01-01T00:00:00Z") def aviso(self, id: str) -> Optional[dict]: return None