楼湘缘
293 lines
9.4 KiB
Python
293 lines
9.4 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
网络监控核心模块
|
|
使用OOP设计实现的网络监控器
|
|
"""
|
|
|
|
import time
|
|
import platform
|
|
from datetime import datetime
|
|
from typing import Optional, List
|
|
from dataclasses import dataclass
|
|
|
|
from config_manager import ConfigManager, MonitorConfig
|
|
from ping_executor import PingExecutorFactory, PingResult
|
|
from shutdown_handler import ShutdownHandlerFactory
|
|
|
|
|
|
@dataclass
|
|
class MonitorState:
|
|
"""监控状态数据类"""
|
|
is_running: bool = True
|
|
consecutive_failures: int = 0
|
|
failure_start_time: Optional[float] = None
|
|
ping_count: int = 0
|
|
last_status: Optional[bool] = None
|
|
session_start_time: float = 0.0
|
|
|
|
def reset(self) -> None:
|
|
"""重置状态"""
|
|
self.is_running = True
|
|
self.consecutive_failures = 0
|
|
self.failure_start_time = None
|
|
self.ping_count = 0
|
|
self.last_status = None
|
|
self.session_start_time = time.time()
|
|
|
|
def get_session_duration(self) -> float:
|
|
"""获取会话持续时间"""
|
|
return time.time() - self.session_start_time
|
|
|
|
|
|
class NetworkMonitor:
|
|
"""网络监控器主类（观察者模式的核心）"""
|
|
|
|
def __init__(self, config: Optional[MonitorConfig] = None):
|
|
"""
|
|
初始化网络监控器
|
|
|
|
Args:
|
|
config: 监控配置，如果为None则使用默认配置
|
|
"""
|
|
# 配置管理
|
|
self.config_manager = ConfigManager()
|
|
self.config: MonitorConfig = config or self.config_manager.load_from_args()
|
|
|
|
# 状态管理
|
|
self.state = MonitorState()
|
|
self.state.session_start_time = time.time()
|
|
|
|
# 依赖注入 - Ping执行器
|
|
self.ping_executor = PingExecutorFactory.create()
|
|
|
|
# 依赖注入 - 关机处理器（延迟初始化）
|
|
self._shutdown_handler: Optional['ShutdownHandler'] = None
|
|
|
|
# 观察者列表
|
|
self._observers: List['MonitorObserver'] = []
|
|
|
|
# 初始化日志
|
|
self._print_startup_info()
|
|
|
|
@property
|
|
def shutdown_handler(self) -> 'shutdown_handler.ShutdownHandler':
|
|
"""延迟初始化关机处理器"""
|
|
if self._shutdown_handler is None:
|
|
self._shutdown_handler = ShutdownHandlerFactory.create(
|
|
delay=self.config.shutdown_delay,
|
|
message=self.config.shutdown_message
|
|
)
|
|
return self._shutdown_handler
|
|
|
|
def add_observer(self, observer: 'MonitorObserver') -> None:
|
|
"""
|
|
添加观察者
|
|
|
|
Args:
|
|
observer: 监控观察者
|
|
"""
|
|
self._observers.append(observer)
|
|
|
|
def remove_observer(self, observer: 'MonitorObserver') -> None:
|
|
"""
|
|
移除观察者
|
|
|
|
Args:
|
|
observer: 监控观察者
|
|
"""
|
|
if observer in self._observers:
|
|
self._observers.remove(observer)
|
|
|
|
def _notify_ping(self, result: PingResult) -> None:
|
|
"""通知观察者ping事件"""
|
|
for observer in self._observers:
|
|
observer.on_ping(result)
|
|
|
|
def _notify_status_change(self, is_connected: bool, duration: Optional[float] = None) -> None:
|
|
"""通知观察者状态变化事件"""
|
|
for observer in self._observers:
|
|
observer.on_status_change(is_connected, duration)
|
|
|
|
def _notify_shutdown(self, reason: str) -> None:
|
|
"""通知观察者关机事件"""
|
|
for observer in self._observers:
|
|
observer.on_shutdown(reason)
|
|
|
|
def _get_timestamp(self) -> str:
|
|
"""获取当前时间戳"""
|
|
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
def _print_startup_info(self) -> None:
|
|
"""打印启动信息"""
|
|
print("=" * 60)
|
|
print("网络监控器")
|
|
print("=" * 60)
|
|
print(f"[{self._get_timestamp()}] 系统信息:")
|
|
print(f" - 操作系统: {platform.system()}")
|
|
print(f" - 目标IP: {self.config.target_ip}")
|
|
print(f" - 延迟阈值: {self.config.delay_threshold}秒")
|
|
print(f" - Ping间隔: {self.config.ping_interval}秒")
|
|
print(f" - Ping超时: {self.config.ping_timeout}秒")
|
|
print("=" * 60)
|
|
print()
|
|
|
|
def ping(self) -> PingResult:
|
|
"""
|
|
执行ping操作
|
|
|
|
Returns:
|
|
Ping结果
|
|
"""
|
|
result = self.ping_executor.execute(
|
|
self.config.target_ip,
|
|
self.config.ping_timeout
|
|
)
|
|
self.state.ping_count += 1
|
|
self._notify_ping(result)
|
|
return result
|
|
|
|
def _handle_success(self) -> None:
|
|
"""处理ping成功"""
|
|
if self.state.last_status is False:
|
|
self._notify_status_change(True, None)
|
|
|
|
self.state.consecutive_failures = 0
|
|
self.state.failure_start_time = None
|
|
self.state.last_status = True
|
|
|
|
def _handle_failure(self) -> None:
|
|
"""处理ping失败"""
|
|
if self.state.last_status is True or self.state.last_status is None:
|
|
self._notify_status_change(False, None)
|
|
|
|
self.state.consecutive_failures += 1
|
|
self.state.last_status = False
|
|
|
|
if self.state.failure_start_time is None:
|
|
self.state.failure_start_time = time.time()
|
|
self._print_message("开始记录连接失败时间")
|
|
else:
|
|
failure_duration = time.time() - self.state.failure_start_time
|
|
remaining_time = self.config.delay_threshold - failure_duration
|
|
|
|
if remaining_time > 0:
|
|
self._print_message(
|
|
f"连续失败: {self.state.consecutive_failures}次, "
|
|
f"已持续: {failure_duration:.1f}秒, "
|
|
f"剩余: {remaining_time:.1f}秒"
|
|
)
|
|
self._notify_status_change(False, failure_duration)
|
|
|
|
if failure_duration >= self.config.delay_threshold:
|
|
self._trigger_shutdown()
|
|
|
|
def _trigger_shutdown(self) -> None:
|
|
"""触发关机"""
|
|
reason = f"连续{self.config.delay_threshold}秒无法连接到 {self.config.target_ip}"
|
|
self.shutdown_handler.log_shutdown(reason)
|
|
self._notify_shutdown(reason)
|
|
self.shutdown_handler.execute()
|
|
self.state.is_running = False
|
|
|
|
def _print_message(self, message: str) -> None:
|
|
"""打印消息"""
|
|
print(f"[{self._get_timestamp()}] {message}")
|
|
|
|
def monitor(self) -> None:
|
|
"""启动监控循环"""
|
|
self._print_message("开始监控网络连接...")
|
|
self._print_message(f"每 {self.config.ping_interval} 秒执行一次ping操作")
|
|
print("-" * 60)
|
|
|
|
try:
|
|
while self.state.is_running:
|
|
result = self.ping()
|
|
|
|
# 显示ping结果
|
|
status_str = f"Ping #{self.state.ping_count} {result}"
|
|
self._print_message(status_str)
|
|
|
|
# 处理结果
|
|
if result.success:
|
|
self._handle_success()
|
|
else:
|
|
self._handle_failure()
|
|
|
|
if not self.state.is_running:
|
|
break
|
|
|
|
# 等待下一次ping
|
|
print(f"[{self._get_timestamp()}] 等待 {self.config.ping_interval} 秒...")
|
|
time.sleep(self.config.ping_interval)
|
|
|
|
except KeyboardInterrupt:
|
|
self._stop_by_user()
|
|
except Exception as e:
|
|
self._stop_by_error(e)
|
|
|
|
def _stop_by_user(self) -> None:
|
|
"""用户中断停止"""
|
|
print(f"\n[{self._get_timestamp()}] 用户中断，监控已停止")
|
|
self._print_statistics()
|
|
|
|
def _stop_by_error(self, error: Exception) -> None:
|
|
"""错误停止"""
|
|
print(f"\n[{self._get_timestamp()}] 监控过程中发生错误: {error}")
|
|
self._print_statistics()
|
|
|
|
def _print_statistics(self) -> None:
|
|
"""打印统计信息"""
|
|
duration = self.state.get_session_duration()
|
|
print(f"[{self._get_timestamp()}] 统计信息:")
|
|
print(f" - 总ping次数: {self.state.ping_count}")
|
|
print(f" - 连续失败次数: {self.state.consecutive_failures}")
|
|
print(f" - 运行时长: {duration:.1f}秒")
|
|
|
|
def get_statistics(self) -> dict:
|
|
"""
|
|
获取统计信息
|
|
|
|
Returns:
|
|
统计信息字典
|
|
"""
|
|
return {
|
|
'total_pings': self.state.ping_count,
|
|
'consecutive_failures': self.state.consecutive_failures,
|
|
'session_duration': self.state.get_session_duration(),
|
|
'target_ip': self.config.target_ip,
|
|
'is_running': self.state.is_running
|
|
}
|
|
|
|
|
|
class MonitorObserver:
|
|
"""监控观察者抽象基类"""
|
|
|
|
def on_ping(self, result: PingResult) -> None:
|
|
"""
|
|
Ping事件回调
|
|
|
|
Args:
|
|
result: Ping结果
|
|
"""
|
|
pass
|
|
|
|
def on_status_change(self, is_connected: bool, duration: Optional[float] = None) -> None:
|
|
"""
|
|
状态变化事件回调
|
|
|
|
Args:
|
|
is_connected: 是否连接
|
|
duration: 持续时间
|
|
"""
|
|
pass
|
|
|
|
def on_shutdown(self, reason: str) -> None:
|
|
"""
|
|
关机事件回调
|
|
|
|
Args:
|
|
reason: 关机原因
|
|
"""
|
|
pass
|