""" 配置管理模块 负责管理服务器配置参数 """ import time import json import os from typing import Dict, Any, List, Optional class ConfigManager: """ 配置管理器 负责管理服务器配置参数 """ def __init__(self, plugin): """ 初始化配置管理器 Args: plugin: 服务器架构插件实例 """ self.plugin = plugin self.enabled = False self.initialized = False # 配置文件路径 self.config_file_path = "/home/hello/EG/plugins/user/server_architecture/config/server_config.json" # 默认配置 self.default_config = { "server": { "host": "0.0.0.0", "port": 8080, "max_connections": 1000, "enable_ssl": False, "ssl_cert_file": "", "ssl_key_file": "", "connection_timeout": 30.0, "heartbeat_interval": 5.0 }, "client": { "enable_authentication": True, "auth_timeout": 30.0, "max_clients": 1000, "enable_heartbeat": True, "heartbeat_interval": 30.0, "enable_rate_limiting": True, "max_requests_per_second": 100, "enable_banning": True, "ban_duration": 3600 }, "room": { "max_rooms": 100, "max_clients_per_room": 50, "enable_room_persistence": False, "room_timeout": 3600, "enable_room_passwords": True }, "storage": { "database_path": "/home/hello/EG/plugins/user/server_architecture/data/server_database.db", "enable_caching": True, "cache_size": 1000, "enable_compression": True, "backup_interval": 3600, "enable_auto_backup": True }, "loadbalancer": { "enable_load_balancing": True, "balancing_algorithm": "round_robin", "health_check_interval": 30.0, "enable_health_check": True }, "monitoring": { "enable_monitoring": True, "monitoring_interval": 5.0, "log_level": "INFO", "enable_file_logging": True, "log_file_path": "/home/hello/EG/plugins/user/server_architecture/logs/server_monitor.log" }, "security": { "enable_encryption": False, "enable_message_authentication": True, "enable_session_keys": True, "session_key_lifetime": 3600 }, "api": { "enable_api": True, "api_port": 8081, "enable_cors": True, "allowed_origins": ["*"] } } # 当前配置 self.current_config = self.default_config.copy() # 配置状态 self.config_state = { "last_loaded": 0.0, "last_saved": 0.0, "config_version": "1.0.0", "is_dirty": False } # 配置统计 self.config_stats = { "loads": 0, "saves": 0, "updates": 0, "errors": 0 } # 配置监听器 self.config_listeners = {} # 回调函数 self.config_callbacks = { "config_loaded": [], "config_saved": [], "config_updated": [], "config_error": [] } print("✓ 配置管理器已创建") def initialize(self) -> bool: """ 初始化配置管理器 Returns: 是否初始化成功 """ try: print("正在初始化配置管理器...") # 创建配置目录 config_dir = os.path.dirname(self.config_file_path) if not os.path.exists(config_dir): os.makedirs(config_dir) # 创建日志目录 if "monitoring" in self.default_config: log_dir = os.path.dirname(self.default_config["monitoring"]["log_file_path"]) if not os.path.exists(log_dir): os.makedirs(log_dir) # 创建数据目录 if "storage" in self.default_config: data_dir = os.path.dirname(self.default_config["storage"]["database_path"]) if not os.path.exists(data_dir): os.makedirs(data_dir) # 加载配置 self.load_config() self.initialized = True print("✓ 配置管理器初始化完成") return True except Exception as e: print(f"✗ 配置管理器初始化失败: {e}") self.config_stats["errors"] += 1 import traceback traceback.print_exc() return False def enable(self) -> bool: """ 启用配置管理器 Returns: 是否启用成功 """ try: if not self.initialized: print("✗ 配置管理器未初始化") return False self.enabled = True print("✓ 配置管理器已启用") return True except Exception as e: print(f"✗ 配置管理器启用失败: {e}") self.config_stats["errors"] += 1 import traceback traceback.print_exc() return False def disable(self): """禁用配置管理器""" try: self.enabled = False # 保存配置（如果有更改） if self.config_state["is_dirty"]: self.save_config() print("✓ 配置管理器已禁用") except Exception as e: print(f"✗ 配置管理器禁用失败: {e}") self.config_stats["errors"] += 1 import traceback traceback.print_exc() def finalize(self): """清理配置管理器资源""" try: # 禁用配置管理器 if self.enabled: self.disable() # 清理回调 self.config_callbacks.clear() self.config_listeners.clear() self.initialized = False print("✓ 配置管理器资源已清理") except Exception as e: print(f"✗ 配置管理器资源清理失败: {e}") import traceback traceback.print_exc() def update(self, dt: float): """ 更新配置管理器状态 Args: dt: 时间增量（秒） """ try: if not self.enabled: return # 检查配置文件是否被外部修改 self._check_config_file_changes() except Exception as e: print(f"✗ 配置管理器更新失败: {e}") self.config_stats["errors"] += 1 import traceback traceback.print_exc() def _check_config_file_changes(self): """检查配置文件是否被外部修改""" try: if not os.path.exists(self.config_file_path): return last_modified = os.path.getmtime(self.config_file_path) if last_modified > self.config_state["last_loaded"]: # 配置文件已被修改，重新加载 print("检测到配置文件被外部修改，正在重新加载...") self.load_config() except Exception as e: print(f"✗ 配置文件变更检查失败: {e}") self.config_stats["errors"] += 1 def load_config(self) -> bool: """ 加载配置 Returns: 是否加载成功 """ try: # 检查配置文件是否存在 if os.path.exists(self.config_file_path): with open(self.config_file_path, 'r', encoding='utf-8') as f: loaded_config = json.load(f) # 合并配置（使用默认配置作为基础） self.current_config = self._merge_config(self.default_config, loaded_config) print(f"✓ 配置已从文件加载: {self.config_file_path}") else: # 使用默认配置 self.current_config = self.default_config.copy() print("✓ 使用默认配置") self.config_state["last_loaded"] = time.time() self.config_state["is_dirty"] = False self.config_stats["loads"] += 1 # 触发配置加载回调 self._trigger_config_callback("config_loaded", { "config": self.current_config, "timestamp": self.config_state["last_loaded"] }) return True except Exception as e: print(f"✗ 配置加载失败: {e}") self.config_stats["errors"] += 1 # 使用默认配置 self.current_config = self.default_config.copy() return False def _merge_config(self, base_config: Dict[str, Any], override_config: Dict[str, Any]) -> Dict[str, Any]: """ 合并配置（递归） Args: base_config: 基础配置 override_config: 覆盖配置 Returns: 合并后的配置 """ try: merged = base_config.copy() for key, value in override_config.items(): if key in merged and isinstance(merged[key], dict) and isinstance(value, dict): # 递归合并嵌套字典 merged[key] = self._merge_config(merged[key], value) else: # 直接覆盖 merged[key] = value return merged except Exception as e: print(f"✗ 配置合并失败: {e}") self.config_stats["errors"] += 1 return base_config def save_config(self) -> bool: """ 保存配置 Returns: 是否保存成功 """ try: # 创建配置目录（如果不存在） config_dir = os.path.dirname(self.config_file_path) if not os.path.exists(config_dir): os.makedirs(config_dir) # 保存配置到文件 with open(self.config_file_path, 'w', encoding='utf-8') as f: json.dump(self.current_config, f, indent=2, ensure_ascii=False) self.config_state["last_saved"] = time.time() self.config_state["is_dirty"] = False self.config_stats["saves"] += 1 # 触发配置保存回调 self._trigger_config_callback("config_saved", { "config": self.current_config, "timestamp": self.config_state["last_saved"] }) print(f"✓ 配置已保存到文件: {self.config_file_path}") return True except Exception as e: print(f"✗ 配置保存失败: {e}") self.config_stats["errors"] += 1 return False def get_config(self, section: str = None, key: str = None, default_value: Any = None) -> Any: """ 获取配置值 Args: section: 配置节（可选） key: 配置键（可选） default_value: 默认值 Returns: 配置值 """ try: if section is None: return self.current_config.copy() if section not in self.current_config: return default_value if key is None: return self.current_config[section].copy() if key not in self.current_config[section]: return default_value return self.current_config[section][key] except Exception as e: print(f"✗ 获取配置失败: {e}") self.config_stats["errors"] += 1 return default_value def set_config(self, section: str, key: str, value: Any) -> bool: """ 设置配置值 Args: section: 配置节 key: 配置键 value: 配置值 Returns: 是否设置成功 """ try: if section not in self.current_config: self.current_config[section] = {} self.current_config[section][key] = value self.config_state["is_dirty"] = True self.config_stats["updates"] += 1 # 通知配置监听器 self._notify_config_listeners(section, key, value) # 触发配置更新回调 self._trigger_config_callback("config_updated", { "section": section, "key": key, "value": value, "timestamp": time.time() }) return True except Exception as e: print(f"✗ 设置配置失败: {e}") self.config_stats["errors"] += 1 return False def update_config_section(self, section: str, config_data: Dict[str, Any]) -> bool: """ 更新配置节 Args: section: 配置节 config_data: 配置数据 Returns: 是否更新成功 """ try: if section not in self.current_config: self.current_config[section] = {} self.current_config[section].update(config_data) self.config_state["is_dirty"] = True self.config_stats["updates"] += 1 # 通知配置监听器 for key, value in config_data.items(): self._notify_config_listeners(section, key, value) # 触发配置更新回调 self._trigger_config_callback("config_updated", { "section": section, "config_data": config_data, "timestamp": time.time() }) return True except Exception as e: print(f"✗ 更新配置节失败: {e}") self.config_stats["errors"] += 1 return False def reset_config(self) -> bool: """ 重置配置为默认值 Returns: 是否重置成功 """ try: self.current_config = self.default_config.copy() self.config_state["is_dirty"] = True # 通知所有配置监听器 for section_name, section_data in self.current_config.items(): if isinstance(section_data, dict): for key, value in section_data.items(): self._notify_config_listeners(section_name, key, value) print("✓ 配置已重置为默认值") return True except Exception as e: print(f"✗ 配置重置失败: {e}") self.config_stats["errors"] += 1 return False def register_config_listener(self, section: str, key: str, listener: callable): """ 注册配置监听器 Args: section: 配置节 key: 配置键 listener: 监听器函数 """ try: listener_key = f"{section}.{key}" if listener_key not in self.config_listeners: self.config_listeners[listener_key] = [] self.config_listeners[listener_key].append(listener) print(f"✓ 配置监听器已注册: {listener_key}") except Exception as e: print(f"✗ 配置监听器注册失败: {e}") self.config_stats["errors"] += 1 def unregister_config_listener(self, section: str, key: str, listener: callable): """ 注销配置监听器 Args: section: 配置节 key: 配置键 listener: 监听器函数 """ try: listener_key = f"{section}.{key}" if listener_key in self.config_listeners: if listener in self.config_listeners[listener_key]: self.config_listeners[listener_key].remove(listener) print(f"✓ 配置监听器已注销: {listener_key}") except Exception as e: print(f"✗ 配置监听器注销失败: {e}") self.config_stats["errors"] += 1 def _notify_config_listeners(self, section: str, key: str, value: Any): """ 通知配置监听器 Args: section: 配置节 key: 配置键 value: 配置值 """ try: listener_key = f"{section}.{key}" if listener_key in self.config_listeners: for listener in self.config_listeners[listener_key]: try: listener(section, key, value) except Exception as e: print(f"✗ 配置监听器执行失败: {listener_key} - {e}") except Exception as e: print(f"✗ 配置监听器通知失败: {e}") self.config_stats["errors"] += 1 def get_config_stats(self) -> Dict[str, Any]: """ 获取配置统计信息 Returns: 配置统计字典 """ return { "state": self.config_state.copy(), "stats": self.config_stats.copy(), "config_sections": list(self.current_config.keys()) } def reset_stats(self): """重置配置统计信息""" try: self.config_stats = { "loads": 0, "saves": 0, "updates": 0, "errors": 0 } print("✓ 配置统计信息已重置") except Exception as e: print(f"✗ 配置统计信息重置失败: {e}") def validate_config(self) -> bool: """ 验证配置 Returns: 配置是否有效 """ try: # 检查必需的配置项 required_sections = ["server", "client", "room", "storage"] for section in required_sections: if section not in self.current_config: print(f"✗ 缺少必需的配置节: {section}") return False # 检查服务器配置 server_config = self.current_config["server"] if "port" not in server_config or not isinstance(server_config["port"], int): print("✗ 无效的服务器端口配置") return False if "max_connections" not in server_config or not isinstance(server_config["max_connections"], int): print("✗ 无效的最大连接数配置") return False # 检查客户端配置 client_config = self.current_config["client"] if "max_clients" not in client_config or not isinstance(client_config["max_clients"], int): print("✗ 无效的最大客户端数配置") return False # 检查存储配置 storage_config = self.current_config["storage"] if "database_path" not in storage_config or not isinstance(storage_config["database_path"], str): print("✗ 无效的数据库路径配置") return False print("✓ 配置验证通过") return True except Exception as e: print(f"✗ 配置验证失败: {e}") self.config_stats["errors"] += 1 return False def export_config(self, file_path: str) -> bool: """ 导出配置到文件 Args: file_path: 导出文件路径 Returns: 是否导出成功 """ try: # 创建导出目录 export_dir = os.path.dirname(file_path) if not os.path.exists(export_dir): os.makedirs(export_dir) # 导出配置 with open(file_path, 'w', encoding='utf-8') as f: json.dump(self.current_config, f, indent=2, ensure_ascii=False) print(f"✓ 配置已导出到: {file_path}") return True except Exception as e: print(f"✗ 配置导出失败: {e}") self.config_stats["errors"] += 1 return False def import_config(self, file_path: str) -> bool: """ 从文件导入配置 Args: file_path: 导入文件路径 Returns: 是否导入成功 """ try: if not os.path.exists(file_path): print(f"✗ 配置文件不存在: {file_path}") return False # 读取配置文件 with open(file_path, 'r', encoding='utf-8') as f: imported_config = json.load(f) # 验证配置 if not isinstance(imported_config, dict): print("✗ 无效的配置文件格式") return False # 合并配置 self.current_config = self._merge_config(self.default_config, imported_config) self.config_state["is_dirty"] = True # 通知配置监听器 for section_name, section_data in self.current_config.items(): if isinstance(section_data, dict): for key, value in section_data.items(): self._notify_config_listeners(section_name, key, value) print(f"✓ 配置已从文件导入: {file_path}") return True except Exception as e: print(f"✗ 配置导入失败: {e}") self.config_stats["errors"] += 1 return False def _trigger_config_callback(self, callback_type: str, data: Dict[str, Any]): """ 触发配置回调 Args: callback_type: 回调类型 data: 回调数据 """ try: if callback_type in self.config_callbacks: for callback in self.config_callbacks[callback_type]: try: callback(data) except Exception as e: print(f"✗ 配置回调执行失败: {callback_type} - {e}") except Exception as e: print(f"✗ 配置回调触发失败: {e}") def register_config_callback(self, callback_type: str, callback: callable): """ 注册配置回调 Args: callback_type: 回调类型 callback: 回调函数 """ try: if callback_type in self.config_callbacks: self.config_callbacks[callback_type].append(callback) print(f"✓ 配置回调已注册: {callback_type}") else: print(f"✗ 无效的回调类型: {callback_type}") except Exception as e: print(f"✗ 配置回调注册失败: {e}") def unregister_config_callback(self, callback_type: str, callback: callable): """ 注销配置回调 Args: callback_type: 回调类型 callback: 回调函数 """ try: if callback_type in self.config_callbacks: if callback in self.config_callbacks[callback_type]: self.config_callbacks[callback_type].remove(callback) print(f"✓ 配置回调已注销: {callback_type}") else: print(f"✗ 无效的回调类型: {callback_type}") except Exception as e: print(f"✗ 配置回调注销失败: {e}")