from datetime import datetime from typing import List from core.logger import get_logger logger = get_logger("core.notifications") class Notifier: def __init__(self, app: "OverUB") -> None: self.app = app cfg = app.config.get().get("notifications", {}) self.channels: List[str] = cfg.get("channels", ["telegram"]) self.enabled = bool(cfg.get("enabled", False)) self.quiet_hours = cfg.get("quiet_hours", {}) self.events = cfg.get("events", []) async def notify(self, text: str, event: str = "") -> None: if not self.enabled: return if event and self.events and event not in self.events: return if self._in_quiet_hours(): return if "telegram" in self.channels and self.app.client.client: try: await self.app.client.client.send_message("me", text) except Exception: logger.exception("Telegram notification failed") if "desktop" in self.channels: logger.info("Desktop notification: %s", text) if "email" in self.channels: logger.info("Email notification: %s", text) if "webhook" in self.channels: logger.info("Webhook notification: %s", text) def _in_quiet_hours(self) -> bool: start = self.quiet_hours.get("start") end = self.quiet_hours.get("end") if not start or not end: return False now = datetime.now().strftime("%H:%M") if start <= end: return start <= now <= end return now >= start or now <= end