47 lines
1.6 KiB
Python
47 lines
1.6 KiB
Python
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
|