import os import json import hashlib from typing import Any, Optional from collections import OrderedDict class LRUCache: """LRU缓存实现""" def __init__(self, capacity: int = 100): self.capacity = capacity self.cache = OrderedDict() def get(self, key: str) -> Optional[Any]: """获取缓存值""" if key in self.cache: # 移动到末尾表示最近使用 self.cache.move_to_end(key) return self.cache[key] return None def put(self, key: str, value: Any): """设置缓存值""" if key in self.cache: # 更新现有值 self.cache.move_to_end(key) elif len(self.cache) >= self.capacity: # 删除最久未使用的项 self.cache.popitem(last=False) self.cache[key] = value def delete(self, key: str): """删除缓存项""" if key in self.cache: del self.cache[key] class CacheManager: """多级缓存管理器""" def __init__(self, memory_capacity: int = 100, disk_cache_dir: str = "cache"): self.memory_cache = LRUCache(memory_capacity) self.disk_cache_dir = disk_cache_dir # 创建磁盘缓存目录 if not os.path.exists(disk_cache_dir): os.makedirs(disk_cache_dir) def _get_disk_cache_path(self, key: str) -> str: """获取磁盘缓存文件路径""" # 使用哈希值作为文件名避免特殊字符问题 key_hash = hashlib.md5(key.encode()).hexdigest() return os.path.join(self.disk_cache_dir, f"{key_hash}.cache") def get(self, key: str) -> Optional[Any]: """获取缓存值""" # 首先从内存缓存获取 value = self.memory_cache.get(key) if value is not None: return value # 然后从磁盘缓存获取 disk_path = self._get_disk_cache_path(key) if os.path.exists(disk_path): try: with open(disk_path, 'r', encoding='utf-8') as f: data = json.load(f) # 放入内存缓存 self.memory_cache.put(key, data) return data except (json.JSONDecodeError, IOError): # 磁盘缓存损坏,删除文件 try: os.remove(disk_path) except OSError: pass return None def put(self, key: str, value: Any): """设置缓存值""" # 放入内存缓存 self.memory_cache.put(key, value) # 同时写入磁盘缓存 disk_path = self._get_disk_cache_path(key) try: with open(disk_path, 'w', encoding='utf-8') as f: json.dump(value, f, ensure_ascii=False, indent=2) except IOError: # 磁盘写入失败,不影响内存缓存 pass def delete(self, key: str): """删除缓存项""" # 从内存缓存删除 self.memory_cache.delete(key) # 从磁盘缓存删除 disk_path = self._get_disk_cache_path(key) if os.path.exists(disk_path): try: os.remove(disk_path) except OSError: pass def clear(self): """清空所有缓存""" # 清空内存缓存 self.memory_cache.cache.clear() # 清空磁盘缓存 for filename in os.listdir(self.disk_cache_dir): file_path = os.path.join(self.disk_cache_dir, filename) try: if os.path.isfile(file_path): os.remove(file_path) except OSError: pass # 创建全局缓存管理器实例 cache_manager = CacheManager()