226 lines
7.8 KiB
Python
226 lines
7.8 KiB
Python
import json
|
|
import os
|
|
from typing import Dict, List, Optional
|
|
from logger_config import global_logger as logger
|
|
|
|
class ConfigManager:
|
|
"""
|
|
配置管理器，负责加载、保存和管理应用配置
|
|
"""
|
|
|
|
def __init__(self, config_file: str = 'config.json'):
|
|
"""
|
|
初始化配置管理器
|
|
|
|
Args:
|
|
config_file (str): 配置文件路径
|
|
"""
|
|
self.config_file = config_file
|
|
self.default_config = {
|
|
'duration': 30,
|
|
'sites': [
|
|
{'name': '百度', 'url': 'https://www.baidu.com'},
|
|
{'name': '新浪', 'url': 'https://www.sina.com.cn'}
|
|
]
|
|
}
|
|
# 在初始化时加载配置并保存到实例属性
|
|
self._config = self._load_config_from_file()
|
|
|
|
def _load_config_from_file(self) -> Dict:
|
|
"""
|
|
从文件加载配置的内部方法
|
|
|
|
Returns:
|
|
dict: 配置数据
|
|
"""
|
|
try:
|
|
if os.path.exists(self.config_file):
|
|
with open(self.config_file, 'r', encoding='utf-8') as f:
|
|
config = json.load(f)
|
|
logger.info(f"成功加载配置文件: {self.config_file}")
|
|
|
|
# 确保配置中包含duration字段
|
|
if 'duration' not in config:
|
|
logger.info("配置中没有duration字段，使用默认值30秒")
|
|
config['duration'] = self.default_config['duration']
|
|
|
|
# 检查是否存在网站配置，如果不存在则使用默认示例网站并保存
|
|
sites = config.get('sites', [])
|
|
if not sites:
|
|
logger.warning("配置中没有设置任何网站，使用默认示例网站")
|
|
config['sites'] = self.default_config['sites']
|
|
# 保存默认网站到配置中
|
|
try:
|
|
self._save_config_to_file(config)
|
|
logger.info("已保存默认网站配置")
|
|
except Exception as e:
|
|
logger.error(f"保存默认网站配置失败: {e}")
|
|
|
|
return config
|
|
else:
|
|
logger.warning(f"配置文件不存在: {self.config_file}")
|
|
logger.info("使用默认配置")
|
|
|
|
# 创建并保存默认配置
|
|
default_config = self.default_config.copy()
|
|
try:
|
|
self._save_config_to_file(default_config)
|
|
logger.info("已创建默认配置文件")
|
|
except Exception as e:
|
|
logger.error(f"保存默认配置文件失败: {e}")
|
|
|
|
return default_config
|
|
except json.JSONDecodeError as e:
|
|
logger.error(f"配置文件格式错误: {e}")
|
|
logger.info("使用默认配置")
|
|
|
|
# 使用默认配置并保存
|
|
default_config = self.default_config.copy()
|
|
try:
|
|
self._save_config_to_file(default_config)
|
|
logger.info("已覆盖错误配置文件")
|
|
except Exception as e:
|
|
logger.error(f"保存默认配置文件失败: {e}")
|
|
|
|
return default_config
|
|
except Exception as e:
|
|
logger.error(f"加载配置文件时发生错误: {e}")
|
|
logger.info("使用默认配置")
|
|
return self.default_config.copy()
|
|
|
|
def load_config(self) -> Dict:
|
|
"""
|
|
获取当前配置（从实例属性中返回）
|
|
|
|
Returns:
|
|
dict: 配置数据
|
|
"""
|
|
return self._config.copy() # 返回配置的副本，避免外部直接修改
|
|
|
|
def _save_config_to_file(self, config: Dict) -> bool:
|
|
"""
|
|
保存配置到文件的内部方法
|
|
|
|
Args:
|
|
config (dict): 要保存的配置数据
|
|
|
|
Returns:
|
|
bool: 保存是否成功
|
|
"""
|
|
try:
|
|
# 确保配置中包含必要的键
|
|
if 'sites' not in config:
|
|
config['sites'] = []
|
|
|
|
with open(self.config_file, 'w', encoding='utf-8') as f:
|
|
json.dump(config, f, ensure_ascii=False, indent=2)
|
|
|
|
logger.info(f"成功保存配置到文件: {self.config_file}")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"保存配置文件时发生错误: {e}")
|
|
return False
|
|
|
|
def save_config(self, config: Dict) -> bool:
|
|
"""
|
|
保存配置并更新实例属性
|
|
|
|
Args:
|
|
config (dict): 要保存的配置数据
|
|
|
|
Returns:
|
|
bool: 保存是否成功
|
|
"""
|
|
if self._save_config_to_file(config):
|
|
# 保存成功后更新实例属性
|
|
self._config = config
|
|
return True
|
|
return False
|
|
|
|
def get_sites(self) -> List[Dict]:
|
|
"""
|
|
获取所有网站配置
|
|
|
|
Returns:
|
|
list: 网站配置列表
|
|
"""
|
|
sites = self._config.get('sites', [])
|
|
logger.info(f"获取到 {len(sites)} 个网站配置")
|
|
return sites.copy() # 返回副本以避免直接修改内部数据
|
|
|
|
def add_site(self, site_config: Dict) -> bool:
|
|
"""
|
|
添加网站配置
|
|
|
|
Args:
|
|
site_config (dict): 网站配置
|
|
|
|
Returns:
|
|
bool: 添加是否成功
|
|
"""
|
|
if not isinstance(site_config, dict) or 'name' not in site_config or 'url' not in site_config:
|
|
logger.error("网站配置格式错误，必须包含name和url字段")
|
|
return False
|
|
|
|
# 检查是否已存在同名网站
|
|
for site in self._config.get('sites', []):
|
|
if site.get('name') == site_config.get('name'):
|
|
logger.warning(f"网站 '{site_config.get('name')}' 已存在")
|
|
return False
|
|
|
|
# 创建配置副本并添加新网站
|
|
config = self._config.copy()
|
|
sites = config.get('sites', [])
|
|
sites.append(site_config)
|
|
config['sites'] = sites
|
|
return self.save_config(config)
|
|
|
|
def update_site(self, site_name: str, updated_config: Dict) -> bool:
|
|
"""
|
|
更新网站配置
|
|
|
|
Args:
|
|
site_name (str): 网站名称
|
|
updated_config (dict): 更新后的配置
|
|
|
|
Returns:
|
|
bool: 更新是否成功
|
|
"""
|
|
# 创建配置副本
|
|
config = self._config.copy()
|
|
sites = config.get('sites', [])
|
|
|
|
for i, site in enumerate(sites):
|
|
if site.get('name') == site_name:
|
|
# 保留原有的name和url，只更新其他字段
|
|
updated_site = site.copy()
|
|
updated_site.update(updated_config)
|
|
sites[i] = updated_site
|
|
config['sites'] = sites
|
|
return self.save_config(config)
|
|
|
|
logger.warning(f"找不到名称为 '{site_name}' 的网站配置")
|
|
return False
|
|
|
|
def delete_site(self, site_name: str) -> bool:
|
|
"""
|
|
删除网站配置
|
|
|
|
Args:
|
|
site_name (str): 网站名称
|
|
|
|
Returns:
|
|
bool: 删除是否成功
|
|
"""
|
|
# 创建配置副本
|
|
config = self._config.copy()
|
|
sites = config.get('sites', [])
|
|
|
|
new_sites = [site for site in sites if site.get('name') != site_name]
|
|
|
|
if len(new_sites) == len(sites):
|
|
logger.warning(f"找不到名称为 '{site_name}' 的网站配置")
|
|
return False
|
|
|
|
config['sites'] = new_sites
|
|
return self.save_config(config) |