Zhiwen Assistant MCP

""" 文档服务模块 提供README生成、文档更新等文档相关服务。 """ import os from pathlib import Path from typing import Dict, Any, Optional, List from datetime import datetime from data_access.file_system import FileSystemInterface, FolderStructure from data_access.cache import CacheInterface from data_access.config import ConfigInterface import logging from middleware.logging import LoggingMiddleware from middleware.logging import log_performance logger = logging.getLogger(__name__) class DocumentService: """文档服务类""" def __init__( self, file_system: FileSystemInterface, template_service: Any, # TemplateService cache_service: Any, # CacheService config_interface: Optional[ConfigInterface] = None ): """ 初始化文档服务 Args: file_system: 文件系统接口 template_service: 模板服务 cache_service: 缓存服务 config_interface: 配置接口 """ self.file_system = file_system self.template_service = template_service self.cache_service = cache_service self.config_interface = config_interface or ConfigInterface() self.config = self.config_interface.get_config() @log_performance("generate_readme") async def generate_readme( self, folder_path: str, template_name: str = "simple", force_update: bool = False, include_hidden: bool = False ) -> Dict[str, Any]: """ 生成README文档 Args: folder_path: 文件夹路径 template_name: 模板名称 force_update: 是否强制更新 include_hidden: 是否包含隐藏文件 Returns: 生成结果 """ try: folder_path = os.path.abspath(folder_path) readme_path = Path(folder_path) / "README.md" # 检查是否需要更新 if not force_update and readme_path.exists(): # 检查文件是否有更新 should_update = await self.file_system.should_update_file( str(readme_path), [folder_path] ) if not should_update: return { "success": True, "message": "README已存在且无需更新", "path": str(readme_path), "updated": False } # 获取文件夹结构 structure = await self.file_system.get_folder_structure( folder_path, exclude_dirs=[], include_hidden=include_hidden ) # 生成README内容 content = await self._generate_readme_content( structure, template_name ) # 写入文件 success, error_msg = await self.file_system.write_file_content( str(readme_path), content, backup=True ) if not success: return { "success": False, "error": f"写入README失败: {error_msg}" } # 缓存生成结果 cache_key = self.cache_service.cache_file_content(str(readme_path)) self.cache_service.set(cache_key, content, ttl=3600) # 1小时 logging_middleware.logger.info( "README生成成功", path=str(readme_path), template=template_name, size=len(content) ) return { "success": True, "message": "README生成成功", "path": str(readme_path), "size": len(content), "template": template_name, "updated": True } except (RuntimeError, ValueError) as e: return { "success": False, "error": f"生成README失败: {str(e)}" } @log_performance("preview_readme") async def preview_readme( self, folder_path: str, template_name: str = "simple", include_hidden: bool = False ) -> Dict[str, Any]: """ 预览README内容 Args: folder_path: 文件夹路径 template_name: 模板名称 include_hidden: 是否包含隐藏文件 Returns: 预览结果 """ try: folder_path = os.path.abspath(folder_path) # 获取文件夹结构 structure = await self.file_system.get_folder_structure( folder_path, exclude_dirs=[], include_hidden=include_hidden ) # 生成README内容 content = await self._generate_readme_content( structure, template_name ) return { "success": True, "content": content, "template": template_name, "folder_path": folder_path, "size": len(content) } except (RuntimeError, ValueError) as e: return { "success": False, "error": f"预览README失败: {str(e)}" } async def _generate_readme_content( self, structure: FolderStructure, template_name: str ) -> str: """ 生成README内容 Args: structure: 文件夹结构 template_name: 模板名称 Returns: README内容 """ # 获取模板 if hasattr(self.template_service, 'render_template'): # 使用模板服务 context = self._build_template_context(structure) content = self.template_service.render_template(template_name, context) else: # 使用默认模板 content = self._generate_default_readme(structure) return content def _build_template_context(self, structure: FolderStructure) -> Dict[str, Any]: """ 构建模板上下文 Args: structure: 文件夹结构 Returns: 模板上下文 """ # 获取文件夹描述 folder_descriptions = self.config.folder_descriptions # 处理文件夹信息 folders = [] for folder_info in structure.folders: folder_name = folder_info.name description = folder_descriptions.get( folder_name, f"{folder_name}目录" ) folders.append({ "name": folder_name, "description": description, "size": folder_info.size, "modified_time": folder_info.modified_time }) # 处理文件信息 files = [] for file_info in structure.files: file_name = file_info.name description = self._get_file_description(file_info) files.append({ "name": file_name, "description": description, "size": file_info.size, "extension": file_info.extension, "type": file_info.mime_type }) return { "folder_name": structure.root_path.name, "description": folder_descriptions.get( structure.root_path.name, f"{structure.root_path.name}项目目录" ), "folders": folders, "files": files, "folder_count": structure.folder_count, "file_count": structure.file_count, "total_size": self._format_size(structure.total_size), "generation_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } def _generate_default_readme(self, structure: FolderStructure) -> str: """ 生成默认README内容 Args: structure: 文件夹结构 Returns: README内容 """ context = self._build_template_context(structure) # 简单的默认模板 content = f"""# {context['folder_name']} {context['description']} ## 📁 文件夹结构 """ if context['folders']: content += "### 📂 子文件夹\n\n" for folder in context['folders']: content += f"- `{folder['name']}/` - {folder['description']}\n" content += "\n" if context['files']: content += "### 📄 文件列表\n\n" for file in context['files']: content += f"- `{file['name']}` - {file['description']}\n" content += "\n" content += f"""## 📊 统计信息 - **文件夹数量**: {context['folder_count']} - **文件数量**: {context['file_count']} - **总大小**: {context['total_size']} - **生成时间**: {context['generation_time']} --- *此文档由 MCP服务器 自动生成于 {context['generation_time']}* """ return content def _get_file_description(self, file_info: Any) -> str: """ 获取文件描述 Args: file_info: 文件信息 Returns: 文件描述 """ # 根据文件扩展名生成描述 ext = file_info.extension.lower() descriptions = { '.py': 'Python源代码文件', '.js': 'JavaScript源代码文件', '.ts': 'TypeScript源代码文件', '.html': 'HTML网页文件', '.css': 'CSS样式文件', '.json': 'JSON数据文件', '.yaml': 'YAML配置文件', '.yml': 'YAML配置文件', '.md': 'Markdown文档文件', '.txt': '文本文件', '.sql': 'SQL数据库文件', '.sh': 'Shell脚本文件', '.bat': '批处理文件', '.png': 'PNG图片文件', '.jpg': 'JPEG图片文件', '.jpeg': 'JPEG图片文件', '.gif': 'GIF图片文件', '.svg': 'SVG矢量图文件' } return descriptions.get(ext, f'{ext}文件') def _format_size(self, size_bytes: int) -> str: """ 格式化文件大小 Args: size_bytes: 字节数 Returns: 格式化的大小字符串 """ if size_bytes == 0: return "0 B" size_names = ["B", "KB", "MB", "GB", "TB"] i = 0 size = float(size_bytes) while size >= 1024.0 and i < len(size_names) - 1: size /= 1024.0 i += 1 return f"{size:.1f} {size_names[i]}"