plugins/cache/core.py

"""
文件缓存核心逻辑 - 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}"}