Zhiwen Assistant MCP

""" 命令行界面模块 提供用户友好的命令行操作界面 """ import argparse import sys import asyncio from typing import Dict, Any, Optional, List from pathlib import Path from utils import ( ConfigManager, create_default_config_manager, InteractionManager, set_interaction_level, InteractionLevel, info, success, warning, error, select, confirm, handle_exception ) from server import OptimizedFolderDocMCPServer class CLI: """命令行界面""" def __init__(self): self.interaction = InteractionManager() self.config_manager: Optional[ConfigManager] = None def run(self, args: Optional[List[str]] = None): """运行命令行界面""" try: # 解析命令行参数 parser = self._create_parser() parsed_args = parser.parse_args(args) # 设置交互级别 interaction_level = InteractionLevel(parsed_args.interaction_level) set_interaction_level(interaction_level) # 执行命令 if hasattr(parsed_args, 'func'): # 创建配置管理器 self.config_manager = create_default_config_manager(parsed_args.config) # 执行命令 return parsed_args.func(parsed_args) else: # 显示帮助 parser.print_help() except KeyboardInterrupt: print("\n操作已取消") sys.exit(0) except (RuntimeError, ValueError) as e: handle_exception(e, "命令行操作") sys.exit(1) def _create_parser(self) -> argparse.ArgumentParser: """创建命令行参数解析器""" parser = argparse.ArgumentParser( description="优化版文件夹文档生成MCP服务器", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例用法: %(prog)s start # 启动服务器 %(prog)s config # 交互式配置 %(prog)s config --show # 显示当前配置 %(prog)s health-check # 健康检查 %(prog)s generate ./my-project # 生成项目文档 %(prog)s analyze ./my-project # 分析项目结构 """ ) # 全局参数 parser.add_argument( "--config", type=str, default="config/user_config.yaml", help="配置文件路径" ) parser.add_argument( "--interaction-level", type=str, choices=["silent", "basic", "normal", "verbose", "debug"], default="normal", help="交互级别" ) # 创建子命令 subparsers = parser.add_subparsers( title="可用命令", dest="command", help="选择要执行的操作" ) # start 命令 start_parser = subparsers.add_parser( "start", help="启动MCP服务器" ) start_parser.set_defaults(func=self._cmd_start) # config 命令 config_parser = subparsers.add_parser( "config", help="配置管理" ) config_group = config_parser.add_mutually_exclusive_group(required=False) config_group.add_argument( "--show", action="store_true", help="显示当前配置" ) config_group.add_argument( "--reset", action="store_true", help="重置为默认配置" ) config_group.add_argument( "--validate", action="store_true", help="验证配置" ) config_parser.set_defaults(func=self._cmd_config) # health-check 命令 health_parser = subparsers.add_parser( "health-check", help="执行健康检查" ) health_parser.set_defaults(func=self._cmd_health_check) # generate 命令 generate_parser = subparsers.add_parser( "generate", help="生成项目文档" ) generate_parser.add_argument( "directory", type=str, help="要生成文档的目录路径" ) generate_parser.add_argument( "--output", type=str, help="输出目录（覆盖配置中的设置）" ) generate_parser.add_argument( "--template", type=str, help="使用的模板（覆盖配置中的设置）" ) generate_parser.add_argument( "--format", type=str, choices=["readme", "mindmap", "both"], default="both", help="生成格式" ) generate_parser.set_defaults(func=self._cmd_generate) # analyze 命令 analyze_parser = subparsers.add_parser( "analyze", help="分析项目结构" ) analyze_parser.add_argument( "directory", type=str, help="要分析的目录路径" ) analyze_parser.add_argument( "--depth", type=int, default=10, help="分析深度" ) analyze_parser.add_argument( "--export", type=str, help="导出分析结果到文件" ) analyze_parser.set_defaults(func=self._cmd_analyze) return parser def _cmd_start(self, args): """启动服务器命令""" info("启动MCP服务器") # 创建服务器实例 server = OptimizedFolderDocMCPServer(args.config, InteractionLevel(args.interaction_level)) try: # 启动服务器 asyncio.run(server.start()) except KeyboardInterrupt: info("收到停止信号") except (RuntimeError, ValueError) as e: error("服务器启动失败", str(e)) sys.exit(1) def _cmd_config(self, args): """配置管理命令""" if args.show: self._show_config() elif args.reset: self._reset_config() elif args.validate: self._validate_config() else: # 默认行为：交互式配置 self._interactive_config() def _show_config(self): """显示当前配置""" if not self.config_manager.config_file.exists(): warning("配置文件不存在", "将显示默认配置") self.config_manager.load_config() info("当前配置:") for key, value in self.config_manager.config.items(): print(f" {key}: {value}") def _reset_config(self): """重置配置""" if confirm("重置配置", "确定要重置为默认配置吗？这将丢失所有自定义设置"): self.config_manager.reset_to_defaults() success("配置已重置", "所有配置已恢复为默认值") def _validate_config(self): """验证配置""" self.config_manager.load_config() errors = self.config_manager.validate_config() if not errors: success("配置验证通过", "所有配置项都有效") else: error("配置验证失败") for key, error_list in errors.items(): for error in error_list: print(f" {key}: {error}") def _interactive_config(self): """交互式配置""" self.config_manager.interactive_config() def _cmd_health_check(self, args): """健康检查命令""" async def run_health_check(): # 创建服务器实例 server = OptimizedFolderDocMCPServer(args.config, InteractionLevel(args.interaction_level)) # 执行健康检查 health = await server.health_check() print("健康检查结果:") print(f" 状态: {health['status']}") if health['status'] == 'unhealthy': print(f" 错误: {health.get('error', '未知错误')}") return False print(" 组件状态:") for component, status in health.get('components', {}).items(): comp_status = status.get('status', '未知') print(f" {component}: {comp_status}") if comp_status == 'healthy' and 'stats' in status: print(f" 统计: {status['stats']}") return True try: result = asyncio.run(run_health_check()) sys.exit(0 if result else 1) except (RuntimeError, ValueError) as e: error("健康检查失败", str(e)) sys.exit(1) def _cmd_generate(self, args): """生成文档命令""" directory = Path(args.directory) if not directory.exists(): error("目录不存在", f"目录 '{directory}' 不存在") sys.exit(1) if not directory.is_dir(): error("不是目录", f"'{directory}' 不是目录") sys.exit(1) info("生成文档", f"正在为 '{directory}' 生成文档") # 这里应该调用文档生成功能 # 由于我们还没有实现完整的CLI集成，暂时显示计划 print("文档生成功能正在开发中...") print(f"目标目录: {directory}") print(f"输出目录: {args.output or '使用配置中的设置'}") print(f"使用模板: {args.template or '使用配置中的设置'}") print(f"生成格式: {args.format}") def _cmd_analyze(self, args): """分析项目命令""" directory = Path(args.directory) if not directory.exists(): error("目录不存在", f"目录 '{directory}' 不存在") sys.exit(1) if not directory.is_dir(): error("不是目录", f"'{directory}' 不是目录") sys.exit(1) info("分析项目", f"正在分析 '{directory}'") # 这里应该调用项目分析功能 # 由于我们还没有实现完整的CLI集成，暂时显示计划 print("项目分析功能正在开发中...") print(f"目标目录: {directory}") print(f"分析深度: {args.depth}") if args.export: print(f"导出文件: {args.export}") def main(): """主函数""" cli = CLI() cli.run() if __name__ == "__main__": main()