"""系统通知适配器 — 三平台原生通知中心。 职责: 把后端产生的告警事件推送到操作系统通知中心。 窗口最小化 / 被遮挡 / 后台运行时都能弹通知 (不依赖前端 WebView)。 平台实现: - Windows: winotify (进现代操作中心, 支持图标) - macOS: osascript (系统已内置, 无需额外依赖) - Linux: notify-send (系统已内置) / plyer 兜底 设计: 失败静默降级, 绝不因通知失败阻断告警主流程 (落盘 / SSE 推送)。 通知去重不在本层做, 复用 MonitorRuleEngine 的 cooldown 逻辑。 """ from __future__ import annotations import logging import subprocess import sys from pathlib import Path logger = logging.getLogger(__name__) # 单次通知最长字符 (避免某些平台截断报错) _MAX_LEN = 200 # 避免重复探测平台能力, 缓存一次 _backend_cache: str | None = None def _detect_backend() -> str | None: """探测当前平台可用的通知后端。""" global _backend_cache if _backend_cache is not None: return _backend_cache if _backend_cache != "none" else None backend = None if sys.platform == "win32": try: import winotify # type: ignore[import-not-found] # noqa: F401 backend = "winotify" except ImportError: logger.debug("winotify 不可用, Windows 通知降级") backend = None elif sys.platform == "darwin": backend = "osascript" elif sys.platform.startswith("linux"): backend = "notify-send" _backend_cache = backend or "none" return backend def _truncate(text: str) -> str: """截断超长文本, 避免平台通知上限报错。""" text = (text or "").strip() return text[:_MAX_LEN] + ("…" if len(text) > _MAX_LEN else "") def notify(title: str, message: str, icon: Path | None = None) -> bool: """推送一条系统通知。 Args: title: 通知标题 message: 通知正文 icon: 可选图标路径 (部分平台支持) Returns: True=成功送达, False=失败或无可用后端。 失败静默, 不抛异常 (通知是辅助通道, 不能阻断告警主流程)。 """ title = _truncate(title) message = _truncate(message) if not title: return False backend = _detect_backend() if backend is None: return False try: if backend == "winotify": return _notify_winotify(title, message) if backend == "osascript": return _notify_osascript(title, message) if backend == "notify-send": return _notify_notify_send(title, message, icon) except Exception as e: # noqa: BLE001 logger.debug("系统通知失败 (%s): %s", backend, e) return False return False def _notify_winotify(title: str, message: str) -> bool: """Windows 通知 (winotify) — 进现代操作中心。""" from winotify import Notifier # type: ignore[import-not-found] Notifier().create_notification( title=title, msg=message, # winotify 要求 duration 为 "short" 或 "long" duration="short", # 无可点击动作 (桌面版不实现"点击回到窗口"的复杂交互) ).show() return True def _notify_osascript(title: str, message: str) -> bool: """macOS 通知 (osascript) — 调用系统 AppleScript。""" # 转义双引号, 避免 AppleScript 注入 safe_title = title.replace('"', '\\"') safe_msg = message.replace('"', '\\"') script = ( f'display notification "{safe_msg}" with title "{safe_title}"' ) result = subprocess.run( # noqa: S603, S607 ["osascript", "-e", script], capture_output=True, timeout=5, ) return result.returncode == 0 def _notify_notify_send(title: str, message: str, icon: Path | None) -> bool: """Linux 通知 (notify-send) — freedesktop.org 标准。""" args = ["notify-send", title] if icon and Path(icon).exists(): args.extend(["--icon", str(icon)]) args.append(message) result = subprocess.run( # noqa: S603 args, capture_output=True, timeout=5, ) return result.returncode == 0