GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-08-11 00:25:36 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(cache): simplify recognition cache persistence

Browse Source
This commit is contained in:
jxxghp
2026-08-06 07:48:43 +08:00
parent a23ac6c56d
commit 4b1df72a4a
15 changed files with 518 additions and 640 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
app/api/endpoints/douban.py
View File

@@ -4,70 +4,13 @@ from fastapi import APIRouter, Depends
from app import schemas
from app.chain.douban import DoubanChain
from app.core.config import settings
from app.core.context import MediaInfo
from app.core.security import verify_token
from app.db.models.user import User
from app.db.systemconfig_oper import SystemConfigOper
from app.db.user_oper import get_current_active_superuser_async
from app.modules.douban.douban_cache import DoubanCache
from app.schemas import MediaType
from app.schemas.types import SystemConfigKey
router = APIRouter()
@router.get(
"/cache", summary="查询豆瓣识别缓存", response_model=schemas.Response
)
async def douban_recognition_cache(
_: User = Depends(get_current_active_superuser_async),
) -> schemas.Response:
"""查询可管理的豆瓣识别缓存。"""
cache_items = DoubanCache().list_items()
recognized_count = sum(1 for item in cache_items if item["douban_id"])
return schemas.Response(
success=True,
data={
"count": len(cache_items),
"recognized": recognized_count,
"unrecognized": len(cache_items) - recognized_count,
"shared_recognized": SystemConfigOper().get(
SystemConfigKey.MediaRecognizeShareCount
) or 0,
"shared_recognize_enabled": settings.MEDIA_RECOGNIZE_SHARE,
"data": cache_items,
},
)
@router.delete(
"/cache/{cache_key:path}",
summary="删除指定豆瓣识别缓存",
response_model=schemas.Response,
)
async def delete_douban_recognition_cache(
cache_key: str,
_: User = Depends(get_current_active_superuser_async),
) -> schemas.Response:
"""按缓存键删除单条豆瓣识别缓存。"""
deleted_item = DoubanCache().delete(cache_key)
if not deleted_item:
return schemas.Response(success=False, message="豆瓣识别缓存不存在")
return schemas.Response(success=True, message="豆瓣识别缓存删除成功")
@router.delete(
"/cache", summary="清空豆瓣识别缓存", response_model=schemas.Response
)
async def clear_douban_recognition_cache(
_: User = Depends(get_current_active_superuser_async),
) -> schemas.Response:
"""清空全部豆瓣识别缓存。"""
DoubanCache().clear()
return schemas.Response(success=True, message="豆瓣识别缓存清理完成")
@router.get(
"/person/{person_id}", summary="人物详情", response_model=schemas.MediaPerson
)

2
app/core/config.py
View File

@@ -178,7 +178,7 @@ class ConfigModel(BaseModel):
PACKAGE_CACHE_DAYS: int = 90
# pip/uv 包下载缓存根目录，留空时使用配置目录下的 .cache
PACKAGE_CACHE_ROOT: Optional[str] = None
# 元数据识别缓存过期时间（小时），0为自动
# 单条元数据识别缓存有效期（小时），0为自动
META_CACHE_EXPIRE: int = 0
# ==================== 网络代理配置 ====================

3
app/locales/en-US.json
View File

@@ -183,9 +183,6 @@
"TheMovieDb 识别缓存不存在": "TheMovieDb recognition cache does not exist",
"TheMovieDb 识别缓存删除成功": "TheMovieDb recognition cache deleted successfully",
"TheMovieDb 识别缓存清理完成": "TheMovieDb recognition cache cleanup completed",
"豆瓣识别缓存不存在": "Douban recognition cache does not exist",
"豆瓣识别缓存删除成功": "Douban recognition cache deleted successfully",
"豆瓣识别缓存清理完成": "Douban recognition cache cleanup completed",
"重新识别完成": "Re-recognition completed",
"未识别到新名称": "Unable to recognize new name",
"缺少参数": "Missing parameters",

5
app/locales/zh-CN.json
View File

@@ -110,10 +110,7 @@
"Redis连接失败，请检查配置": "Redis连接失败，请检查配置",
"TheMovieDb 识别缓存不存在": "TheMovieDb 识别缓存不存在",
"TheMovieDb 识别缓存删除成功": "TheMovieDb 识别缓存删除成功",
"TheMovieDb 识别缓存清理完成": "TheMovieDb 识别缓存清理完成",
"豆瓣识别缓存不存在": "豆瓣识别缓存不存在",
"豆瓣识别缓存删除成功": "豆瓣识别缓存删除成功",
"豆瓣识别缓存清理完成": "豆瓣识别缓存清理完成"
"TheMovieDb 识别缓存清理完成": "TheMovieDb 识别缓存清理完成"
},
"message_patterns": [
{

3
app/locales/zh-TW.json
View File

@@ -183,9 +183,6 @@
"TheMovieDb 识别缓存不存在": "TheMovieDb 識別快取不存在",
"TheMovieDb 识别缓存删除成功": "TheMovieDb 識別快取刪除成功",
"TheMovieDb 识别缓存清理完成": "TheMovieDb 識別快取清理完成",
"豆瓣识别缓存不存在": "豆瓣識別快取不存在",
"豆瓣识别缓存删除成功": "豆瓣識別快取刪除成功",
"豆瓣识别缓存清理完成": "豆瓣識別快取清理完成",
"重新识别完成": "重新識別完成",
"未识别到新名称": "未識別到新名稱",
"缺少参数": "缺少參數",

195
app/modules/douban/__init__.py
View File

@@ -11,7 +11,6 @@ from app.core.metainfo import MetaInfo
from app.log import logger
from app.modules import _ModuleBase
from app.modules.douban.apiv2 import DoubanApi
from app.modules.douban.douban_cache import DoubanCache
from app.modules.douban.scraper import DoubanScraper
from app.schemas import MediaPerson, APIRateLimitException
from app.schemas.types import MediaType, ModuleType, MediaRecognizeType
@@ -24,12 +23,10 @@ from app.utils.zhconv import convert as zhconv_convert
class DoubanModule(_ModuleBase):
doubanapi: DoubanApi = None
scraper: DoubanScraper = None
cache: DoubanCache = None
def init_module(self) -> None:
self.doubanapi = DoubanApi()
self.scraper = DoubanScraper()
self.cache = DoubanCache()
def stop(self):
self.doubanapi.close()
@@ -110,7 +107,6 @@ class DoubanModule(_ModuleBase):
def _recognize_media_core(self, meta: MetaBase = None,
mtype: MediaType = None,
doubanid: Optional[str] = None,
cache: Optional[bool] = True,
douban_info_func=None,
match_doubaninfo_func=None,
**kwargs) -> Optional[MediaInfo]:
@@ -119,7 +115,6 @@ class DoubanModule(_ModuleBase):
:param meta: 识别的元数据
:param mtype: 识别的媒体类型，与doubanid配套
:param doubanid: 豆瓣ID
:param cache: 是否使用缓存
:param douban_info_func: 获取豆瓣信息的函数
:param match_doubaninfo_func: 匹配豆瓣信息的函数
:return: 识别的媒体信息，包括剧集信息
@@ -134,69 +129,39 @@ class DoubanModule(_ModuleBase):
):
return None
if not meta:
# 未提供元数据时，直接查询豆瓣信息，不使用缓存
cache_info = {}
if doubanid:
info = douban_info_func(
doubanid=doubanid,
mtype=mtype or (meta.type if meta else None),
)
elif not meta.name:
logger.error("识别媒体信息时未提供元数据名称")
return None
else:
# 读取缓存
if mtype:
meta.type = mtype
if doubanid:
meta.doubanid = doubanid
cache_info = self.cache.get(meta) if cache else {}
cache_hit = False
# 识别豆瓣信息
if not cache_info or not cache:
# 缓存没有或者强制不使用缓存
if doubanid:
# 直接查询详情
info = douban_info_func(doubanid=doubanid, mtype=mtype or meta.type)
elif meta:
info = {}
for name in self._prepare_search_names(meta):
if meta.begin_season is not None:
logger.info(f"正在识别 {name} 第{meta.begin_season}季 ...")
else:
logger.info(f"正在识别 {name} ...")
# 匹配豆瓣信息
match_info = match_doubaninfo_func(name=name,
mtype=mtype or meta.type,
year=meta.year,
season=meta.begin_season)
if match_info:
# 匹配到豆瓣信息
info = douban_info_func(
doubanid=match_info.get("id"),
mtype=mtype or meta.type
)
if info:
break
else:
logger.error("识别媒体信息时未提供元数据或豆瓣ID")
return None
# 保存到缓存
if meta and cache:
self.cache.update(meta, info)
else:
# 使用缓存信息
cache_hit = True
if cache_info.get("title"):
logger.info(f"{meta.name} 使用豆瓣识别缓存：{cache_info.get('title')}")
info = douban_info_func(mtype=cache_info.get("type"),
doubanid=cache_info.get("id"))
else:
logger.info(f"{meta.name} 使用豆瓣识别缓存：无法识别")
info = None
info = {}
for name in self._prepare_search_names(meta):
if meta.begin_season is not None:
logger.info(f"正在识别 {name} 第{meta.begin_season}季 ...")
else:
logger.info(f"正在识别 {name} ...")
match_info = match_doubaninfo_func(
name=name,
mtype=mtype or meta.type,
year=meta.year,
season=meta.begin_season,
)
if match_info:
info = douban_info_func(
doubanid=match_info.get("id"),
mtype=mtype or meta.type,
)
if info:
break
if info:
# 赋值TMDB信息并返回
mediainfo = MediaInfo(douban_info=info)
mediainfo.recognize_cache_hit = cache_hit
if meta:
logger.info(f"{meta.name} 豆瓣识别结果：{mediainfo.type.value} "
f"{mediainfo.title_year} "
@@ -213,7 +178,6 @@ class DoubanModule(_ModuleBase):
async def _async_recognize_media_core(self, meta: MetaBase = None,
mtype: MediaType = None,
doubanid: Optional[str] = None,
cache: Optional[bool] = True,
async_douban_info_func=None,
async_match_doubaninfo_func=None,
**kwargs) -> Optional[MediaInfo]:
@@ -222,7 +186,6 @@ class DoubanModule(_ModuleBase):
:param meta: 识别的元数据
:param mtype: 识别的媒体类型，与doubanid配套
:param doubanid: 豆瓣ID
:param cache: 是否使用缓存
:param async_douban_info_func: 获取豆瓣信息的异步函数
:param async_match_doubaninfo_func: 匹配豆瓣信息的异步函数
:return: 识别的媒体信息，包括剧集信息
@@ -237,69 +200,39 @@ class DoubanModule(_ModuleBase):
):
return None
if not meta:
# 未提供元数据时，直接查询豆瓣信息，不使用缓存
cache_info = {}
if doubanid:
info = await async_douban_info_func(
doubanid=doubanid,
mtype=mtype or (meta.type if meta else None),
)
elif not meta.name:
logger.error("识别媒体信息时未提供元数据名称")
return None
else:
# 读取缓存
if mtype:
meta.type = mtype
if doubanid:
meta.doubanid = doubanid
cache_info = self.cache.get(meta) if cache else {}
cache_hit = False
# 识别豆瓣信息
if not cache_info or not cache:
# 缓存没有或者强制不使用缓存
if doubanid:
# 直接查询详情
info = await async_douban_info_func(doubanid=doubanid, mtype=mtype or meta.type)
elif meta:
info = {}
for name in self._prepare_search_names(meta):
if meta.begin_season is not None:
logger.info(f"正在识别 {name} 第{meta.begin_season}季 ...")
else:
logger.info(f"正在识别 {name} ...")
# 匹配豆瓣信息
match_info = await async_match_doubaninfo_func(name=name,
mtype=mtype or meta.type,
year=meta.year,
season=meta.begin_season)
if match_info:
# 匹配到豆瓣信息
info = await async_douban_info_func(
doubanid=match_info.get("id"),
mtype=mtype or meta.type
)
if info:
break
else:
logger.error("识别媒体信息时未提供元数据或豆瓣ID")
return None
# 保存到缓存
if meta and cache:
self.cache.update(meta, info)
else:
# 使用缓存信息
cache_hit = True
if cache_info.get("title"):
logger.info(f"{meta.name} 使用豆瓣识别缓存：{cache_info.get('title')}")
info = await async_douban_info_func(mtype=cache_info.get("type"),
doubanid=cache_info.get("id"))
else:
logger.info(f"{meta.name} 使用豆瓣识别缓存：无法识别")
info = None
info = {}
for name in self._prepare_search_names(meta):
if meta.begin_season is not None:
logger.info(f"正在识别 {name} 第{meta.begin_season}季 ...")
else:
logger.info(f"正在识别 {name} ...")
match_info = await async_match_doubaninfo_func(
name=name,
mtype=mtype or meta.type,
year=meta.year,
season=meta.begin_season,
)
if match_info:
info = await async_douban_info_func(
doubanid=match_info.get("id"),
mtype=mtype or meta.type,
)
if info:
break
if info:
# 赋值TMDB信息并返回
mediainfo = MediaInfo(douban_info=info)
mediainfo.recognize_cache_hit = cache_hit
if meta:
logger.info(f"{meta.name} 豆瓣识别结果：{mediainfo.type.value} "
f"{mediainfo.title_year} "
@@ -316,21 +249,18 @@ class DoubanModule(_ModuleBase):
def recognize_media(self, meta: MetaBase = None,
mtype: MediaType = None,
doubanid: Optional[str] = None,
cache: Optional[bool] = True,
**kwargs) -> Optional[MediaInfo]:
"""
识别媒体信息
:param meta: 识别的元数据
:param mtype: 识别的媒体类型，与doubanid配套
:param doubanid: 豆瓣ID
:param cache: 是否使用缓存
:return: 识别的媒体信息，包括剧集信息
"""
return self._recognize_media_core(
meta=meta,
mtype=mtype,
doubanid=doubanid,
cache=cache,
douban_info_func=self.douban_info,
match_doubaninfo_func=self.match_doubaninfo,
**kwargs
@@ -339,51 +269,23 @@ class DoubanModule(_ModuleBase):
async def async_recognize_media(self, meta: MetaBase = None,
mtype: MediaType = None,
doubanid: Optional[str] = None,
cache: Optional[bool] = True,
**kwargs) -> Optional[MediaInfo]:
"""
识别媒体信息（异步版本）
:param meta: 识别的元数据
:param mtype: 识别的媒体类型，与doubanid配套
:param doubanid: 豆瓣ID
:param cache: 是否使用缓存
:return: 识别的媒体信息，包括剧集信息
"""
return await self._async_recognize_media_core(
meta=meta,
mtype=mtype,
doubanid=doubanid,
cache=cache,
async_douban_info_func=self.async_douban_info,
async_match_doubaninfo_func=self.async_match_doubaninfo,
**kwargs
)
def update_recognize_cache(
self,
meta: MetaBase,
mediainfo: MediaInfo,
) -> Optional[bool]:
"""
回填豆瓣本地识别缓存，覆盖名称负缓存，避免共享识别后重复回查。
"""
if not meta or not mediainfo:
return None
if mediainfo.source != "douban" or not mediainfo.douban_info:
return None
self.cache.update(meta, mediainfo.douban_info)
return True
async def async_update_recognize_cache(
self,
meta: MetaBase,
mediainfo: MediaInfo,
) -> Optional[bool]:
"""
异步回填豆瓣本地识别缓存。
"""
return self.update_recognize_cache(meta=meta, mediainfo=mediainfo)
@rate_limit_exponential(source="douban_info")
def douban_info(self, doubanid: str, mtype: MediaType = None, raise_exception: bool = True) -> Optional[dict]:
"""
@@ -1272,7 +1174,6 @@ class DoubanModule(_ModuleBase):
"""
logger.info("开始清除豆瓣缓存 ...")
self.doubanapi.clear_cache()
self.cache.clear()
logger.info("豆瓣缓存清除完成")
def douban_movie_credits(self, doubanid: str) -> List[schemas.MediaPerson]:

199
app/modules/douban/douban_cache.py
View File

@@ -1,199 +0,0 @@
import pickle
import traceback
from pathlib import Path
from threading import RLock
from typing import Optional
from app.core.cache import TTLCache
from app.core.config import settings
from app.core.meta import MetaBase
from app.core.metainfo import MetaInfo
from app.log import logger
from app.schemas.types import MediaType
from app.utils.singleton import WeakSingleton
lock = RLock()
class DoubanCache(metaclass=WeakSingleton):
"""
豆瓣缓存数据
{
"id": '',
"title": '',
"year": '',
"type": MediaType
}
"""
# 豆瓣缓存过期
_douban_cache_expire: bool = True
def __init__(self):
"""初始化豆瓣识别缓存并恢复本地持久化数据。"""
self.maxsize = settings.CONF.douban
self.ttl = settings.CONF.meta
self.region = "__douban_cache__"
self._meta_filepath = settings.TEMP_PATH / self.region
# 初始化缓存
self._cache = TTLCache(region=self.region, maxsize=self.maxsize, ttl=self.ttl)
# 非Redis加载本地缓存数据
if not self._cache.is_redis():
for key, value in self.__load(self._meta_filepath).items():
self._cache.set(key, value)
def clear(self):
"""
清空所有豆瓣缓存
"""
with lock:
self._cache.clear()
self.save(force=True)
def list_items(self) -> list[dict]:
"""返回可供管理界面展示的豆瓣识别缓存列表。"""
with lock:
cache_items = []
for key, value in self._cache.items():
if not isinstance(value, dict):
continue
media_type = value.get("type")
if not isinstance(media_type, MediaType):
try:
media_type = MediaType(media_type)
except (TypeError, ValueError):
media_type = None
cache_items.append({
"key": key,
"douban_id": value.get("id") or 0,
"title": value.get("title") or "",
"year": value.get("year") or "",
"media_type": media_type.to_agent() if media_type else "unknown",
"poster_path": value.get("poster_path") or "",
})
return sorted(cache_items, key=lambda item: item["key"])
@staticmethod
def __get_key(meta: MetaBase) -> str:
"""
获取缓存KEY
"""
return f"[{meta.type.value if meta.type else '未知'}]" \
f"{meta.doubanid or meta.name}-{meta.year}-{meta.begin_season}"
def get(self, meta: MetaBase):
"""
根据KEY值获取缓存值
"""
key = self.__get_key(meta)
with lock:
return self._cache.get(key) or {}
def delete(self, key: str) -> dict:
"""
删除缓存信息
@param key: 缓存key
@return: 被删除的缓存内容
"""
with lock:
redis_data = self._cache.get(key)
if redis_data:
self._cache.delete(key)
self.save(force=True)
return redis_data
return {}
def modify(self, key: str, title: str) -> dict:
"""
修改缓存信息
@param key: 缓存key
@param title: 标题
@return: 被修改后缓存内容
"""
with lock:
redis_data = self._cache.get(key)
if redis_data:
redis_data["title"] = title
self._cache.set(key, redis_data)
return redis_data
return {}
@staticmethod
def __load(path: Path) -> dict:
"""
从文件中加载缓存
"""
try:
if path.exists():
with open(path, 'rb') as f:
data = pickle.load(f)
return data
except Exception as e:
logger.error(f"加载缓存失败: {str(e)} - {traceback.format_exc()}")
return {}
def update(self, meta: MetaBase, info: dict) -> None:
"""
新增或更新缓存条目
"""
if info:
# 缓存标题
cache_title = info.get("title")
# 缓存年份
cache_year = info.get('year')
# 类型
if isinstance(info.get('media_type'), MediaType):
mtype = info.get('media_type')
elif info.get("type"):
mtype = MediaType.MOVIE if info.get("type") == "movie" else MediaType.TV
else:
meta = MetaInfo(cache_title)
if meta.begin_season is not None:
mtype = MediaType.TV
else:
mtype = MediaType.MOVIE
# 海报
poster_path = info.get("pic", {}).get("large")
if not poster_path and info.get("cover_url"):
poster_path = info.get("cover_url")
if not poster_path and info.get("cover"):
poster_path = info.get("cover").get("url")
with lock:
self._cache.set(self.__get_key(meta), {
"id": info.get("id"),
"type": mtype,
"year": cache_year,
"title": cache_title,
"poster_path": poster_path
})
elif info is not None:
# None时不缓存，此时代表网络错误，允许重复请求
with lock:
self._cache.set(self.__get_key(meta), {
"id": 0
})
def save(self, force: Optional[bool] = False) -> None:
"""
保存缓存数据到文件
"""
# Redis不需要保存到本地文件
if self._cache.is_redis():
return
# 本地文件
meta_data = self.__load(self._meta_filepath)
# 当前缓存数据（去除无法识别）
new_meta_data = {k: v for k, v in self._cache.items() if v.get("id")}
if not force \
and meta_data.keys() == new_meta_data.keys():
return
# 写入本地
with open(self._meta_filepath, 'wb') as f:
pickle.dump(new_meta_data, f, pickle.HIGHEST_PROTOCOL) # noqa
def __del__(self):
"""实例释放前保存非 Redis 缓存。"""
self.save()

172
app/modules/themoviedb/tmdb_cache.py
View File

@@ -1,9 +1,10 @@
import pickle
import traceback
from pathlib import Path
from math import ceil
from threading import RLock
from time import time
from app.core.cache import TTLCache
from app.core.cache import FileCache, TTLCache
from app.core.config import settings
from app.core.meta import MetaBase
from app.log import logger
@@ -11,6 +12,9 @@ from app.schemas.types import MediaType
from app.utils.singleton import WeakSingleton
lock = RLock()
PERSISTENCE_VERSION = 1
PERSISTENCE_REGION = "recognize"
PERSISTENCE_KEY = "tmdb"
class TmdbCache(metaclass=WeakSingleton):
@@ -23,21 +27,78 @@ class TmdbCache(metaclass=WeakSingleton):
"type": MediaType
}
"""
# TMDB缓存过期
_tmdb_cache_expire: bool = True
def __init__(self):
"""初始化 TMDB 识别缓存并恢复本地持久化数据。"""
self.maxsize = settings.CONF.douban
"""初始化 TMDB 识别缓存并恢复未过期的持久化数据。"""
self.maxsize = settings.CONF.tmdb
self.ttl = settings.CONF.meta
self.region = "__tmdb_cache__"
self._meta_filepath = settings.TEMP_PATH / self.region
# 初始化缓存
self._cache = TTLCache(region=self.region, maxsize=self.maxsize, ttl=self.ttl)
# 非Redis加载本地缓存数据
self._expires_at: dict[str, float] = {}
self._dirty = False
self._file_cache = None
self._legacy_file_cache = None
self._legacy_cache_found = False
if not self._cache.is_redis():
for key, value in self.__load(self._meta_filepath).items():
self._cache.set(key, value)
self._file_cache = FileCache(base=settings.CACHE_PATH, ttl=self.ttl)
self._legacy_file_cache = FileCache(base=settings.TEMP_PATH.parent, ttl=self.ttl)
self._restore()
def _restore(self) -> None:
"""从统一文件缓存恢复仍在有效期内的 TMDB 识别数据。"""
try:
content = self._file_cache.get(PERSISTENCE_KEY, region=PERSISTENCE_REGION)
if not content:
content = self._legacy_file_cache.get(
self.region,
region=settings.TEMP_PATH.name,
)
if content:
self._legacy_cache_found = True
self._dirty = True
if not content:
return
payload = pickle.loads(content)
now = time()
if (
isinstance(payload, dict)
and payload.get("version") == PERSISTENCE_VERSION
and isinstance(payload.get("items"), dict)
):
items = payload["items"]
elif isinstance(payload, dict):
# 旧版缓存没有保存过期时间，迁移时从当前时刻重新计算一次有效期。
items = {
key: {"value": value, "expires_at": now + self.ttl}
for key, value in payload.items()
}
self._dirty = True
else:
return
for key, item in items.items():
if not isinstance(item, dict):
self._dirty = True
continue
value = item.get("value")
expires_at = item.get("expires_at")
if not isinstance(value, dict) or not isinstance(expires_at, (int, float)):
self._dirty = True
continue
remaining_ttl = expires_at - now
if remaining_ttl <= 0:
self._dirty = True
continue
self._cache.set(key, value, ttl=ceil(remaining_ttl))
self._expires_at[key] = expires_at
except Exception as err:
logger.error(f"加载TMDB识别缓存失败：{str(err)} - {traceback.format_exc()}")
def _set(self, key: str, value: dict) -> None:
"""写入单条 TMDB 识别缓存并记录其独立过期时间。"""
self._cache.set(key, value)
if not self._cache.is_redis():
self._expires_at[key] = time() + self.ttl
self._dirty = True
def clear(self):
"""
@@ -45,6 +106,8 @@ class TmdbCache(metaclass=WeakSingleton):
"""
with lock:
self._cache.clear()
self._expires_at.clear()
self._dirty = True
self.save(force=True)
def list_items(self) -> list[dict]:
@@ -87,7 +150,10 @@ class TmdbCache(metaclass=WeakSingleton):
key = self.__get_key(meta)
with lock:
return self._cache.get(key) or {}
cache_data = self._cache.get(key)
if not cache_data and self._expires_at.pop(key, None) is not None:
self._dirty = True
return cache_data or {}
def delete(self, key: str) -> dict:
"""
@@ -99,6 +165,8 @@ class TmdbCache(metaclass=WeakSingleton):
redis_data = self._cache.get(key)
if redis_data:
self._cache.delete(key)
self._expires_at.pop(key, None)
self._dirty = True
self.save(force=True)
return redis_data
return {}
@@ -114,24 +182,10 @@ class TmdbCache(metaclass=WeakSingleton):
redis_data = self._cache.get(key)
if redis_data:
redis_data['title'] = title
self._cache.set(key, redis_data)
self._set(key, redis_data)
return redis_data
return {}
@staticmethod
def __load(path: Path) -> dict:
"""
从文件中加载缓存
"""
try:
if path.exists():
with open(path, 'rb') as f:
data = pickle.load(f)
return data
except Exception as e:
logger.error(f'加载缓存失败：{str(e)} - {traceback.format_exc()}')
return {}
def update(self, meta: MetaBase, info: dict) -> None:
"""
新增或更新缓存条目
@@ -157,32 +211,68 @@ class TmdbCache(metaclass=WeakSingleton):
"poster_path": info.get("poster_path"),
"backdrop_path": info.get("backdrop_path")
}
self._cache.set(key, cache_data)
self._set(key, cache_data)
elif info is not None:
# None时不缓存，此时代表网络错误，允许重复请求
with lock:
self._cache.set(key, {"id": 0})
self._set(key, {"id": 0})
def save(self, force: bool = False) -> None:
"""
保存缓存数据到文件
使用统一文件缓存保存未过期的 TMDB 识别数据。
"""
# Redis不需要保存到本地文件
if self._cache.is_redis():
return
with lock:
now = time()
cache_items = dict(self._cache.items())
active_keys = set(cache_items)
stale_keys = set(self._expires_at) - active_keys
if stale_keys:
for key in stale_keys:
self._expires_at.pop(key, None)
self._dirty = True
# Redis不可用时，保存到本地文件
meta_data = self.__load(self._meta_filepath)
# 当前缓存，去除无法识别
new_meta_data = {k: v for k, v in self._cache.items() if v.get("id")}
persisted_items = {}
for key, value in cache_items.items():
expires_at = self._expires_at.get(key)
if expires_at is None:
expires_at = now + self.ttl
self._expires_at[key] = expires_at
self._dirty = True
if expires_at <= now or not value.get("id"):
continue
persisted_items[key] = {
"value": value,
"expires_at": expires_at,
}
if not force \
and meta_data.keys() == new_meta_data.keys():
return
if not force and not self._dirty:
return
with open(self._meta_filepath, 'wb') as f:
pickle.dump(new_meta_data, f, pickle.HIGHEST_PROTOCOL) # type: ignore
try:
if persisted_items:
payload = {
"version": PERSISTENCE_VERSION,
"items": persisted_items,
}
self._file_cache.set(
PERSISTENCE_KEY,
pickle.dumps(payload, pickle.HIGHEST_PROTOCOL),
region=PERSISTENCE_REGION,
)
else:
self._file_cache.delete(PERSISTENCE_KEY, region=PERSISTENCE_REGION)
if self._legacy_cache_found:
self._legacy_file_cache.delete(
self.region,
region=settings.TEMP_PATH.name,
)
self._legacy_cache_found = False
self._dirty = False
except Exception as err:
logger.error(f"保存TMDB识别缓存失败：{str(err)} - {traceback.format_exc()}")
def __del__(self):
"""实例释放前保存非 Redis 缓存。"""

10
app/scheduler.py
View File

@@ -660,16 +660,6 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
kwargs={"job_id": "scheduler_job"},
)
# 缓存清理服务，每隔24小时
self._scheduler.add_job(
self.start,
"interval",
id="clear_cache",
name="缓存清理",
hours=settings.CONF.meta / 3600,
kwargs={"job_id": "clear_cache"},
)
# 数据表清理服务，每天凌晨执行一次
if settings.DATA_CLEANUP_ENABLE:
self._scheduler.add_job(

5
docs/mcp-api.md
View File

@@ -213,11 +213,8 @@ AniList 榜单、探索、详情、人物和推荐接口优先通过 `anilist-ch
| GET | `/api/v1/tmdb/cache` | 查询 TheMovieDb 识别缓存统计、共享识别累计成功命中次数及开关状态 |
| DELETE | `/api/v1/tmdb/cache/{cache_key}` | 按缓存键删除单条 TheMovieDb 识别缓存，缓存键需要进行 URL 编码 |
| DELETE | `/api/v1/tmdb/cache` | 清空全部 TheMovieDb 识别缓存 |
| GET | `/api/v1/douban/cache` | 查询豆瓣识别缓存统计、共享识别累计成功命中次数及开关状态 |
| DELETE | `/api/v1/douban/cache/{cache_key}` | 按缓存键删除单条豆瓣识别缓存，缓存键需要进行 URL 编码 |
| DELETE | `/api/v1/douban/cache` | 清空全部豆瓣识别缓存 |
缓存查询响应的 `data` 包含 `count`、`recognized`、`unrecognized`、`data`，以及共享识别统计字段
TMDB 缓存查询响应的 `data` 包含 `count`、`recognized`、`unrecognized`、`data`，以及共享识别统计字段
`shared_recognized` 和开关字段 `shared_recognize_enabled`。共享命中次数仅在共享结果驱动的二次媒体识别成功后累计。
### 插件补充接口

7
skills/moviepilot-api/SKILL.md
View File

@@ -444,9 +444,9 @@ Streaming search sends `{"type":"heartbeat"}` every 15 seconds without business
| POST | `/api/v1/torrent/cache/refresh` | Refresh torrent cache |
| POST | `/api/v1/torrent/cache/reidentify/{domain}/{torrent_hash}` | Re-identify torrent. Params: `tmdbid`, `doubanid` |
### Recognition Cache (6 endpoints)
### Recognition Cache (3 endpoints)
The two list endpoints return local cache totals plus `shared_recognized` and
The list endpoint returns local cache totals plus `shared_recognized` and
`shared_recognize_enabled` for the persisted successful shared-recognition count.
| Method | Path | Description |
@@ -454,9 +454,6 @@ The two list endpoints return local cache totals plus `shared_recognized` and
| GET | `/api/v1/tmdb/cache` | Get TheMovieDb recognition cache statistics |
| DELETE | `/api/v1/tmdb/cache/{cache_key}` | Delete one URL-encoded TheMovieDb recognition cache key |
| DELETE | `/api/v1/tmdb/cache` | Clear TheMovieDb recognition cache |
| GET | `/api/v1/douban/cache` | Get Douban recognition cache statistics |
| DELETE | `/api/v1/douban/cache/{cache_key}` | Delete one URL-encoded Douban recognition cache key |
| DELETE | `/api/v1/douban/cache` | Clear Douban recognition cache |
### Message (8 endpoints)

166
tests/test_douban_cache_management.py
View File

@@ -1,166 +0,0 @@
import asyncio
import inspect
from unittest.mock import Mock
from app.api.endpoints import douban as douban_endpoint
from app.db.user_oper import get_current_active_superuser_async
from app.modules.douban.douban_cache import DoubanCache
from app.schemas.types import MediaType, SystemConfigKey
class _MemoryCacheStub:
"""提供豆瓣缓存管理测试所需的最小内存后端。"""
def __init__(self, data: dict):
"""使用给定字典初始化测试缓存。"""
self.data = data
def items(self):
"""返回全部缓存条目。"""
return self.data.items()
def get(self, key: str):
"""读取指定缓存条目。"""
return self.data.get(key)
def delete(self, key: str):
"""删除指定缓存条目。"""
self.data.pop(key, None)
def set(self, key: str, value):
"""写入指定缓存条目。"""
self.data[key] = value
def clear(self):
"""清空全部缓存条目。"""
self.data.clear()
def _build_douban_cache(data: dict) -> DoubanCache:
"""构造绕过单例初始化的豆瓣缓存测试实例。"""
cache = object.__new__(DoubanCache)
cache._cache = _MemoryCacheStub(data)
cache.save = lambda force=False: None
return cache
def test_douban_cache_management_endpoints_require_superuser():
"""豆瓣识别缓存管理接口必须仅允许超级管理员访问。"""
endpoints = [
douban_endpoint.douban_recognition_cache,
douban_endpoint.delete_douban_recognition_cache,
douban_endpoint.clear_douban_recognition_cache,
]
for endpoint in endpoints:
dependency = inspect.signature(endpoint).parameters["_"].default.dependency
assert dependency is get_current_active_superuser_async
def test_douban_cache_list_items_normalizes_media_type_and_sorting():
"""豆瓣管理列表应输出稳定顺序和前端可识别的媒体类型。"""
cache = _build_douban_cache({
"[电视剧]Zulu-2024-1": {
"id": "2",
"title": "Zulu",
"type": MediaType.TV,
"year": "2024",
},
"[电影]Alpha-2023-None": {
"id": "1",
"title": "Alpha",
"type": "电影",
"year": "2023",
"poster_path": "https://example.com/alpha.jpg",
},
"[电影]Missing-2022-None": {"id": 0},
})
items = cache.list_items()
assert [item["title"] for item in items] == ["Alpha", "", "Zulu"]
assert [item["media_type"] for item in items] == ["movie", "unknown", "tv"]
assert items[0]["poster_path"] == "https://example.com/alpha.jpg"
assert items[1]["douban_id"] == 0
def test_douban_cache_infers_special_season_title_as_tv():
"""缺少显式类型时，S00 标题仍应按电视剧写入缓存。"""
cache = _build_douban_cache({})
cache.update(
meta=None,
info={"id": "special", "title": "测试剧 S00", "year": "2024"},
)
cached = next(iter(cache._cache.data.values()))
assert cached["type"] == MediaType.TV
def test_douban_cache_delete_and_clear_persist_immediately(monkeypatch):
"""豆瓣管理操作应修改运行时缓存并立即触发本地持久化。"""
cache = _build_douban_cache({"first": {"id": "1"}, "second": {"id": "2"}})
saved_forces = []
monkeypatch.setattr(cache, "save", lambda force=False: saved_forces.append(force))
assert cache.delete("first") == {"id": "1"}
assert cache.delete("missing") == {}
cache.clear()
assert cache.list_items() == []
assert saved_forces == [True, True]
def test_douban_cache_endpoint_returns_management_statistics(monkeypatch):
"""豆瓣查询接口应返回识别成功和失败条目的统计。"""
cache = _build_douban_cache({
"recognized": {"id": "1", "title": "Alpha", "type": MediaType.MOVIE},
"unrecognized": {"id": 0},
})
get_system_config = Mock(return_value=None)
monkeypatch.setattr(douban_endpoint, "DoubanCache", lambda: cache)
monkeypatch.setattr(
douban_endpoint,
"SystemConfigOper",
lambda: type("SystemConfigStub", (), {"get": get_system_config})(),
)
monkeypatch.setattr(douban_endpoint.settings, "MEDIA_RECOGNIZE_SHARE", False)
response = asyncio.run(douban_endpoint.douban_recognition_cache(None))
assert response.success is True
assert response.data["count"] == 2
assert response.data["recognized"] == 1
assert response.data["unrecognized"] == 1
assert response.data["shared_recognized"] == 0
assert response.data["shared_recognize_enabled"] is False
get_system_config.assert_called_once_with(
SystemConfigKey.MediaRecognizeShareCount
)
def test_douban_cache_delete_endpoint_reports_missing_item(monkeypatch):
"""豆瓣删除接口应区分成功删除与缓存不存在。"""
cache = _build_douban_cache({"existing": {"id": "1"}})
monkeypatch.setattr(douban_endpoint, "DoubanCache", lambda: cache)
deleted_response = asyncio.run(
douban_endpoint.delete_douban_recognition_cache("existing", None)
)
missing_response = asyncio.run(
douban_endpoint.delete_douban_recognition_cache("missing", None)
)
assert deleted_response.success is True
assert missing_response.success is False
def test_douban_cache_clear_endpoint_removes_all_items(monkeypatch):
"""豆瓣清空接口应删除全部识别缓存。"""
cache = _build_douban_cache({"existing": {"id": "1"}})
monkeypatch.setattr(douban_endpoint, "DoubanCache", lambda: cache)
response = asyncio.run(douban_endpoint.clear_douban_recognition_cache(None))
assert response.success is True
assert cache.list_items() == []

80
tests/test_douban_recognition.py Normal file
View File

@@ -0,0 +1,80 @@
import asyncio
from unittest.mock import Mock
from unittest.mock import AsyncMock
from app.core.meta import MetaBase
from app.modules.douban import DoubanModule
from app.schemas.types import MediaType
def test_douban_recognize_does_not_keep_dedicated_mapping_cache():
"""豆瓣识别应每次执行匹配，不再保留专用标题映射缓存。"""
module = DoubanModule()
meta = MetaBase("测试电影")
meta.name = "测试电影"
meta.type = MediaType.MOVIE
meta.year = "2024"
match_doubaninfo = Mock(return_value={"id": "200"})
douban_info = Mock(return_value={
"id": "200",
"title": "测试电影",
"type": "movie",
"year": "2024",
})
first_result = module._recognize_media_core(
meta=meta,
source="douban",
match_doubaninfo_func=match_doubaninfo,
douban_info_func=douban_info,
)
second_result = module._recognize_media_core(
meta=meta,
source="douban",
match_doubaninfo_func=match_doubaninfo,
douban_info_func=douban_info,
)
assert first_result.douban_id == "200"
assert second_result.douban_id == "200"
assert match_doubaninfo.call_count == 2
assert douban_info.call_count == 2
def test_async_douban_recognize_does_not_keep_dedicated_mapping_cache():
"""异步豆瓣识别也应每次执行匹配，不使用专用标题映射缓存。"""
module = DoubanModule()
meta = MetaBase("测试剧集")
meta.name = "测试剧集"
meta.type = MediaType.TV
meta.year = "2024"
match_doubaninfo = AsyncMock(return_value={"id": "201"})
douban_info = AsyncMock(return_value={
"id": "201",
"title": "测试剧集",
"type": "tv",
"year": "2024",
})
async def recognize_twice():
"""连续执行两次异步豆瓣识别。"""
first_result = await module._async_recognize_media_core(
meta=meta,
source="douban",
async_match_doubaninfo_func=match_doubaninfo,
async_douban_info_func=douban_info,
)
second_result = await module._async_recognize_media_core(
meta=meta,
source="douban",
async_match_doubaninfo_func=match_doubaninfo,
async_douban_info_func=douban_info,
)
return first_result, second_result
first_result, second_result = asyncio.run(recognize_twice())
assert first_result.douban_id == "201"
assert second_result.douban_id == "201"
assert match_doubaninfo.await_count == 2
assert douban_info.await_count == 2

79
tests/test_scheduler_cache_expiry.py Normal file
View File

@@ -0,0 +1,79 @@
import threading
from unittest.mock import Mock
from app import scheduler as scheduler_module
from app.scheduler import Scheduler
class _BackgroundSchedulerStub:
"""记录系统定时任务注册结果的调度器替身。"""
def __init__(self):
"""初始化任务记录。"""
self.jobs = []
self.started = False
def add_job(self, func, trigger, **kwargs):
"""记录一次任务注册。"""
self.jobs.append({"func": func, "trigger": trigger, **kwargs})
def start(self):
"""记录调度器已启动。"""
self.started = True
def test_meta_cache_expire_does_not_schedule_bulk_cache_clear(monkeypatch):
"""单条缓存 TTL 不应再被用于注册整批缓存清理任务。"""
background_scheduler = _BackgroundSchedulerStub()
generic_chain = Mock()
for name in [
"MediaServerChain",
"RecommendChain",
"SchedulerChain",
"SiteChain",
"SubscribeChain",
"TransferChain",
"WallpaperHelper",
"WorkflowChain",
"PluginManager",
]:
monkeypatch.setattr(scheduler_module, name, lambda: generic_chain)
monkeypatch.setattr(
scheduler_module.ServiceConfigHelper,
"get_mediaserver_configs",
lambda: [],
)
monkeypatch.setattr(
scheduler_module,
"BackgroundScheduler",
lambda **kwargs: background_scheduler,
)
monkeypatch.setattr(Scheduler, "stop", lambda self: None)
monkeypatch.setattr(Scheduler, "init_workflow_jobs", lambda self: None)
monkeypatch.setattr(Scheduler, "init_agent_task_jobs", lambda self: None)
monkeypatch.setattr(Scheduler, "init_plugin_jobs", lambda self: None)
monkeypatch.setattr(scheduler_module.settings, "DEV", False)
monkeypatch.setattr(scheduler_module.settings, "COOKIECLOUD_INTERVAL", 0)
monkeypatch.setattr(scheduler_module.settings, "SUBSCRIBE_SEARCH", False)
monkeypatch.setattr(scheduler_module.settings, "SUBSCRIBE_MODE", "rss")
monkeypatch.setattr(scheduler_module.settings, "SUBSCRIBE_RSS_INTERVAL", 30)
monkeypatch.setattr(scheduler_module.settings, "SITEDATA_REFRESH_INTERVAL", 0)
monkeypatch.setattr(scheduler_module.settings, "MEMORY_GC_INTERVAL", 0)
monkeypatch.setattr(scheduler_module.settings, "AI_AGENT_ENABLE", False)
monkeypatch.setattr(scheduler_module.settings, "DATA_CLEANUP_ENABLE", False)
monkeypatch.setattr(scheduler_module.settings, "USAGE_STATISTIC_SHARE", False)
scheduler = object.__new__(Scheduler)
scheduler._scheduler = None
scheduler._event = threading.Event()
scheduler._lock = threading.RLock()
scheduler._jobs = {}
scheduler._auth_count = 0
scheduler._auth_message = False
scheduler.init()
scheduled_job_ids = {job["id"] for job in background_scheduler.jobs}
assert "clear_cache" not in scheduled_job_ids
assert "clear_cache" in scheduler._jobs
assert background_scheduler.started is True

175
tests/test_tmdb_cache_management.py
View File

@@ -1,9 +1,11 @@
import asyncio
import inspect
import pickle
from unittest.mock import Mock
from app.api.endpoints import tmdb as tmdb_endpoint
from app.db.user_oper import get_current_active_superuser_async
from app.modules.themoviedb import tmdb_cache as tmdb_cache_module
from app.modules.themoviedb.tmdb_cache import TmdbCache
from app.schemas.types import MediaType, SystemConfigKey
@@ -27,19 +29,83 @@ class _MemoryCacheStub:
"""删除指定缓存条目。"""
self.data.pop(key, None)
def set(self, key: str, value, ttl=None):
"""写入指定缓存条目。"""
self.data[key] = value
def clear(self):
"""清空全部缓存条目。"""
self.data.clear()
class _FileCacheStub:
"""提供 TMDB 持久化测试所需的统一文件缓存替身。"""
def __init__(self, content: bytes = None):
"""使用预置序列化内容初始化文件缓存。"""
self.content = content
self.set_calls = []
self.delete_calls = []
def get(self, key: str, region: str):
"""读取预置缓存内容。"""
return self.content
def set(self, key: str, value: bytes, region: str):
"""记录统一文件缓存写入。"""
self.content = value
self.set_calls.append((key, region))
def delete(self, key: str, region: str):
"""记录统一文件缓存删除。"""
self.content = None
self.delete_calls.append((key, region))
class _TTLCacheStub(_MemoryCacheStub):
"""记录每条数据恢复时剩余 TTL 的内存缓存替身。"""
def __init__(self):
"""初始化空缓存和 TTL 记录。"""
super().__init__({})
self.ttls = {}
@staticmethod
def is_redis() -> bool:
"""测试替身固定使用非 Redis 后端。"""
return False
def set(self, key: str, value, ttl=None):
"""写入缓存并记录本次设置的 TTL。"""
super().set(key, value, ttl=ttl)
self.ttls[key] = ttl
def _build_tmdb_cache(data: dict) -> TmdbCache:
"""构造绕过单例初始化的 TMDB 缓存测试实例。"""
cache = object.__new__(TmdbCache)
cache._cache = _MemoryCacheStub(data)
cache._expires_at = {key: float("inf") for key in data}
cache._dirty = False
cache._file_cache = None
cache._legacy_file_cache = None
cache._legacy_cache_found = False
cache.save = lambda force=False: None
return cache
def _build_initialized_tmdb_cache(monkeypatch, file_cache: _FileCacheStub,
runtime_cache: _TTLCacheStub,
now: float = 1000) -> TmdbCache:
"""使用可控时间和缓存替身初始化完整 TMDB 缓存实例。"""
monkeypatch.setattr(tmdb_cache_module, "time", lambda: now)
monkeypatch.setattr(tmdb_cache_module, "TTLCache", lambda **kwargs: runtime_cache)
monkeypatch.setattr(tmdb_cache_module, "FileCache", lambda **kwargs: file_cache)
cache = object.__new__(TmdbCache)
cache.__init__()
return cache
def test_tmdb_cache_management_endpoints_require_superuser():
"""识别缓存管理接口必须仅允许超级管理员访问。"""
endpoints = [
@@ -92,6 +158,115 @@ def test_tmdb_cache_delete_and_clear_persist_immediately(monkeypatch):
assert saved_forces == [True, True]
def test_tmdb_cache_restores_only_unexpired_persisted_items(monkeypatch):
"""TMDB 持久化恢复应保留每条数据原有期限并跳过已过期条目。"""
payload = {
"version": tmdb_cache_module.PERSISTENCE_VERSION,
"items": {
"fresh": {
"value": {"id": 1, "title": "有效"},
"expires_at": 1030,
},
"expired": {
"value": {"id": 2, "title": "过期"},
"expires_at": 999,
},
},
}
file_cache = _FileCacheStub(pickle.dumps(payload))
runtime_cache = _TTLCacheStub()
cache = _build_initialized_tmdb_cache(
monkeypatch=monkeypatch,
file_cache=file_cache,
runtime_cache=runtime_cache,
)
assert runtime_cache.data == {"fresh": {"id": 1, "title": "有效"}}
assert runtime_cache.ttls == {"fresh": 30}
assert cache._expires_at == {"fresh": 1030}
assert cache._dirty is True
def test_tmdb_cache_persists_individual_expiration_with_file_cache(monkeypatch):
"""TMDB 持久化应通过统一文件缓存保存每条数据的独立过期时间。"""
file_cache = _FileCacheStub()
runtime_cache = _TTLCacheStub()
cache = _build_initialized_tmdb_cache(
monkeypatch=monkeypatch,
file_cache=file_cache,
runtime_cache=runtime_cache,
)
runtime_cache.data = {
"recognized": {"id": 1, "title": "有效"},
"unrecognized": {"id": 0},
}
cache._expires_at = {
"recognized": 1060,
"unrecognized": 1070,
}
cache._dirty = True
cache.save()
payload = pickle.loads(file_cache.content)
assert file_cache.set_calls == [(
tmdb_cache_module.PERSISTENCE_KEY,
tmdb_cache_module.PERSISTENCE_REGION,
)]
assert payload == {
"version": tmdb_cache_module.PERSISTENCE_VERSION,
"items": {
"recognized": {
"value": {"id": 1, "title": "有效"},
"expires_at": 1060,
},
},
}
def test_tmdb_cache_migrates_legacy_file_to_global_file_cache(monkeypatch):
"""旧 TMDB 缓存应迁移到全局文件缓存并删除旧文件。"""
primary_cache = _FileCacheStub()
legacy_cache = _FileCacheStub(pickle.dumps({
"legacy": {"id": 1, "title": "旧缓存"},
}))
file_caches = iter([primary_cache, legacy_cache])
file_cache_calls = []
def build_file_cache(**kwargs):
"""记录全局文件缓存构造参数并返回对应替身。"""
file_cache_calls.append(kwargs)
return next(file_caches)
runtime_cache = _TTLCacheStub()
monkeypatch.setattr(tmdb_cache_module, "time", lambda: 1000)
monkeypatch.setattr(
tmdb_cache_module,
"TTLCache",
lambda **kwargs: runtime_cache,
)
monkeypatch.setattr(tmdb_cache_module, "FileCache", build_file_cache)
cache = object.__new__(TmdbCache)
cache.__init__()
cache.save()
assert file_cache_calls == [
{"base": tmdb_cache_module.settings.CACHE_PATH, "ttl": cache.ttl},
{"base": tmdb_cache_module.settings.TEMP_PATH.parent, "ttl": cache.ttl},
]
assert runtime_cache.data == {"legacy": {"id": 1, "title": "旧缓存"}}
assert primary_cache.set_calls == [(
tmdb_cache_module.PERSISTENCE_KEY,
tmdb_cache_module.PERSISTENCE_REGION,
)]
assert legacy_cache.delete_calls == [(
cache.region,
tmdb_cache_module.settings.TEMP_PATH.name,
)]
def test_tmdb_cache_endpoint_returns_management_statistics(monkeypatch):
"""查询接口应返回识别成功和失败条目的统计。"""
cache = _build_tmdb_cache({