122 lines
3.8 KiB
Python
122 lines
3.8 KiB
Python
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() |