Zhiwen Assistant MCP

""" 用户配置管理模块 提供交互式配置、配置验证、配置持久化等功能 """ import os import json import yaml from typing import Dict, Any, Optional, List, Union, Callable from dataclasses import dataclass, asdict from pathlib import Path import copy from .interaction import InteractionManager, confirm, select, input_text, info, success, warning, error, ValidationError @dataclass class ConfigOption: """配置选项定义""" key: str label: str description: str = "" type: str = "string" # string, int, float, bool, list, dict, choice, path, file, dir default: Any = None required: bool = False validator: Optional[Callable[[Any], bool]] = None choices: Optional[List[Any]] = None help_text: str = "" depends_on: Optional[str] = None # 依赖的其他配置项 condition: Optional[Callable[[Dict[str, Any]], bool]] = None # 条件判断 def validate(self, value: Any) -> bool: """验证值是否有效""" # 类型检查 if self.type == "string" and not isinstance(value, str): return False elif self.type == "int" and not isinstance(value, int): return False elif self.type == "float" and not isinstance(value, (int, float)): return False elif self.type == "bool" and not isinstance(value, bool): return False elif self.type == "list" and not isinstance(value, list): return False elif self.type == "dict" and not isinstance(value, dict): return False elif self.type == "choice" and self.choices and value not in self.choices: return False elif self.type == "path" and not isinstance(value, str): return False elif self.type == "file" and not (isinstance(value, str) and os.path.isfile(value)): return False elif self.type == "dir" and not (isinstance(value, str) and os.path.isdir(value)): return False # 自定义验证器 if self.validator: return self.validator(value) return True class ConfigManager: """配置管理器""" def __init__(self, config_file: str = "config/user_config.yaml", interaction_manager: Optional[InteractionManager] = None): self.config_file = Path(config_file) self.config_dir = self.config_file.parent self.interaction = interaction_manager or InteractionManager() # 配置项定义 self.options: Dict[str, ConfigOption] = {} # 当前配置 self.config: Dict[str, Any] = {} # 确保配置目录存在 self.config_dir.mkdir(parents=True, exist_ok=True) def add_option(self, option: ConfigOption) -> None: """添加配置选项""" self.options[option.key] = option def add_options(self, options: List[ConfigOption]) -> None: """批量添加配置选项""" for option in options: self.add_option(option) def load_config(self) -> Dict[str, Any]: """加载配置文件""" if self.config_file.exists(): try: with open(self.config_file, 'r', encoding='utf-8') as f: if self.config_file.suffix.lower() in ['.yaml', '.yml']: loaded_config = yaml.safe_load(f) else: loaded_config = json.load(f) # 合并默认值 self.config = self._merge_with_defaults(loaded_config or {}) self.interaction.info("配置加载成功", f"已从 {self.config_file} 加载配置") return self.config except (RuntimeError, ValueError) as e: self.interaction.error("配置加载失败", f"无法加载配置文件: {e}") return self._get_defaults() else: self.interaction.info("配置文件不存在", "将使用默认配置") return self._get_defaults() def save_config(self) -> bool: """保存配置文件""" try: # 确保配置目录存在 self.config_dir.mkdir(parents=True, exist_ok=True) with open(self.config_file, 'w', encoding='utf-8') as f: if self.config_file.suffix.lower() in ['.yaml', '.yml']: yaml.dump(self.config, f, default_flow_style=False, allow_unicode=True, indent=2) else: json.dump(self.config, f, ensure_ascii=False, indent=2) self.interaction.success("配置保存成功", f"配置已保存到 {self.config_file}") return True except (RuntimeError, ValueError) as e: self.interaction.error("配置保存失败", f"无法保存配置文件: {e}") return False def get(self, key: str, default: Any = None) -> Any: """获取配置值""" return self.config.get(key, default) def set(self, key: str, value: Any) -> None: """设置配置值""" option = self.options.get(key) if option and not option.validate(value): raise ValidationError(key, str(value), f"值不符合配置项要求") self.config[key] = value def update(self, config_dict: Dict[str, Any]) -> None: """批量更新配置""" for key, value in config_dict.items(): self.set(key, value) def interactive_config(self, show_all: bool = False) -> Dict[str, Any]: """交互式配置""" self.interaction.info("开始交互式配置", "将逐项设置配置参数") # 加载现有配置 self.load_config() # 获取需要配置的选项 options_to_configure = self._get_options_to_configure(show_all) if not options_to_configure: self.interaction.info("无需配置", "所有配置项已设置或使用默认值") return self.config # 逐项配置 for key, option in options_to_configure.items(): try: value = self._configure_option(option) if value is not None: self.set(key, value) except (RuntimeError, ValueError) as e: self.interaction.error("配置失败", f"配置 {option.label} 时出错: {e}") # 保存配置 if self.interaction.confirm("保存配置", "是否保存配置到文件？"): self.save_config() return self.config def reset_to_defaults(self) -> None: """重置为默认值""" self.config = self._get_defaults() self.interaction.info("配置重置", "已重置为默认配置") def validate_config(self) -> Dict[str, List[str]]: """验证配置""" errors = {} for key, option in self.options.items(): if option.required and key not in self.config: errors.setdefault(key, []).append("必填项未设置") continue if key in self.config and not option.validate(self.config[key]): errors.setdefault(key, []).append("值无效") return errors def _get_defaults(self) -> Dict[str, Any]: """获取所有默认值""" return {key: option.default for key, option in self.options.items() if option.default is not None} def _merge_with_defaults(self, loaded_config: Dict[str, Any]) -> Dict[str, Any]: """合并加载的配置和默认值""" defaults = self._get_defaults() result = copy.deepcopy(defaults) result.update(loaded_config) return result def _get_options_to_configure(self, show_all: bool) -> Dict[str, ConfigOption]: """获取需要配置的选项""" result = {} for key, option in self.options.items(): # 检查条件 if option.condition and not option.condition(self.config): continue # 检查依赖 if option.depends_on and option.depends_on not in self.config: continue # 检查是否已设置 if show_all or key not in self.config or option.required: result[key] = option return result def _configure_option(self, option: ConfigOption) -> Any: """配置单个选项""" self.interaction.info("配置项", option.label) if option.description: self.interaction.info("说明", option.description) if option.help_text: self.interaction.info("帮助", option.help_text) current_value = self.config.get(option.key, option.default) # 根据类型进行交互式输入 if option.type == "bool": return self._configure_bool_option(option, current_value) elif option.type == "choice": return self._configure_choice_option(option, current_value) elif option.type in ["file", "dir", "path"]: return self._configure_path_option(option, current_value) elif option.type in ["int", "float"]: return self._configure_number_option(option, current_value) elif option.type == "list": return self._configure_list_option(option, current_value) elif option.type == "dict": return self._configure_dict_option(option, current_value) else: return self._configure_string_option(option, current_value) def _configure_bool_option(self, option: ConfigOption, current_value: Any) -> bool: """配置布尔选项""" return self.interaction.confirm( f"{option.label} (布尔值)", f"当前值: {current_value}", default=current_value if isinstance(current_value, bool) else option.default ) def _configure_choice_option(self, option: ConfigOption, current_value: Any) -> Any: """配置选择项""" if not option.choices: raise ValidationError(option.key, "", "选择项必须提供choices选项") choices_display = [{"value": choice, "label": str(choice)} for choice in option.choices] default_index = 0 if current_value in option.choices: default_index = option.choices.index(current_value) return self.interaction.select( f"{option.label} (选择项)", choices_display, default_index ) def _configure_path_option(self, option: ConfigOption, current_value: Any) -> str: """配置路径选项""" path_type = "文件" if option.type == "file" else "目录" if option.type == "dir" else "路径" def path_validator(value): if not value: return False if option.type == "file": return os.path.isfile(value) elif option.type == "dir": return os.path.isdir(value) else: # path return os.path.exists(os.path.dirname(value)) if os.path.dirname(value) else True return self.interaction.input_text( f"{option.label} ({path_type})", str(current_value) if current_value else "", path_validator, f"请输入有效的{path_type}" ) def _configure_number_option(self, option: ConfigOption, current_value: Any) -> Union[int, float]: """配置数字选项""" number_type = "整数" if option.type == "int" else "数字" def number_validator(value): if not value: return not option.required try: return int(value) if option.type == "int" else float(value) except ValueError: return False value_str = self.interaction.input_text( f"{option.label} ({number_type})", str(current_value) if current_value is not None else "", number_validator, f"请输入有效的{number_type}" ) if not value_str: return option.default return int(value_str) if option.type == "int" else float(value_str) def _configure_list_option(self, option: ConfigOption, current_value: Any) -> List[Any]: """配置列表选项""" current_items = current_value if isinstance(current_value, list) else [] items = [] self.interaction.info("列表配置", "请逐项添加，留空结束") while True: item = self.interaction.input_text(f"项目 {len(items)+1}") if not item: break items.append(item) return items if items else option.default def _configure_dict_option(self, option: ConfigOption, current_value: Any) -> Dict[str, Any]: """配置字典选项""" current_dict = current_value if isinstance(current_value, dict) else {} result = {} self.interaction.info("字典配置", "请逐项添加键值对，留空键名结束") while True: key = self.interaction.input_text(f"键名") if not key: break value = self.interaction.input_text(f"值 for '{key}'") result[key] = value return result if result else option.default def _configure_string_option(self, option: ConfigOption, current_value: Any) -> str: """配置字符串选项""" return self.interaction.input_text( option.label, str(current_value) if current_value else "", option.validator, "输入无效" ) def create_default_config_manager(config_file: str = "config/user_config.yaml") -> ConfigManager: """创建默认配置管理器""" manager = ConfigManager(config_file) # 添加默认配置选项 default_options = [ ConfigOption( key="output_dir", label="输出目录", description="生成的文档文件保存目录", type="dir", default="docs", required=True, help_text="所有生成的文档将保存在此目录下" ), ConfigOption( key="template_dir", label="模板目录", description="Jinja2模板文件目录", type="dir", default="config/templates", required=True ), ConfigOption( key="cache_enabled", label="启用缓存", description="是否启用文档缓存功能", type="bool", default=True, help_text="启用缓存可以提高重复操作的性能" ), ConfigOption( key="cache_dir", label="缓存目录", description="缓存文件保存目录", type="dir", default=".cache", depends_on="cache_enabled" ), ConfigOption( key="log_level", label="日志级别", description="系统日志输出级别", type="choice", default="INFO", choices=["DEBUG", "INFO", "WARNING", "ERROR"], help_text="DEBUG显示最详细信息，ERROR只显示错误信息" ), ConfigOption( key="interaction_level", label="交互级别", description="用户交互程度", type="choice", default="normal", choices=["silent", "basic", "normal", "verbose", "debug"], help_text="silent静默模式，debug最详细模式" ), ConfigOption( key="auto_confirm", label="自动确认", description="是否自动确认所有提示", type="bool", default=False, help_text="启用后将跳过所有确认提示" ), ConfigOption( key="include_hidden", label="包含隐藏文件", description="是否处理隐藏文件和目录", type="bool", default=False, help_text="隐藏文件以.开头" ), ConfigOption( key="max_depth", label="最大递归深度", description="目录递归分析的最大深度", type="int", default=10, validator=lambda x: x > 0 and x <= 50, help_text="设置过大可能影响性能" ), ConfigOption( key="file_extensions", label="文件扩展名过滤", description="只处理特定扩展名的文件，空则处理所有文件", type="list", default=[], help_text="如: ['.py', '.js', '.md']" ), ConfigOption( key="exclude_patterns", label="排除模式", description="要排除的文件或目录模式", type="list", default=["node_modules", ".git", "__pycache__"], help_text="支持glob模式" ) ] manager.add_options(default_options) return manager