import logging import httpx from app.config import get_settings from app.models import Event, Problem logger = logging.getLogger(__name__) SEVERITY_ORDER = {"info": 10, "warning": 20, "high": 30, "critical": 40} def _severity_value(value: str) -> int: return SEVERITY_ORDER.get(value, 0) def _is_enabled() -> bool: settings = get_settings() return bool( settings.telegram_enabled and settings.telegram_bot_token.strip() and settings.telegram_chat_id.strip() ) def _send_text(message: str) -> None: settings = get_settings() if not _is_enabled(): return url = f"https://api.telegram.org/bot{settings.telegram_bot_token}/sendMessage" payload = {"chat_id": settings.telegram_chat_id, "text": message, "disable_web_page_preview": True} try: with httpx.Client(timeout=8.0) as client: response = client.post(url, json=payload) response.raise_for_status() except Exception: logger.exception("telegram send failed") def notify_event(event: Event) -> None: settings = get_settings() min_level = _severity_value(settings.telegram_min_severity) if _severity_value(event.severity) < min_level: return host = event.host.hostname if event.host else "unknown" message = ( f"🚨 SAC событие\n" f"Host: {host}\n" f"Severity: {event.severity}\n" f"Type: {event.type}\n" f"Title: {event.title}\n" f"Summary: {event.summary}" ) _send_text(message) def notify_problem(problem: Problem, event: Event | None = None) -> None: host = problem.host.hostname if problem.host else "unknown" related = "" if event is not None: related = f"\nEvent: {event.type} ({event.severity})" message = ( f"🔥 SAC Problem opened\n" f"Host: {host}\n" f"Severity: {problem.severity}\n" f"Rule: {problem.rule_id or '-'}\n" f"Title: {problem.title}\n" f"Summary: {problem.summary}{related}" ) _send_text(message)