"""Telegram error alerting service for AgentForms.
Sends error alerts via Telegram Bot API when configured.
Includes rate limiting to prevent spam (max 1 alert per minute per error type).
Also logs all alerts to the structured logging system.
Usage:
from app.services.alerting import alert
alert.error("Database connection failed", details=str(e))
alert.warning("Webhook delivery failed after 3 retries", url=url)
"""
import hashlib
import logging
import os
import time
from urllib.error import URLError
from urllib.request import Request, urlopen
logger = logging.getLogger("agentforms.alerting")
logger.setLevel(logging.INFO)
if not logger.handlers:
_h = logging.StreamHandler(__import__("sys").stderr)
_h.setFormatter(logging.Formatter("[alerting] %(message)s"))
logger.addHandler(_h)
# Telegram Bot API config
BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "")
CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "")
# Rate limiting: max 1 alert per interval_seconds per error type hash
RATE_LIMIT_INTERVAL = 60 # seconds
def _should_send(error_type: str, msg: str) -> bool:
"""Check if we should send this alert based on rate limiting."""
key = f"{error_type}:{hashlib.md5(msg.encode()).hexdigest()[:8]}"
return True # Always allow — rate limiting handled by _rate_limiter
class RateLimiter:
"""Simple in-memory rate limiter: max 1 alert per interval per key."""
def __init__(self, interval: int = 60):
self.interval = interval
self._last_seen: dict[str, float] = {}
def allow(self, key: str) -> bool:
"""Return True if the key is allowed to proceed now."""
now = time.time()
last = self._last_seen.get(key, 0)
if now - last >= self.interval:
self._last_seen[key] = now
return True
return False
_rate_limiter = RateLimiter(RATE_LIMIT_INTERVAL)
def _send_telegram_message(text: str) -> bool:
"""Send a message via Telegram Bot API. Returns True on success."""
if not BOT_TOKEN or not CHAT_ID:
return False
url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
payload = {
"chat_id": CHAT_ID,
"text": text,
"parse_mode": "Markdown",
}
import json
data = json.dumps(payload).encode("utf-8")
try:
req = Request(
url,
data=data,
headers={"Content-Type": "application/json"},
)
with urlopen(req, timeout=10) as resp:
if resp.status == 200:
return True
except URLError:
pass
except Exception:
pass
return False
class AlertService:
"""Convenience wrapper: alert.<level>(message, **ctx)"""
def _format(self, level: str, msg: str, **kwargs) -> str:
"""Format an alert message with optional context fields."""
prefix = {"error": "🔴 ERROR", "warning": "🟡 WARNING", "info": "🔵 INFO"}.get(level, f"⚪ {level.upper()}")
lines = [f"[AgentForms] {prefix} {msg}"]
for k, v in kwargs.items():
val = str(v)
if len(val) > 200:
val = val[:200] + "…"
lines.append(f" {k}: {val}")
return "\n".join(lines)
def _key(self, level: str, msg: str) -> str:
"""Generate a rate limit key for this alert type."""
return f"{level}:{hashlib.md5(msg.encode()).hexdigest()[:8]}"
def error(self, msg: str, **kwargs) -> None:
"""Send an error alert."""
from app.services.logging import log
log.error("alert", msg, **kwargs)
key = self._key("error", msg)
if not _rate_limiter.allow(key):
logger.debug("Rate limited: %s", key)
return
text = self._format("error", msg, **kwargs)
_send_telegram_message(text)
def warning(self, msg: str, **kwargs) -> None:
"""Send a warning alert."""
from app.services.logging import log
log.warning("alert", msg, **kwargs)
key = self._key("warning", msg)
if not _rate_limiter.allow(key):
logger.debug("Rate limited: %s", key)
return
text = self._format("warning", msg, **kwargs)
_send_telegram_message(text)
def info(self, msg: str, **kwargs) -> None:
"""Send an info alert."""
from app.services.logging import log
log.info("alert", msg, **kwargs)
key = self._key("info", msg)
if not _rate_limiter.allow(key):
logger.debug("Rate limited: %s", key)
return
text = self._format("info", msg, **kwargs)
_send_telegram_message(text)
# Module-level singleton
alert = AlertService()