File size: 5,502 Bytes
bdcdaf4 d0c18f0 bdcdaf4 d0c18f0 bdcdaf4 d0c18f0 bdcdaf4 d0c18f0 bdcdaf4 d0c18f0 bdcdaf4 d0c18f0 bdcdaf4 d0c18f0 bdcdaf4 d0c18f0 bdcdaf4 d0c18f0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 | """
文件缓存核心逻辑 - LRU缓存 + 后台清理
"""
import time
import asyncio
from collections import OrderedDict
from typing import Optional, Tuple
from app.utils.url_input_handler import resolve_content, encode_text_to_base64, decode_base64_to_text, encode_bytes_to_base64
MAX_CACHE_SIZE = 100
CLEANUP_INTERVAL = 60
class LRUCache:
"""LRU缓存，自动淘汰最久未使用的条目"""
def __init__(self, max_size: int = MAX_CACHE_SIZE):
self._store: OrderedDict[str, dict] = OrderedDict()
self._max_size = max_size
self._cleanup_task: Optional[asyncio.Task] = None
self._lock = asyncio.Lock() # 线程安全锁
async def start_cleanup(self):
"""启动后台清理任务"""
if self._cleanup_task is None:
self._cleanup_task = asyncio.create_task(self._cleanup_loop())
async def stop_cleanup(self):
"""停止后台清理任务"""
if self._cleanup_task is not None:
self._cleanup_task.cancel()
try:
await self._cleanup_task
except asyncio.CancelledError:
pass
self._cleanup_task = None
async def _cleanup_loop(self):
"""定期清理过期条目"""
while True:
await asyncio.sleep(CLEANUP_INTERVAL)
await self._cleanup_expired()
async def _cleanup_expired(self):
"""清理所有过期条目"""
async with self._lock:
now = time.time()
expired = [k for k, v in self._store.items() if v["expires_at"] < now]
for k in expired:
del self._store[k]
async def set(self, key: str, value: str, ttl_seconds: int) -> dict:
"""设置缓存，自动淘汰 LRU 条目"""
async with self._lock:
if key in self._store:
del self._store[key]
while len(self._store) >= self._max_size:
self._store.popitem(last=False)
self._store[key] = {
"value": value,
"expires_at": time.time() + ttl_seconds,
}
self._store.move_to_end(key)
return {"success": True, "message": f"缓存已设置，TTL {ttl_seconds}秒"}
async def get(self, key: str) -> Optional[dict]:
"""获取缓存，更新 LRU位置"""
async with self._lock:
entry = self._store.get(key)
if not entry:
return None
if time.time() > entry["expires_at"]:
del self._store[key]
return None
self._store.move_to_end(key)
return entry
async def delete(self, key: str) -> bool:
"""删除缓存"""
async with self._lock:
if key in self._store:
del self._store[key]
return True
return False
async def stats(self) -> dict:
"""获取缓存统计"""
async with self._lock:
return {
"size": len(self._store),
"max_size": self._max_size,
}
# 全局缓存实例
_cache: Optional[LRUCache] = None
def get_cache() -> LRUCache:
"""获取全局缓存实例"""
global _cache
if _cache is None:
_cache = LRUCache()
return _cache
async def init_cache():
"""初始化缓存并启动清理任务"""
cache = get_cache()
await cache.start_cleanup()
async def shutdown_cache():
"""关闭缓存清理任务"""
global _cache
if _cache is not None:
await _cache.stop_cleanup()
async def do_set(key: str, value_base64: str, ttl_seconds: int = 3600) -> dict:
"""设置缓存值"""
cache = get_cache()
return await cache.set(key, value_base64, ttl_seconds)
async def do_get(key: str) -> dict:
"""获取缓存值"""
cache = get_cache()
entry = await cache.get(key)
if not entry:
return {"success": False, "message": f"缓存键 {key} 不存在或已过期"}
return {"success": True, "value": entry["value"], "message": "缓存获取成功"}
async def do_delete(key: str) -> dict:
"""删除缓存值"""
cache = get_cache()
if await cache.delete(key):
return {"success": True, "message": f"缓存键 {key} 已删除"}
return {"success": False, "message": f"缓存键 {key} 不存在"}
async def do_set_text(key: str, text_content: str, ttl_seconds: int = 3600) -> dict:
"""设置文本缓存 - 自动编码为 base64"""
value_base64 = encode_text_to_base64(text_content)
return await do_set(key, value_base64, ttl_seconds)
async def do_get_text(key: str) -> dict:
"""获取文本缓存 - 自动从 base64 解码"""
result = await do_get(key)
if result["success"]:
try:
text = decode_base64_to_text(result["value"])
return {"success": True, "text": text, "message": "获取成功"}
except Exception as e:
return {"success": False, "message": f"解码失败: {e}"}
return result
async def do_set_from_url(key: str, url: str, ttl_seconds: int = 3600) -> dict:
"""从 URL 下载文件并缓存"""
try:
content, source = await resolve_content(None, url)
value_base64 = encode_bytes_to_base64(content)
result = await do_set(key, value_base64, ttl_seconds)
result["source"] = source
return result
except ValueError as e:
return {"success": False, "message": f"URL获取失败: {e}"} |