PubChem Chemical Safety MCP Server

""" 缓存服务模块 - 使用本地文件缓存 """ import json import os import logging from typing import Optional, Dict, Any from pathlib import Path import time logger = logging.getLogger(__name__) class CacheService: """本地文件缓存服务""" def __init__(self, cache_dir: str = ".cache"): self.cache_dir = Path(cache_dir) self.cache_dir.mkdir(exist_ok=True) self.memory_cache: Dict[str, Any] = {} async def initialize(self): """初始化缓存服务""" logger.info("Local file cache initialized") async def close(self): """关闭缓存服务""" pass def _get_cache_file_path(self, key: str) -> Path: """获取缓存文件路径""" # 使用hash避免文件名过长 import hashlib hash_key = hashlib.md5(key.encode()).hexdigest() return self.cache_dir / f"{hash_key}.json" async def get(self, key: str) -> Optional[Any]: """获取缓存值""" try: # 先检查内存缓存 if key in self.memory_cache: return self.memory_cache[key] # 检查文件缓存 cache_file = self._get_cache_file_path(key) if cache_file.exists(): with open(cache_file, 'r', encoding='utf-8') as f: data = json.load(f) # 检查是否过期 if time.time() - data.get('timestamp', 0) < data.get('ttl', 3600): # 更新内存缓存 self.memory_cache[key] = data['value'] return data['value'] else: # 删除过期文件 cache_file.unlink(missing_ok=True) except Exception as e: logger.error(f"Cache get error: {e}") return None async def set(self, key: str, value: Any, ttl: int = 3600): """设置缓存值""" try: # 更新内存缓存 self.memory_cache[key] = value # 写入文件缓存 cache_file = self._get_cache_file_path(key) cache_data = { 'value': value, 'timestamp': time.time(), 'ttl': ttl } with open(cache_file, 'w', encoding='utf-8') as f: json.dump(cache_data, f, ensure_ascii=False, indent=2) except Exception as e: logger.error(f"Cache set error: {e}") async def delete(self, key: str): """删除缓存值""" try: # 删除内存缓存 self.memory_cache.pop(key, None) # 删除文件缓存 cache_file = self._get_cache_file_path(key) cache_file.unlink(missing_ok=True) except Exception as e: logger.error(f"Cache delete error: {e}") def _make_key(self, prefix: str, identifier: str) -> str: """生成缓存键""" return f"{prefix}:{identifier}" async def get_compound_info(self, name: str) -> Optional[Dict[str, Any]]: """获取化合物信息缓存""" key = self._make_key("compound", name.lower()) return await self.get(key) async def set_compound_info(self, name: str, data: Dict[str, Any]): """设置化合物信息缓存""" key = self._make_key("compound", name.lower()) await self.set(key, data, ttl=7200) # 2小时 async def get_safety_info(self, cid: int) -> Optional[Dict[str, Any]]: """获取安全信息缓存""" key = self._make_key("safety", str(cid)) return await self.get(key) async def set_safety_info(self, cid: int, data: Dict[str, Any]): """设置安全信息缓存""" key = self._make_key("safety", str(cid)) await self.set(key, data, ttl=3600) # 1小时 async def get_toxicity_data(self, cid: int) -> Optional[Dict[str, Any]]: """获取毒性数据缓存""" key = self._make_key("toxicity", str(cid)) return await self.get(key) async def set_toxicity_data(self, cid: int, data: Dict[str, Any]): """设置毒性数据缓存""" key = self._make_key("toxicity", str(cid)) await self.set(key, data, ttl=3600) # 1小时 def clear_all_cache(self): """清空所有缓存""" try: # 清空内存缓存 self.memory_cache.clear() # 删除所有缓存文件 for cache_file in self.cache_dir.glob("*.json"): cache_file.unlink(missing_ok=True) logger.info("All cache cleared") except Exception as e: logger.error(f"Error clearing cache: {e}") def get_cache_stats(self) -> Dict[str, Any]: """获取缓存统计信息""" try: cache_files = list(self.cache_dir.glob("*.json")) total_size = sum(f.stat().st_size for f in cache_files) return { "memory_cache_size": len(self.memory_cache), "file_cache_count": len(cache_files), "total_cache_size_bytes": total_size, "cache_directory": str(self.cache_dir) } except Exception as e: logger.error(f"Error getting cache stats: {e}") return {}