"""Notification toggle helpers — shared by Telegram and alert dispatch.""" from server.config import settings _DISABLED = frozenset({"false", "0", "no", "off"}) # alert_type (cpu/mem/disk) → Settings attribute _RESOURCE_ALERT_ATTR: dict[str, str] = { "cpu": "NOTIFY_ALERT_CPU", "mem": "NOTIFY_ALERT_MEM", "disk": "NOTIFY_ALERT_DISK", } def notify_enabled(setting_value: str | bool | None) -> bool: """Return True when a notify_* setting is enabled.""" if setting_value is None: return True if isinstance(setting_value, bool): return setting_value return setting_value.strip().lower() not in _DISABLED def resource_alert_notify_enabled(alert_type: str) -> bool: """Check whether Telegram should fire for a resource alert type.""" attr = _RESOURCE_ALERT_ATTR.get(alert_type.lower(), "NOTIFY_ALERT_CPU") return notify_enabled(getattr(settings, attr, "true")) def recovery_notify_enabled() -> bool: """Global toggle for resource metric recovery Telegram.""" return notify_enabled(settings.NOTIFY_RECOVERY) def resource_recovery_notify_enabled(metric: str) -> bool: """Telegram recovery requires global NOTIFY_RECOVERY and per-metric alert toggle.""" if not recovery_notify_enabled(): return False return resource_alert_notify_enabled(metric) def notify_attr_enabled(attr_name: str, default: str = "true") -> bool: """Check a named NOTIFY_* settings attribute.""" return notify_enabled(getattr(settings, attr_name, default)) def offline_alert_notify_enabled() -> bool: """Telegram toggle for server offline (Agent heartbeat lost) alerts.""" return notify_attr_enabled("NOTIFY_ALERT_OFFLINE")