"""系统通知适配器 — 调用操作系统原生通知命令。 职责: 把后端产生的告警事件推送到操作系统通知中心。 窗口最小化 / 被遮挡 / 后台运行时都能弹通知。 平台实现: - macOS: osascript (系统已内置) - Linux: notify-send (系统已内置) - Windows: 暂不支持原生通知中心 (无额外依赖实现) 设计: 失败静默降级, 绝不因通知失败阻断告警主流程 (落盘 / 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 == "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 == "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_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