feat: 添加网络监控和自动关机脚本
This commit is contained in:
楼湘缘
@@ -0,0 +1,189 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
网络监控脚本
|
||||
持续监控指定IP地址的连接状态，连续30秒无法连接时自动关机
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import subprocess
|
||||
import platform
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class NetworkMonitor:
|
||||
"""网络监控器类"""
|
||||
|
||||
def __init__(self, target_ip: str, delay_threshold: float = 30.0, ping_interval: float = 5.0):
|
||||
"""
|
||||
初始化网络监控器
|
||||
|
||||
Args:
|
||||
target_ip: 要监控的IP地址
|
||||
delay_threshold: 延迟阈值（秒），默认30秒
|
||||
ping_interval: ping检测间隔（秒），默认5秒
|
||||
"""
|
||||
self.target_ip = target_ip
|
||||
self.delay_threshold = delay_threshold
|
||||
self.ping_interval = ping_interval
|
||||
self.failure_start_time: Optional[float] = None
|
||||
self.is_running = True
|
||||
self.ping_count = 0
|
||||
|
||||
# 根据操作系统设置ping命令
|
||||
self.os_type = platform.system().lower()
|
||||
if self.os_type == "windows":
|
||||
self.ping_command = ["ping", "-n", "1", "-w", "2000"]
|
||||
else:
|
||||
self.ping_command = ["ping", "-c", "1", "-W", "2"]
|
||||
|
||||
print(f"[{self._get_timestamp()}] 网络监控器已启动")
|
||||
print(f"[{self._get_timestamp()}] 目标IP: {self.target_ip}")
|
||||
print(f"[{self._get_timestamp()}] 延迟阈值: {self.delay_threshold}秒")
|
||||
print(f"[{self._get_timestamp()}] Ping间隔: {self.ping_interval}秒")
|
||||
print(f"[{self._get_timestamp()}] 操作系统: {platform.system()}")
|
||||
print("-" * 60)
|
||||
|
||||
def _get_timestamp(self) -> str:
|
||||
"""获取当前时间戳"""
|
||||
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
|
||||
def ping_host(self) -> bool:
|
||||
"""
|
||||
执行ping操作检查主机连接状态
|
||||
|
||||
Returns:
|
||||
bool: 如果ping成功返回True，否则返回False
|
||||
"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
self.ping_command + [self.target_ip],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
timeout=5
|
||||
)
|
||||
self.ping_count += 1
|
||||
print(f"[{self._get_timestamp()}] Ping #{self.ping_count} 执行完成", end="")
|
||||
return result.returncode == 0
|
||||
except subprocess.TimeoutExpired:
|
||||
self.ping_count += 1
|
||||
print(f"[{self._get_timestamp()}] Ping #{self.ping_count} 超时", end="")
|
||||
return False
|
||||
except Exception as e:
|
||||
self.ping_count += 1
|
||||
print(f"[{self._get_timestamp()}] Ping #{self.ping_count} 错误: {e}", end="")
|
||||
return False
|
||||
|
||||
def shutdown_system(self):
|
||||
"""执行系统关机操作"""
|
||||
print(f"\n[{self._get_timestamp()}] !!! 连续{self.delay_threshold}秒无法连接，准备关机 !!!")
|
||||
|
||||
if self.os_type == "windows":
|
||||
print(f"[{self._get_timestamp()}] 正在执行Windows关机...")
|
||||
os.system("shutdown /s /t 5 /c \"网络连接丢失，系统将在5秒后关机\"")
|
||||
elif self.os_type == "linux":
|
||||
print(f"[{self._get_timestamp()}] 正在执行Linux关机...")
|
||||
os.system("shutdown -h +1 \"网络连接丢失，系统将在1分钟后关机\"")
|
||||
elif self.os_type == "darwin":
|
||||
print(f"[{self._get_timestamp()}] 正在执行macOS关机...")
|
||||
os.system("shutdown -h +1 \"网络连接丢失，系统将在1分钟后关机\"")
|
||||
else:
|
||||
print(f"[{self._get_timestamp()}] 未知操作系统，无法自动关机")
|
||||
sys.exit(1)
|
||||
|
||||
def monitor(self):
|
||||
"""启动网络监控"""
|
||||
print(f"[{self._get_timestamp()}] 开始监控网络连接...")
|
||||
print(f"[{self._get_timestamp()}] 每 {self.ping_interval} 秒执行一次ping操作")
|
||||
print("-" * 60)
|
||||
|
||||
consecutive_failures = 0
|
||||
last_status = None
|
||||
|
||||
try:
|
||||
while self.is_running:
|
||||
# 执行ping操作
|
||||
is_connected = self.ping_host()
|
||||
|
||||
# 显示ping结果
|
||||
status_str = " → ✓ 连接正常" if is_connected else " → ✗ 连接失败"
|
||||
print(status_str)
|
||||
|
||||
# 状态变化时额外输出日志
|
||||
if last_status != is_connected:
|
||||
if not is_connected:
|
||||
print(f"[{self._get_timestamp()}] 状态变化: 检测到连接失败")
|
||||
last_status = is_connected
|
||||
|
||||
if is_connected:
|
||||
# 连接成功，重置失败计数
|
||||
consecutive_failures = 0
|
||||
self.failure_start_time = None
|
||||
else:
|
||||
# 连接失败
|
||||
consecutive_failures += 1
|
||||
|
||||
if self.failure_start_time is None:
|
||||
# 第一次失败，记录开始时间
|
||||
self.failure_start_time = time.time()
|
||||
print(f"[{self._get_timestamp()}] 开始记录连接失败时间")
|
||||
else:
|
||||
# 计算失败持续时间
|
||||
failure_duration = time.time() - self.failure_start_time
|
||||
remaining_time = self.delay_threshold - failure_duration
|
||||
|
||||
if remaining_time > 0:
|
||||
print(f"[{self._get_timestamp()}] 连续失败: {consecutive_failures}次, "
|
||||
f"已持续: {failure_duration:.1f}秒, "
|
||||
f"剩余: {remaining_time:.1f}秒")
|
||||
|
||||
if failure_duration >= self.delay_threshold:
|
||||
# 超过阈值，执行关机
|
||||
self.shutdown_system()
|
||||
self.is_running = False
|
||||
break
|
||||
|
||||
# 按照设定的间隔等待下一次ping操作
|
||||
print(f"[{self._get_timestamp()}] 等待 {self.ping_interval} 秒后执行下一次ping...")
|
||||
time.sleep(self.ping_interval)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print(f"\n[{self._get_timestamp()}] 用户中断，监控已停止")
|
||||
print(f"[{self._get_timestamp()}] 总共执行了 {self.ping_count} 次ping操作")
|
||||
except Exception as e:
|
||||
print(f"\n[{self._get_timestamp()}] 监控过程中发生错误: {e}")
|
||||
print(f"[{self._get_timestamp()}] 总共执行了 {self.ping_count} 次ping操作")
|
||||
|
||||
|
||||
def main():
|
||||
"""主函数"""
|
||||
# 可配置参数
|
||||
CONFIG = {
|
||||
"target_ip": "192.168.2.6", # Google DNS服务器
|
||||
"delay_threshold": 30.0, # 延迟阈值（秒）
|
||||
"ping_interval": 5.0 # Ping检测间隔（秒）
|
||||
}
|
||||
|
||||
# 可以通过命令行参数覆盖配置
|
||||
if len(sys.argv) > 1:
|
||||
CONFIG["target_ip"] = sys.argv[1]
|
||||
if len(sys.argv) > 2:
|
||||
CONFIG["delay_threshold"] = float(sys.argv[2])
|
||||
if len(sys.argv) > 3:
|
||||
CONFIG["ping_interval"] = float(sys.argv[3])
|
||||
|
||||
# 创建并启动监控器
|
||||
monitor = NetworkMonitor(
|
||||
target_ip=CONFIG["target_ip"],
|
||||
delay_threshold=CONFIG["delay_threshold"],
|
||||
ping_interval=CONFIG["ping_interval"]
|
||||
)
|
||||
|
||||
monitor.monitor()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user