MoviePilot/app/core/cache.py

import contextvars
import inspect
import shutil
import tempfile
import threading
from abc import ABC, abstractmethod
from contextlib import contextmanager, asynccontextmanager
from functools import wraps
from pathlib import Path
from typing import Any, Dict, Optional, Generator, AsyncGenerator, Tuple, Literal, Union

import aiofiles
import aioshutil
from anyio import Path as AsyncPath
from cachetools import LRUCache as MemoryLRUCache
from cachetools import TTLCache as MemoryTTLCache
from cachetools.keys import hashkey

from app.core.config import settings
from app.helper.redis import RedisHelper, AsyncRedisHelper
from app.log import logger

# 默认缓存区
DEFAULT_CACHE_REGION = "DEFAULT"
# 默认缓存大小
DEFAULT_CACHE_SIZE = 1024
# 默认缓存有效期
DEFAULT_CACHE_TTL = 365 * 24 * 60 * 60

# 上下文变量来控制缓存行为
_fresh = contextvars.ContextVar('fresh', default=False)


class CacheBackend(ABC):
    """
    缓存后端基类，定义通用的缓存接口
    """

    def __getitem__(self, key: str) -> Any:
        """
        获取缓存项，类似 dict[key]
        """
        value = self.get(key)
        if value is None:
            raise KeyError(key)
        return value

    def __setitem__(self, key: str, value: Any) -> None:
        """
        设置缓存项，类似 dict[key] = value
        """
        self.set(key, value)

    def __delitem__(self, key: str) -> None:
        """
        删除缓存项，类似 del dict[key]
        """
        if not self.exists(key):
            raise KeyError(key)
        self.delete(key)

    def __contains__(self, key: str) -> bool:
        """
        检查键是否存在，类似 key in dict
        """
        return self.exists(key)

    def __iter__(self):
        """
        返回缓存的迭代器，类似 iter(dict)
        """
        for key, _ in self.items():
            yield key

    def __len__(self) -> int:
        """
        返回缓存项的数量，类似 len(dict)
        """
        return sum(1 for _ in self.items())

    @abstractmethod
    def set(self, key: str, value: Any, ttl: Optional[int] = None,
            region: Optional[str] = DEFAULT_CACHE_REGION, **kwargs) -> None:
        """
        设置缓存

        :param key: 缓存的键
        :param value: 缓存的值
        :param ttl: 缓存的存活时间，单位秒
        :param region: 缓存的区
        :param kwargs: 其他参数
        """
        pass

    @abstractmethod
    def exists(self, key: str, region: Optional[str] = DEFAULT_CACHE_REGION) -> bool:
        """
        判断缓存键是否存在

        :param key: 缓存的键
        :param region: 缓存的区
        :return: 存在返回 True，否则返回 False
        """
        pass

    @abstractmethod
    def get(self, key: str, region: Optional[str] = DEFAULT_CACHE_REGION) -> Any:
        """
        获取缓存

        :param key: 缓存的键
        :param region: 缓存的区
        :return: 返回缓存的值，如果缓存不存在返回 None
        """
        pass

    @abstractmethod
    def delete(self, key: str, region: Optional[str] = DEFAULT_CACHE_REGION) -> None:
        """
        删除缓存

        :param key: 缓存的键
        :param region: 缓存的区
        """
        pass

    @abstractmethod
    def clear(self, region: Optional[str] = DEFAULT_CACHE_REGION) -> None:
        """
        清除指定区域的缓存或全部缓存

        :param region: 缓存的区，为None时清空所有区缓存
        """
        pass

    @abstractmethod
    def items(self, region: Optional[str] = DEFAULT_CACHE_REGION) -> Generator[Tuple[str, Any], None, None]:
        """
        获取指定区域的所有缓存项

        :param region: 缓存的区
        :return: 返回一个字典，包含所有缓存键值对
        """
        pass

    def keys(self, region: Optional[str] = DEFAULT_CACHE_REGION) -> Generator[str, None, None]:
        """
        获取所有缓存键，类似 dict.keys()
        """
        for key, _ in self.items(region=region):
            yield key

    def values(self, region: Optional[str] = DEFAULT_CACHE_REGION) -> Generator[Any, None, None]:
        """
        获取所有缓存值，类似 dict.values()
        """
        for _, value in self.items(region=region):
            yield value

    def update(self, other: Dict[str, Any], region: Optional[str] = DEFAULT_CACHE_REGION,
               ttl: Optional[int] = None, **kwargs) -> None:
        """
        更新缓存，类似 dict.update()
        """
        for key, value in other.items():
            self.set(key, value, ttl=ttl, region=region, **kwargs)

    def pop(self, key: str, default: Any = None, region: Optional[str] = DEFAULT_CACHE_REGION) -> Any:
        """
        弹出缓存项，类似 dict.pop()
        """
        value = self.get(key, region=region)
        if value is not None:
            self.delete(key, region=region)
            return value
        if default is not None:
            return default
        raise KeyError(key)

    def popitem(self, region: Optional[str] = DEFAULT_CACHE_REGION) -> Tuple[str, Any]:
        """
        弹出最后一个缓存项，类似 dict.popitem()
        """
        items = list(self.items(region=region))
        if not items:
            raise KeyError("popitem(): cache is empty")
        key, value = items[-1]
        self.delete(key, region=region)
        return key, value

    def setdefault(self, key: str, default: Any = None, region: Optional[str] = DEFAULT_CACHE_REGION,
                   ttl: Optional[int] = None, **kwargs) -> Any:
        """
        设置默认值，类似 dict.setdefault()
        """
        value = self.get(key, region=region)
        if value is None:
            self.set(key, default, ttl=ttl, region=region, **kwargs)
            return default
        return value

    @abstractmethod
    def close(self) -> None:
        """
        关闭缓存连接
        """
        pass

    @staticmethod
    def get_region(region: Optional[str] = None) -> str:
        """
        获取缓存的区
        """
        return f"region:{region}" if region else "region:DEFAULT"

    @staticmethod
    def is_redis() -> bool:
        """
        判断当前缓存后端是否为 Redis
        """
        return settings.CACHE_BACKEND_TYPE == "redis"


class AsyncCacheBackend(CacheBackend):
    """
    缓存后端基类，定义通用的缓存接口（异步）
    """

    @abstractmethod
    async def set(self, key: str, value: Any, ttl: Optional[int] = None,
                  region: Optional[str] = DEFAULT_CACHE_REGION, **kwargs) -> None:
        """
        设置缓存

        :param key: 缓存的键
        :param value: 缓存的值
        :param ttl: 缓存的存活时间，单位秒
        :param region: 缓存的区
        :param kwargs: 其他参数
        """
        pass

    @abstractmethod
    async def exists(self, key: str, region: Optional[str] = DEFAULT_CACHE_REGION) -> bool:
        """
        判断缓存键是否存在

        :param key: 缓存的键
        :param region: 缓存的区
        :return: 存在返回 True，否则返回 False
        """
        pass

    @abstractmethod
    async def get(self, key: str, region: Optional[str] = DEFAULT_CACHE_REGION) -> Any:
        """
        获取缓存

        :param key: 缓存的键
        :param region: 缓存的区
        :return: 返回缓存的值，如果缓存不存在返回 None
        """
        pass

    @abstractmethod
    async def delete(self, key: str, region: Optional[str] = DEFAULT_CACHE_REGION) -> None:
        """
        删除缓存

        :param key: 缓存的键
        :param region: 缓存的区
        """
        pass

    @abstractmethod
    async def clear(self, region: Optional[str] = DEFAULT_CACHE_REGION) -> None:
        """
        清除指定区域的缓存或全部缓存

        :param region: 缓存的区，为None时清空所有区缓存
        """
        pass

    @abstractmethod
    async def items(self, region: Optional[str] = DEFAULT_CACHE_REGION) -> AsyncGenerator[Tuple[str, Any], None]:
        """
        获取指定区域的所有缓存项

        :param region: 缓存的区
        :return: 返回一个字典，包含所有缓存键值对
        """
        pass

    async def keys(self, region: Optional[str] = DEFAULT_CACHE_REGION) -> AsyncGenerator[str, None]:
        """
        获取所有缓存键，类似 dict.keys()（异步）
        """
        async for key, _ in self.items(region=region):
            yield key

    async def values(self, region: Optional[str] = DEFAULT_CACHE_REGION) -> AsyncGenerator[Any, None]:
        """
        获取所有缓存值，类似 dict.values()（异步）
        """
        async for _, value in self.items(region=region):
            yield value

    async def update(self, other: Dict[str, Any], region: Optional[str] = DEFAULT_CACHE_REGION,
                     ttl: Optional[int] = None, **kwargs) -> None:
        """
        更新缓存，类似 dict.update()（异步）
        """
        for key, value in other.items():
            await self.set(key, value, ttl=ttl, region=region, **kwargs)

    async def pop(self, key: str, default: Any = None, region: Optional[str] = DEFAULT_CACHE_REGION) -> Any:
        """
        弹出缓存项，类似 dict.pop()（异步）
        """
        value = await self.get(key, region=region)
        if value is not None:
            await self.delete(key, region=region)
            return value
        if default is not None:
            return default
        raise KeyError(key)

    async def popitem(self, region: Optional[str] = DEFAULT_CACHE_REGION) -> Tuple[str, Any]:
        """
        弹出最后一个缓存项，类似 dict.popitem()（异步）
        """
        items = []
        async for item in self.items(region=region):
            items.append(item)
        if not items:
            raise KeyError("popitem(): cache is empty")
        key, value = items[-1]
        await self.delete(key, region=region)
        return key, value

    async def setdefault(self, key: str, default: Any = None, region: Optional[str] = DEFAULT_CACHE_REGION,
                         ttl: Optional[int] = None, **kwargs) -> Any:
        """
        设置默认值，类似 dict.setdefault()（异步）