175 lines
5.1 KiB
Python
175 lines
5.1 KiB
Python
"""
|
||
线路状态管理模块
|
||
|
||
负责保存和读取每个线路的最新状态信息到文件中
|
||
"""
|
||
import json
|
||
import os
|
||
from datetime import datetime
|
||
from typing import Dict, List, Optional, Any
|
||
from pathlib import Path
|
||
from .config import client_infos, settings
|
||
|
||
class LineStatusManager:
|
||
"""线路状态管理器"""
|
||
|
||
def __init__(self, status_file: str = "line_status.json"):
|
||
"""
|
||
初始化状态管理器
|
||
|
||
Args:
|
||
status_file: 状态文件路径
|
||
"""
|
||
self.status_file = Path(status_file)
|
||
self._ensure_status_file()
|
||
|
||
def _ensure_status_file(self):
|
||
"""确保状态文件存在"""
|
||
if not self.status_file.exists():
|
||
self._save_status({})
|
||
|
||
def _load_status(self) -> Dict[str, Any]:
|
||
"""从文件加载状态"""
|
||
try:
|
||
with open(self.status_file, 'r', encoding='utf-8') as f:
|
||
return json.load(f)
|
||
except (FileNotFoundError, json.JSONDecodeError):
|
||
return {}
|
||
|
||
def _save_status(self, status: Dict[str, Any]):
|
||
"""保存状态到文件"""
|
||
with open(self.status_file, 'w', encoding='utf-8') as f:
|
||
json.dump(status, f, ensure_ascii=False, indent=2)
|
||
|
||
|
||
def get_line_status(self, line_id: int) -> Dict[str, Any]:
|
||
"""
|
||
获取指定线路的状态
|
||
|
||
Args:
|
||
line_id: 线路ID
|
||
|
||
Returns:
|
||
Dict[str, Any]: 线路状态信息
|
||
"""
|
||
status = self._load_status()
|
||
line_key = str(line_id)
|
||
|
||
if line_key not in status:
|
||
return {
|
||
"id": line_id,
|
||
"current_ip": None,
|
||
"last_rotate_time": None,
|
||
"status": "inactive",
|
||
"edge_device": None,
|
||
"geo_location": None,
|
||
"rotate_count": 0
|
||
}
|
||
|
||
return status[line_key]
|
||
|
||
def update_line_status(self, line_id: int, **kwargs):
|
||
"""
|
||
更新线路状态
|
||
|
||
Args:
|
||
line_id: 线路ID
|
||
**kwargs: 要更新的状态字段
|
||
"""
|
||
status = self._load_status()
|
||
line_key = str(line_id)
|
||
|
||
if line_key not in status:
|
||
status[line_key] = {
|
||
"id": line_id,
|
||
"current_ip": None,
|
||
"last_rotate_time": None,
|
||
"status": "inactive",
|
||
"edge_device": None,
|
||
"geo_location": None,
|
||
"rotate_count": 0
|
||
}
|
||
|
||
# 更新字段
|
||
for key, value in kwargs.items():
|
||
status[line_key][key] = value
|
||
|
||
# 更新时间戳
|
||
status[line_key]["last_update_time"] = datetime.now().isoformat()
|
||
|
||
self._save_status(status)
|
||
|
||
def get_all_lines_status(self) -> List[Dict[str, Any]]:
|
||
"""
|
||
获取所有线路状态（仅已保存的状态）
|
||
|
||
Returns:
|
||
List[Dict[str, Any]]: 所有线路状态列表，按线路ID排序
|
||
"""
|
||
status = self._load_status()
|
||
lines = list(status.values())
|
||
|
||
# 按线路ID排序
|
||
lines.sort(key=lambda x: x.get("id", 0))
|
||
|
||
return lines
|
||
|
||
def get_all_lines_with_defaults(self) -> List[Dict[str, Any]]:
|
||
"""
|
||
获取所有线路状态，包括默认配置的线路
|
||
|
||
Returns:
|
||
List[Dict[str, Any]]: 所有线路状态列表，包括未配置的线路
|
||
"""
|
||
|
||
|
||
# 获取已保存的状态
|
||
saved_status = self._load_status()
|
||
lines = []
|
||
|
||
# 遍历所有配置的线路
|
||
for port_id, default_ip in client_infos.items():
|
||
line_id = int(port_id) + settings.port_diff
|
||
line_key = str(line_id)
|
||
|
||
if line_key in saved_status:
|
||
# 使用已保存的状态
|
||
line_data = saved_status[line_key].copy()
|
||
else:
|
||
# 使用默认状态
|
||
line_data = {
|
||
"id": line_id,
|
||
"current_ip": default_ip,
|
||
"last_rotate_time": None,
|
||
"status": "inactive",
|
||
"edge_device": None,
|
||
"geo_location": None,
|
||
"rotate_count": 0
|
||
}
|
||
|
||
lines.append(line_data)
|
||
|
||
# 按线路ID排序
|
||
lines.sort(key=lambda x: x.get("id", 0))
|
||
|
||
return lines
|
||
|
||
def increment_rotate_count(self, line_id: int):
|
||
"""增加轮换次数计数"""
|
||
current_status = self.get_line_status(line_id)
|
||
self.update_line_status(
|
||
line_id,
|
||
rotate_count=current_status.get("rotate_count", 0) + 1
|
||
)
|
||
|
||
|
||
# 创建全局状态管理器实例
|
||
status_manager = LineStatusManager()
|
||
|
||
# 使用示例：
|
||
# 获取所有线路状态（仅已保存的）
|
||
# all_lines = status_manager.get_all_lines_status()
|
||
#
|
||
# 获取所有线路状态（包括默认配置）
|
||
# all_lines_with_defaults = status_manager.get_all_lines_with_defaults()
|