Zhiwen Assistant MCP

""" 缓存管理服务 提供内存缓存和磁盘缓存的统一管理。 """ import json import time import hashlib from typing import Any, Optional, Dict, List from datetime import datetime, timedelta from data_access import CacheInterface class CacheService: """缓存管理服务""" def __init__(self, cache_interface: CacheInterface): """ 初始化缓存服务 Args: cache_interface: 缓存接口实例 """ self.cache = cache_interface self.namespace = "folder_docs" async def get(self, key: str) -> Optional[Any]: """ 获取缓存值 Args: key: 缓存键 Returns: 缓存值，不存在则返回None """ try: full_key = f"{self.namespace}:{key}" return await self.cache.get(full_key) except (CacheError, IndexError, KeyError) as e: # 缓存失败不应该影响主要功能 return None async def set(self, key: str, value: Any, expire_seconds: Optional[int] = None) -> bool: """ 设置缓存值 Args: key: 缓存键 value: 缓存值 expire_seconds: 过期时间（秒） Returns: 是否设置成功 """ try: full_key = f"{self.namespace}:{key}" return await self.cache.set(full_key, value, expire_seconds) except (CacheError, RuntimeError, TypeError) as e: # 缓存失败不应该影响主要功能 return False async def delete(self, key: str) -> bool: """ 删除缓存值 Args: key: 缓存键 Returns: 是否删除成功 """ try: full_key = f"{self.namespace}:{key}" return await self.cache.delete(full_key) except (CacheError, RuntimeError, ValueError) as e: return False async def clear(self) -> bool: """ 清除所有缓存 Returns: 是否清除成功 """ try: # 只清除当前命名空间的缓存 return await self.cache.clear_namespace(self.namespace) except (CacheError, IndexError, KeyError) as e: return False async def get_stats(self) -> Dict[str, Any]: """ 获取缓存统计信息 Returns: 缓存统计信息 """ try: stats = await self.cache.get_stats() return stats except (CacheError, IndexError, KeyError) as e: return { 'memory_hits': 0, 'memory_misses': 0, 'memory_size': 0, 'disk_hits': 0, 'disk_misses': 0, 'disk_size': 0, 'total_requests': 0, 'hit_rate': 0.0 } async def cache_folder_structure(self, folder_path: str, structure: Dict[str, Any]) -> bool: """ 缓存文件夹结构 Args: folder_path: 文件夹路径 structure: 文件夹结构数据 Returns: 是否缓存成功 """ key = f"folder_structure:{folder_path}" # 文件夹结构缓存时间较短，因为可能经常变化 return await self.set(key, structure, expire_seconds=300) # 5分钟 async def get_cached_folder_structure(self, folder_path: str) -> Optional[Dict[str, Any]]: """ 获取缓存的文件夹结构 Args: folder_path: 文件夹路径 Returns: 文件夹结构数据 """ key = f"folder_structure:{folder_path}" return await self.get(key) async def cache_analysis_result(self, folder_path: str, analysis_type: str, result: Dict[str, Any]) -> bool: """ 缓存分析结果 Args: folder_path: 文件夹路径 analysis_type: 分析类型 result: 分析结果 Returns: 是否缓存成功 """ key = f"analysis:{analysis_type}:{folder_path}" # 分析结果缓存时间较长 return await self.set(key, result, expire_seconds=1800) # 30分钟 async def get_cached_analysis_result(self, folder_path: str, analysis_type: str) -> Optional[Dict[str, Any]]: """ 获取缓存的分析结果 Args: folder_path: 文件夹路径 analysis_type: 分析类型 Returns: 分析结果 """ key = f"analysis:{analysis_type}:{folder_path}" return await self.get(key) async def cache_document_content(self, file_path: str, content: str) -> bool: """ 缓存文档内容 Args: file_path: 文件路径 content: 文档内容 Returns: 是否缓存成功 """ key = f"document:{file_path}" # 文档内容缓存时间中等 return await self.set(key, content, expire_seconds=600) # 10分钟 async def get_cached_document_content(self, file_path: str) -> Optional[str]: """ 获取缓存的文档内容 Args: file_path: 文件路径 Returns: 文档内容 """ key = f"document:{file_path}" return await self.get(key) async def cache_template_render(self, template_name: str, context_hash: str, rendered_content: str) -> bool: """ 缓存模板渲染结果 Args: template_name: 模板名称 context_hash: 上下文哈希值 rendered_content: 渲染结果 Returns: 是否缓存成功 """ key = f"template:{template_name}:{context_hash}" # 模板渲染结果可以缓存较长时间 return await self.set(key, rendered_content, expire_seconds=1800) # 30分钟 async def get_cached_template_render(self, template_name: str, context_hash: str) -> Optional[str]: """ 获取缓存的模板渲染结果 Args: template_name: 模板名称 context_hash: 上下文哈希值 Returns: 渲染结果 """ key = f"template:{template_name}:{context_hash}" return await self.get(key) def generate_context_hash(self, context: Dict[str, Any]) -> str: """ 生成上下文哈希值 Args: context: 上下文字典 Returns: 哈希值 """ try: # 将上下文转换为JSON字符串，确保字典顺序一致 context_str = json.dumps(context, sort_keys=True, ensure_ascii=False) return hashlib.md5(context_str.encode('utf-8')).hexdigest() except (CacheError, RuntimeError, TypeError) as e: # 如果序列化失败，使用字符串哈希 return hashlib.md5(str(context).encode('utf-8')).hexdigest() async def invalidate_folder_cache(self, folder_path: str) -> int: """ 使指定文件夹的所有缓存失效 Args: folder_path: 文件夹路径 Returns: 失效的缓存项数量 """ try: # 这里需要遍历所有缓存键，找到与该文件夹相关的项 # 由于CacheInterface的限制，这里实现一个简化版本 keys_to_delete = [] # 检查文件夹结构缓存 structure_key = f"folder_structure:{folder_path}" if await self.get(structure_key) is not None: keys_to_delete.append(structure_key) # 检查分析结果缓存 analysis_types = ['complexity', 'statistics', 'dependencies'] for analysis_type in analysis_types: analysis_key = f"analysis:{analysis_type}:{folder_path}" if await self.get(analysis_key) is not None: keys_to_delete.append(analysis_key) # 删除相关缓存 deleted_count = 0 for key in keys_to_delete: if await self.delete(key): deleted_count += 1 return deleted_count except (CacheError, RuntimeError, TypeError) as e: return 0 async def invalidate_expired_cache(self) -> int: """ 清理过期缓存 Returns: 清理的缓存项数量 """ try: # 这个功能依赖于底层CacheInterface的实现 # 这里提供一个接口，实际实现可能需要扩展CacheInterface if hasattr(self.cache, 'clean_expired'): return await self.cache.clean_expired(self.namespace) return 0 except (CacheError, IndexError, KeyError) as e: return 0 async def get_cache_info(self) -> Dict[str, Any]: """ 获取缓存详细信息 Returns: 缓存信息 """ try: stats = await self.get_stats() # 添加服务特定信息 info = { **stats, 'namespace': self.namespace, 'cache_types': { 'folder_structure': 'folder_structure:*', 'analysis': 'analysis:*:*', 'document': 'document:*', 'template': 'template:*:*' }, 'strategies': { 'folder_structure': {'ttl': 300, 'description': '5分钟过期'}, 'analysis': {'ttl': 1800, 'description': '30分钟过期'}, 'document': {'ttl': 600, 'description': '10分钟过期'}, 'template': {'ttl': 1800, 'description': '30分钟过期'} }, 'last_cleanup': datetime.now().isoformat() } return info except (CacheError, IOError) as e: return { 'error': str(e), 'namespace': self.namespace, 'last_cleanup': datetime.now().isoformat() } async def warm_up_cache(self, folder_path: str) -> Dict[str, bool]: """ 预热指定文件夹的缓存 Args: folder_path: 文件夹路径 Returns: 预热结果 """ try: results = {} # 这里需要调用相应的服务来预生成缓存 # 由于服务间的循环依赖，这个功能可能需要在更高层实现 # 示例：预热文件夹结构缓存 # 实际实现需要传入相应的服务实例 results['folder_structure'] = False results['analysis'] = False return results except (CacheError, RuntimeError, ValueError) as e: return { 'folder_structure': False, 'analysis': False, 'error': '预热失败' }