Zhiwen Assistant MCP

""" 兼容性模块 提供多平台支持、API标准化和版本管理功能 """ import os import sys import platform import subprocess from typing import Dict, Any, Optional, List, Union, Callable from dataclasses import dataclass, asdict from enum import Enum import json import re from pathlib import Path class Platform(Enum): """支持的平台""" WINDOWS = "windows" LINUX = "linux" MACOS = "macos" UNIX = "unix" class Architecture(Enum): """支持的架构""" X86 = "x86" X64 = "x64" ARM64 = "arm64" ARM = "arm" UNKNOWN = "unknown" class APIVersion(Enum): """API版本""" V1 = "v1" V2 = "v2" LATEST = "latest" @dataclass class SystemInfo: """系统信息""" platform: Platform architecture: Architecture python_version: str os_version: str is_container: bool = False is_virtual: bool = False @classmethod def detect(cls) -> 'SystemInfo': """检测当前系统信息""" # 检测平台 system = platform.system().lower() if system == "windows": platform_enum = Platform.WINDOWS elif system == "darwin": platform_enum = Platform.MACOS elif system == "linux": platform_enum = Platform.LINUX else: platform_enum = Platform.UNIX # 检测架构 machine = platform.machine().lower() if machine in ("x86_64", "amd64"): arch_enum = Architecture.X64 elif machine in ("i386", "i686", "x86"): arch_enum = Architecture.X86 elif machine in ("arm64", "aarch64"): arch_enum = Architecture.ARM64 elif machine.startswith("arm"): arch_enum = Architecture.ARM else: arch_enum = Architecture.UNKNOWN # 检测容器环境 is_container = cls._is_container_environment() # 检测虚拟环境 is_virtual = cls._is_virtual_environment() return cls( platform=platform_enum, architecture=arch_enum, python_version=platform.python_version(), os_version=platform.release(), is_container=is_container, is_virtual=is_virtual ) @staticmethod def _is_container_environment() -> bool: """检测是否在容器中运行""" # 检测Docker if os.path.exists("/.dockerenv"): return True # 检测cgroup try: with open("/proc/1/cgroup", "r") as f: content = f.read() if "docker" in content or "containerd" in content: return True except (FileNotFoundError, PermissionError): pass return False @staticmethod def _is_virtual_environment() -> bool: """检测是否在虚拟环境中运行""" return ( hasattr(sys, 'real_prefix') or hasattr(sys, 'base_prefix') or sys.prefix != sys.base_prefix ) @dataclass class APIEndpoint: """API端点定义""" path: str method: str = "GET" version: APIVersion = APIVersion.V1 description: str = "" parameters: Dict[str, Any] = None responses: Dict[str, Any] = None deprecated: bool = False deprecation_message: str = "" def __post_init__(self): if self.parameters is None: self.parameters = {} if self.responses is None: self.responses = {} class CompatibilityManager: """兼容性管理器""" def __init__(self): self.system_info = SystemInfo.detect() self.endpoints: Dict[str, APIEndpoint] = {} self.platform_handlers: Dict[Platform, Dict[str, Callable]] = {} self.version_handlers: Dict[APIVersion, Dict[str, Callable]] = {} # 注册默认平台处理器 self._register_default_platform_handlers() # 注册默认版本处理器 self._register_default_version_handlers() def get_system_info(self) -> SystemInfo: """获取系统信息""" return self.system_info def is_platform_supported(self, platform: Platform) -> bool: """检查平台是否支持""" return platform in self.platform_handlers def register_platform_handler(self, platform: Platform, handler_name: str, handler: Callable) -> None: """注册平台处理器""" if platform not in self.platform_handlers: self.platform_handlers[platform] = {} self.platform_handlers[platform][handler_name] = handler def register_version_handler(self, version: APIVersion, handler_name: str, handler: Callable) -> None: """注册版本处理器""" if version not in self.version_handlers: self.version_handlers[version] = {} self.version_handlers[version][handler_name] = handler def register_endpoint(self, endpoint_id: str, endpoint: APIEndpoint) -> None: """注册API端点""" self.endpoints[endpoint_id] = endpoint def get_endpoint(self, endpoint_id: str, version: Optional[APIVersion] = None) -> Optional[APIEndpoint]: """获取API端点""" endpoint = self.endpoints.get(endpoint_id) if not endpoint: return None # 如果指定了版本，检查是否匹配 if version and endpoint.version != version and endpoint.version != APIVersion.LATEST: return None return endpoint def list_endpoints(self, version: Optional[APIVersion] = None) -> Dict[str, APIEndpoint]: """列出API端点""" if version: return { endpoint_id: endpoint for endpoint_id, endpoint in self.endpoints.items() if endpoint.version == version or endpoint.version == APIVersion.LATEST } return self.endpoints.copy() def get_platform_handler(self, handler_name: str) -> Optional[Callable]: """获取当前平台的处理器""" handlers = self.platform_handlers.get(self.system_info.platform, {}) return handlers.get(handler_name) def get_version_handler(self, version: APIVersion, handler_name: str) -> Optional[Callable]: """获取版本处理器""" handlers = self.version_handlers.get(version, {}) return handlers.get(handler_name) def execute_platform_handler(self, handler_name: str, *args, **kwargs): """执行当前平台的处理器""" handler = self.get_platform_handler(handler_name) if not handler: raise ValueError(f"未找到平台处理器: {handler_name}") return handler(*args, **kwargs) def normalize_path(self, path: str) -> str: """标准化路径（跨平台）""" if self.system_info.platform == Platform.WINDOWS: # Windows路径处理 path = path.replace('/', '\\') # 确保驱动器字母大写 if len(path) >= 2 and path[1] == ':': path = path[0].upper() + path[1:] else: # Unix路径处理 path = path.replace('\\', '/') # 规范化路径 return os.path.normpath(path) def get_executable_extension(self) -> str: """获取可执行文件扩展名""" return ".exe" if self.system_info.platform == Platform.WINDOWS else "" def get_path_separator(self) -> str: """获取路径分隔符""" return ";" if self.system_info.platform == Platform.WINDOWS else ":" def run_command(self, command: str, shell: bool = True, capture_output: bool = True) -> subprocess.CompletedProcess: """运行命令（跨平台）""" # 根据平台调整命令 if self.system_info.platform == Platform.WINDOWS and not shell: # Windows下非shell模式需要调整 command = f'cmd /c "{command}"' return subprocess.run( command, shell=shell, capture_output=capture_output, text=True ) def check_command_exists(self, command: str) -> bool: """检查命令是否存在""" try: result = self.run_command(f"which {command}" if self.system_info.platform != Platform.WINDOWS else f"where {command}") return result.returncode == 0 except (IndexError, KeyError, RuntimeError) as e: return False def get_environment_variable(self, name: str, default: str = "") -> str: """获取环境变量（跨平台）""" return os.environ.get(name, default) def set_environment_variable(self, name: str, value: str) -> None: """设置环境变量""" os.environ[name] = value def get_temp_directory(self) -> str: """获取临时目录""" if self.system_info.platform == Platform.WINDOWS: return os.environ.get("TEMP", "C:\\temp") else: return os.environ.get("TMPDIR", "/tmp") def get_user_home_directory(self) -> str: """获取用户主目录""" return str(Path.home()) def get_config_directory(self) -> str: """获取配置目录""" if self.system_info.platform == Platform.WINDOWS: return os.path.join(os.environ.get("APPDATA", ""), "FolderDocsMCP") elif self.system_info.platform == Platform.MACOS: return os.path.join(os.environ.get("HOME", ""), "Library", "Application Support", "FolderDocsMCP") else: return os.path.join(os.environ.get("HOME", ""), ".config", "folder-docs-mcp") def create_api_documentation(self) -> Dict[str, Any]: """生成API文档""" doc = { "info": { "title": "优化版文件夹文档生成MCP服务器API", "version": "2.0.0", "description": "提供文档生成和分析功能的API", "platform": self.system_info.platform.value, "architecture": self.system_info.architecture.value }, "endpoints": {} } for endpoint_id, endpoint in self.endpoints.items(): doc["endpoints"][endpoint_id] = { "path": endpoint.path, "method": endpoint.method, "version": endpoint.version.value, "description": endpoint.description, "parameters": endpoint.parameters, "responses": endpoint.responses, "deprecated": endpoint.deprecated, "deprecation_message": endpoint.deprecation_message } return doc def _register_default_platform_handlers(self): """注册默认平台处理器""" # Windows处理器 self.register_platform_handler(Platform.WINDOWS, "normalize_line_endings", lambda text: text.replace('\n', '\r\n')) self.register_platform_handler(Platform.WINDOWS, "get_shell", lambda: "cmd") # Unix/Linux处理器 for platform in [Platform.LINUX, Platform.MACOS, Platform.UNIX]: self.register_platform_handler(platform, "normalize_line_endings", lambda text: text) self.register_platform_handler(platform, "get_shell", lambda: "/bin/bash") def _register_default_version_handlers(self): """注册默认版本处理器""" # v1处理器 self.register_version_handler(APIVersion.V1, "transform_response", lambda data: data) # v2处理器 self.register_version_handler(APIVersion.V2, "transform_response", lambda data: { "version": "v2", "data": data, "timestamp": self._get_timestamp() }) def _get_timestamp(self) -> str: """获取时间戳""" import datetime return datetime.datetime.now().isoformat() # 全局兼容性管理器实例 global_compatibility = CompatibilityManager() def get_system_info() -> SystemInfo: """获取系统信息便捷函数""" return global_compatibility.get_system_info() def normalize_path(path: str) -> str: """路径标准化便捷函数""" return global_compatibility.normalize_path(path) def run_command(command: str, shell: bool = True, capture_output: bool = True) -> subprocess.CompletedProcess: """运行命令便捷函数""" return global_compatibility.run_command(command, shell, capture_output) def get_config_directory() -> str: """获取配置目录便捷函数""" return global_compatibility.get_config_directory() def register_endpoint(endpoint_id: str, path: str, method: str = "GET", version: APIVersion = APIVersion.V1, description: str = "", **kwargs): """注册API端点便捷函数""" endpoint = APIEndpoint( path=path, method=method, version=version, description=description, **kwargs ) global_compatibility.register_endpoint(endpoint_id, endpoint) # 注册默认API端点 register_endpoint("generate_readme", "/api/readme", "POST", APIVersion.V1, "生成README文档") register_endpoint("generate_mindmap", "/api/mindmap", "POST", APIVersion.V1, "生成思维导图") register_endpoint("analyze_project", "/api/analyze", "POST", APIVersion.V1, "分析项目结构") register_endpoint("list_templates", "/api/templates", "GET", APIVersion.V1, "列出可用模板") register_endpoint("get_config", "/api/config", "GET", APIVersion.V1, "获取配置信息") register_endpoint("set_config", "/api/config", "PUT", APIVersion.V1, "更新配置") register_endpoint("health_check", "/api/health", "GET", APIVersion.V1, "健康检查") register_endpoint("get_version", "/api/version", "GET", APIVersion.V1, "获取版本信息") # v2版本的端点（增强功能） register_endpoint("generate_readme_v2", "/api/v2/readme", "POST", APIVersion.V2, "生成README文档（增强版）") register_endpoint("generate_mindmap_v2", "/api/v2/mindmap", "POST", APIVersion.V2, "生成思维导图（增强版）") register_endpoint("analyze_project_v2", "/api/v2/analyze", "POST", APIVersion.V2, "分析项目结构（增强版）") register_endpoint("batch_generate", "/api/v2/batch", "POST", APIVersion.V2, "批量生成文档")