#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 通知模块 OOP设计的通知系统 """ import platform from abc import ABC, abstractmethod from typing import List, Optional class Notifier(ABC): """通知器抽象基类(策略模式)""" @abstractmethod def send(self, title: str, message: str) -> bool: """ 发送通知 Args: title: 通知标题 message: 通知内容 Returns: 是否发送成功 """ pass def is_enabled(self) -> bool: """ 检查通知器是否启用 Returns: 是否启用 """ return True class ConsoleNotifier(Notifier): """控制台通知器""" def send(self, title: str, message: str) -> bool: """在控制台显示通知""" separator = "=" * 60 print(f"\n{separator}") print(f"【{title}】") print(f"{message}") print(f"{separator}\n") return True class DesktopNotifier(Notifier): """桌面通知器""" def __init__(self): """初始化桌面通知器""" self.os_type = platform.system().lower() self.enabled = self._initialize() def _initialize(self) -> bool: """初始化通知库""" try: if self.os_type == "windows": import win10toast self.toaster = win10toast.ToastNotifier() elif self.os_type == "linux": import notify2 notify2.init("网络监控器") elif self.os_type == "darwin": import pync return True except ImportError: return False def is_enabled(self) -> bool: """检查是否启用""" return self.enabled def send(self, title: str, message: str) -> bool: """发送桌面通知""" if not self.is_enabled(): return False try: if self.os_type == "windows": self.toaster.show_notification(title, message, duration=10) elif self.os_type == "linux": import notify2 notification = notify2.Notification(title, message) notification.show() elif self.os_type == "darwin": import pync pync.notify(message, title=title) return True except Exception: return False class EmailNotifier(Notifier): """邮件通知器""" def __init__( self, smtp_server: Optional[str] = None, smtp_port: Optional[int] = None, username: Optional[str] = None, password: Optional[str] = None, from_email: Optional[str] = None, to_email: Optional[str] = None ): """ 初始化邮件通知器 Args: smtp_server: SMTP服务器 smtp_port: SMTP端口 username: 用户名 password: 密码 from_email: 发件人邮箱 to_email: 收件人邮箱 """ self.smtp_server = smtp_server self.smtp_port = smtp_port self.username = username self.password = password self.from_email = from_email self.to_email = to_email self.enabled = self._check_enabled() def _check_enabled(self) -> bool: """检查是否配置完整""" return all([ self.smtp_server, self.smtp_port, self.username, self.password, self.from_email, self.to_email ]) def is_enabled(self) -> bool: """检查是否启用""" return self.enabled def send(self, title: str, message: str) -> bool: """发送邮件通知""" if not self.is_enabled(): return False try: import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart msg = MIMEMultipart() msg['From'] = self.from_email msg['To'] = self.to_email msg['Subject'] = title msg.attach(MIMEText(message, 'plain', 'utf-8')) with smtplib.SMTP(self.smtp_server, self.smtp_port) as server: server.starttls() server.login(self.username, self.password) server.send_message(msg) return True except Exception: return False class NotificationManager: """通知管理器(组合模式)""" def __init__(self, notifiers: Optional[List[Notifier]] = None): """ 初始化通知管理器 Args: notifiers: 通知器列表 """ self.notifiers: List[Notifier] = notifiers or [] def add_notifier(self, notifier: Notifier) -> None: """ 添加通知器 Args: notifier: 通知器实例 """ self.notifiers.append(notifier) def remove_notifier(self, notifier: Notifier) -> None: """ 移除通知器 Args: notifier: 通知器实例 """ if notifier in self.notifiers: self.notifiers.remove(notifier) def notify(self, title: str, message: str) -> int: """ 发送通知到所有通知器 Args: title: 通知标题 message: 通知内容 Returns: 成功发送的通知器数量 """ success_count = 0 for notifier in self.notifiers: if notifier.send(title, message): success_count += 1 return success_count def notify_connection_lost(self, ip: str, duration: float) -> None: """ 通知连接丢失 Args: ip: IP地址 duration: 持续时间(秒) """ self.notify( "网络连接丢失", f"无法连接到 {ip}\n已断开 {duration:.1f} 秒" ) def notify_shutdown(self, reason: str) -> None: """ 通知系统即将关机 Args: reason: 关机原因 """ self.notify( "系统即将关机", f"原因: {reason}\n请保存所有工作!" ) class NotificationBuilder: """通知构建器(建造者模式)""" def __init__(self): """初始化构建器""" self._enable_console = True self._enable_desktop = False self._enable_email = False self._email_config: dict = {} def enable_console(self, enable: bool = True) -> 'NotificationBuilder': """ 启用控制台通知 Args: enable: 是否启用 Returns: 构建器实例 """ self._enable_console = enable return self def enable_desktop(self, enable: bool = True) -> 'NotificationBuilder': """ 启用桌面通知 Args: enable: 是否启用 Returns: 构建器实例 """ self._enable_desktop = enable return self def enable_email(self, **config) -> 'NotificationBuilder': """ 启用邮件通知 Args: **config: 邮件配置 Returns: 构建器实例 """ self._enable_email = True self._email_config = config return self def build(self) -> NotificationManager: """ 构建通知管理器 Returns: 通知管理器实例 """ notifiers = [] if self._enable_console: notifiers.append(ConsoleNotifier()) if self._enable_desktop: notifiers.append(DesktopNotifier()) if self._enable_email: notifiers.append(EmailNotifier(**self._email_config)) return NotificationManager(notifiers)