Zhiwen Assistant MCP

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 配置管理模块 支持YAML/JSON配置文件、配置验证和默认值处理 """ import os import yaml import json from typing import Dict, Any, Optional, Union from pathlib import Path import logging # 默认配置 DEFAULT_CONFIG = { 'server': { 'name': '文件夹文档生成MCP服务器', 'version': '2.0.0', 'transport': 'stdio', 'port': 8080 }, 'options': { 'exclude_dirs': ['.git', '__pycache__', 'node_modules', '.trae', '.vscode', '.idea'], 'force_update': False, 'output_file': 'folder_structure_mindmap.md', 'max_depth': 10, 'include_hidden': False }, 'performance': { 'cache_ttl': 3600, 'max_concurrent_operations': 5, 'enable_performance_monitoring': True }, 'security': { 'allowed_paths': [], 'enable_security_validation': True, 'scan_sensitive_content': True, 'block_dangerous_files': True }, 'logging': { 'level': 'INFO', 'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s', 'file': 'mcp_server.log' } } class ConfigManager: """配置管理器""" def __init__(self, config_file: Optional[str] = None): self.logger = logging.getLogger(__name__) self.config_file = config_file or self._find_config_file() self.config = {} self.load_config() def _find_config_file(self) -> Optional[str]: """查找配置文件""" possible_files = [ 'mcp_config.yaml', 'mcp_config.json', 'config/mcp_config.yaml', 'config/mcp_config.json' ] for file_name in possible_files: file_path = Path(file_name) if file_path.exists(): return str(file_path) return None def load_config(self) -> None: """加载配置""" # 首先加载默认配置 self.config = DEFAULT_CONFIG.copy() if not self.config_file: self.logger.info("未找到配置文件，使用默认配置") return try: config_path = Path(self.config_file) if config_path.suffix.lower() == '.yaml' or config_path.suffix.lower() == '.yml': with open(config_path, 'r', encoding='utf-8') as f: file_config = yaml.safe_load(f) elif config_path.suffix.lower() == '.json': with open(config_path, 'r', encoding='utf-8') as f: file_config = json.load(f) else: self.logger.warning(f"不支持的配置文件格式: {config_path.suffix}") return # 合并配置（文件配置覆盖默认配置） self._merge_config(self.config, file_config) self.logger.info(f"成功加载配置文件: {self.config_file}") except Exception as e: self.logger.error(f"加载配置文件失败: {e}") def _merge_config(self, base: Dict[str, Any], overlay: Dict[str, Any]) -> None: """递归合并配置""" for key, value in overlay.items(): if key in base and isinstance(base[key], dict) and isinstance(value, dict): self._merge_config(base[key], value) else: base[key] = value def get(self, key: str, default: Any = None) -> Any: """ 获取配置值 Args: key: 配置键，支持点分隔符嵌套访问（如 'server.port'） default: 默认值 Returns: 配置值 """ keys = key.split('.') value = self.config try: for k in keys: value = value[k] return value except (KeyError, TypeError): return default def set(self, key: str, value: Any) -> None: """ 设置配置值 Args: key: 配置键，支持点分隔符嵌套设置 value: 配置值 """ keys = key.split('.') config = self.config # 导航到目标位置 for k in keys[:-1]: if k not in config: config[k] = {} config = config[k] # 设置值 config[keys[-1]] = value def save_config(self, file_path: Optional[str] = None) -> None: """ 保存配置到文件 Args: file_path: 保存路径，如果为None则使用当前配置文件路径 """ target_file = file_path or self.config_file if not target_file: raise ValueError("未指定配置文件路径") try: config_path = Path(target_file) config_path.parent.mkdir(parents=True, exist_ok=True) if config_path.suffix.lower() in ['.yaml', '.yml']: with open(config_path, 'w', encoding='utf-8') as f: yaml.dump(self.config, f, default_flow_style=False, allow_unicode=True) elif config_path.suffix.lower() == '.json': with open(config_path, 'w', encoding='utf-8') as f: json.dump(self.config, f, indent=2, ensure_ascii=False) else: raise ValueError(f"不支持的配置文件格式: {config_path.suffix}") self.logger.info(f"配置已保存到: {target_file}") except Exception as e: self.logger.error(f"保存配置文件失败: {e}") raise def validate_config(self) -> Dict[str, Any]: """ 验证配置 Returns: 验证结果 """ issues = [] # 验证必需的配置项 required_keys = [ 'server.name', 'server.transport', 'options.exclude_dirs', 'performance.cache_ttl' ] for key in required_keys: if self.get(key) is None: issues.append(f"缺少必需的配置项: {key}") # 验证配置值的合理性 port = self.get('server.port') if port is not None and (not isinstance(port, int) or port < 1 or port > 65535): issues.append(f"无效的端口号: {port}") cache_ttl = self.get('performance.cache_ttl') if cache_ttl is not None and (not isinstance(cache_ttl, int) or cache_ttl < 0): issues.append(f"无效的缓存TTL: {cache_ttl}") exclude_dirs = self.get('options.exclude_dirs') if exclude_dirs is not None and not isinstance(exclude_dirs, list): issues.append("exclude_dirs 必须是列表类型") return { 'valid': len(issues) == 0, 'issues': issues } def reload(self) -> None: """重新加载配置""" self.load_config() self.logger.info("配置已重新加载") # 全局配置管理器实例 _config_manager: Optional[ConfigManager] = None def get_config_manager(config_file: Optional[str] = None) -> ConfigManager: """ 获取全局配置管理器实例 Args: config_file: 配置文件路径 Returns: 配置管理器实例 """ global _config_manager if _config_manager is None: _config_manager = ConfigManager(config_file) return _config_manager def reload_config() -> None: """重新加载全局配置""" global _config_manager if _config_manager is not None: _config_manager.reload() # 便捷函数 def get_config(key: str, default: Any = None) -> Any: """获取配置值的便捷函数""" return get_config_manager().get(key, default) def set_config(key: str, value: Any) -> None: """设置配置值的便捷函数""" get_config_manager().set(key, value)