@@ -0,0 +1,136 @@
|
||||
"""系统通知适配器 — 三平台原生通知中心。
|
||||
|
||||
职责: 把后端产生的告警事件推送到操作系统通知中心。
|
||||
窗口最小化 / 被遮挡 / 后台运行时都能弹通知 (不依赖前端 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
|
||||
Reference in New Issue
Block a user