mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-31 20:34:48 +08:00
6d4ce83837
严格检查全部宿主源码，清零类型错误预算并移除按模块类型豁免。 按原因、所属方法、错误码及语法指纹锁定必要兼容例外和站点类型桩， CI 阻断新增忽略、检查范围退化、本地桩替代真实接口及新的类型错误。 修复空值、异步和供应方类型边界，保持插件与 API 兼容合同， 并补充存储、订阅、通知、Agent 和工作流边界的回归覆盖。
356 lines
15 KiB
Python
356 lines
15 KiB
Python
"""
|
||
TMDB 媒体识别缓存、类型冲突校验与本地快照迁移。
|
||
"""
|
||
|
||
import pickle
|
||
import traceback
|
||
from math import ceil
|
||
from threading import RLock
|
||
from time import time
|
||
from typing import Any, Optional
|
||
|
||
from app.domain.meta.metabase import MetaBase
|
||
from app.foundation.singleton import WeakSingleton
|
||
from app.runtime.cache import FileCache, TTLCache
|
||
from app.runtime.log import logger
|
||
from app.runtime.settings import get_runtime_setting
|
||
from app.schemas.types import MediaSource, MediaType
|
||
|
||
lock = RLock()
|
||
# 包含独立过期时间的本地缓存快照版本。
|
||
PERSISTENCE_VERSION = 1
|
||
PERSISTENCE_REGION = "recognize"
|
||
PERSISTENCE_KEY = "tmdb"
|
||
|
||
|
||
class TmdbCache(metaclass=WeakSingleton):
|
||
"""
|
||
管理 TMDB 识别结果、短期未命中缓存及本地持久化。
|
||
|
||
缓存值记录媒体标识、类型、标题、年份和主图片路径。Redis 后端自行持久化；
|
||
本地后端保存未过期的有效识别结果，不保存未命中条目。
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
"""
|
||
初始化 TMDB 识别缓存并恢复未过期的持久化数据。
|
||
"""
|
||
self.maxsize = get_runtime_setting('CONF').tmdb
|
||
self.ttl = get_runtime_setting('CONF').meta
|
||
self.region = "__tmdb_cache__"
|
||
self._cache = TTLCache(region=self.region, maxsize=self.maxsize, ttl=self.ttl)
|
||
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():
|
||
self._file_cache = FileCache(base=get_runtime_setting('CACHE_PATH'), ttl=self.ttl)
|
||
self._legacy_file_cache = FileCache(base=get_runtime_setting('TEMP_PATH').parent, ttl=self.ttl)
|
||
self._restore()
|
||
|
||
def _restore(self) -> None:
|
||
"""
|
||
从统一文件缓存恢复仍在有效期内的 TMDB 识别数据。
|
||
"""
|
||
if self._file_cache is None or self._legacy_file_cache is None:
|
||
return
|
||
try:
|
||
content = self._file_cache.get(PERSISTENCE_KEY, region=PERSISTENCE_REGION)
|
||
if not content:
|
||
content = self._legacy_file_cache.get(
|
||
self.region,
|
||
region=get_runtime_setting('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[str, Any], ttl: Optional[int] = None) -> None:
|
||
"""
|
||
写入单条 TMDB 识别缓存并记录其独立过期时间，未指定 ttl 时用默认有效期。
|
||
"""
|
||
ttl = self.ttl if ttl is None else ttl
|
||
self._cache.set(key, value, ttl=ttl)
|
||
if not self._cache.is_redis():
|
||
self._expires_at[key] = time() + ttl
|
||
self._dirty = True
|
||
|
||
def clear(self) -> None:
|
||
"""
|
||
清空识别缓存和过期记录，并立即同步到本地持久化。
|
||
"""
|
||
with lock:
|
||
self._cache.clear()
|
||
self._expires_at.clear()
|
||
self._dirty = True
|
||
self.save(force=True)
|
||
|
||
def list_items(self) -> list[dict[str, Any]]:
|
||
"""
|
||
生成按缓存键排序的管理界面条目列表。
|
||
|
||
:return: 包含媒体标识、标题、年份、类型和图片地址的字典列表；无效类型标为 unknown
|
||
"""
|
||
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,
|
||
"tmdb_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 "",
|
||
"backdrop_path": value.get("backdrop_path") or "",
|
||
})
|
||
return sorted(cache_items, key=lambda item: item["key"])
|
||
|
||
@staticmethod
|
||
def __get_key(meta: MetaBase) -> str:
|
||
"""
|
||
按媒体类型、语言、TMDB 标识或名称、年份及起始季生成识别缓存键。
|
||
"""
|
||
media_id = meta.media_id if meta.media_source == MediaSource.TMDB else None
|
||
return f"[{meta.type.value if meta.type else '未知'}][{get_runtime_setting('TMDB_LOCALE')}]{media_id or meta.name}-{meta.year}-{meta.begin_season}"
|
||
|
||
@staticmethod
|
||
def __is_type_conflicted(meta: MetaBase, media_type: Any, tmdb_id: Any) -> bool:
|
||
"""
|
||
判断媒体类型是否与元数据声明的类型冲突。
|
||
|
||
只有「元数据判定为电视剧、结果却是电影」才算冲突。反向不算：名称识别在
|
||
电影分支查不到时会回退到电视剧查询，识别缓存正是用来记住这个纠正结果，
|
||
一律要求 key 与 value 类型一致会让这类条目每次都被丢弃、反复回源。而电视
|
||
剧分支恒定写入电视剧类型，`[电视剧]` 键下出现电影只可能来自 tmdbid 消歧
|
||
或共享识别回填的脏写，会让整季剧集被当成电影反复整理失败。
|
||
|
||
:param meta: 元数据
|
||
:param media_type: 待校验的媒体类型
|
||
:param tmdb_id: 对应的 TMDB ID，为空表示负缓存，不带类型信息
|
||
:return: 是否冲突
|
||
"""
|
||
if meta.type != MediaType.TV or not tmdb_id:
|
||
return False
|
||
if not isinstance(media_type, MediaType):
|
||
try:
|
||
media_type = MediaType(media_type)
|
||
except (TypeError, ValueError):
|
||
return False
|
||
return bool(media_type == MediaType.MOVIE)
|
||
|
||
def get(self, meta: MetaBase) -> dict[str, Any]:
|
||
"""
|
||
读取元数据对应的有效识别缓存，并剔除类型冲突条目。
|
||
|
||
:param meta: 待识别媒体元数据，用于生成与当前语言相关的缓存键
|
||
:return: 缓存条目，id 为 0 表示短期未命中；缺失、无效或冲突时返回空字典
|
||
"""
|
||
key = self.__get_key(meta)
|
||
|
||
with lock:
|
||
cache_data = self._cache.get(key)
|
||
if not cache_data and self._expires_at.pop(key, None) is not None:
|
||
self._dirty = True
|
||
if not isinstance(cache_data, dict):
|
||
return {}
|
||
if not cache_data:
|
||
return {}
|
||
if self.__is_type_conflicted(meta, cache_data.get("type"), cache_data.get("id")):
|
||
# 脏条目不丢弃就会被无限期沿用，正确的识别逻辑永远没有执行机会
|
||
logger.warn(f"识别缓存类型与元数据冲突，已丢弃并重新识别：{key} -> "
|
||
f"{cache_data.get('title')}({cache_data.get('type')})")
|
||
self._cache.delete(key)
|
||
self._expires_at.pop(key, None)
|
||
self._dirty = True
|
||
return {}
|
||
return cache_data
|
||
|
||
def delete(self, key: str) -> dict[str, Any]:
|
||
"""
|
||
删除指定识别缓存并立即保存本地持久化状态。
|
||
|
||
:param key: 管理列表提供的完整缓存键
|
||
:return: 被删除的条目；不存在时返回空字典
|
||
"""
|
||
with lock:
|
||
redis_data = self._cache.get(key)
|
||
if isinstance(redis_data, dict) and redis_data:
|
||
self._cache.delete(key)
|
||
self._expires_at.pop(key, None)
|
||
self._dirty = True
|
||
self.save(force=True)
|
||
return redis_data
|
||
return {}
|
||
|
||
def modify(self, key: str, title: str) -> dict[str, Any]:
|
||
"""
|
||
修改已有缓存标题，并重置该条目的默认有效期。
|
||
|
||
:param key: 管理列表提供的完整缓存键
|
||
:param title: 替换后的展示标题
|
||
:return: 更新后的条目；不存在时返回空字典
|
||
"""
|
||
with lock:
|
||
redis_data = self._cache.get(key)
|
||
if isinstance(redis_data, dict) and redis_data:
|
||
redis_data['title'] = title
|
||
self._set(key, redis_data)
|
||
return redis_data
|
||
return {}
|
||
|
||
def update(self, meta: MetaBase, info: Optional[dict[str, Any]]) -> None:
|
||
"""
|
||
从 TMDB 结果更新识别缓存，区分未命中和网络失败。
|
||
|
||
拒绝把电影结果写入已判定为电视剧的键；电影元数据回退识别为电视剧时允许缓存。
|
||
|
||
:param meta: 用于生成缓存键和检查媒体类型的元数据
|
||
:param info: TMDB 识别结果；空字典记录短期未命中，None 表示失败且不缓存
|
||
"""
|
||
key = self.__get_key(meta)
|
||
if info:
|
||
if self.__is_type_conflicted(meta, info.get("media_type"), info.get("id")):
|
||
# 拒绝写入而不是改写键：识别结果照常返回，只是不把矛盾条目留给下一次
|
||
logger.warn(f"识别结果类型与元数据冲突，不写入识别缓存：{key} -> "
|
||
f"{info.get('title')}({info.get('media_type')})")
|
||
return
|
||
cache_title = info.get("title") \
|
||
if info.get("media_type") == MediaType.MOVIE else info.get("name")
|
||
cache_year = info.get('release_date') \
|
||
if info.get("media_type") == MediaType.MOVIE else info.get('first_air_date')
|
||
if cache_year:
|
||
cache_year = cache_year[:4]
|
||
|
||
with lock:
|
||
cache_data = {
|
||
"id": info.get("id"),
|
||
"type": info.get("media_type"),
|
||
"year": cache_year,
|
||
"title": cache_title,
|
||
"poster_path": info.get("poster_path"),
|
||
"backdrop_path": info.get("backdrop_path")
|
||
}
|
||
self._set(key, cache_data)
|
||
|
||
elif info is not None:
|
||
# None时不缓存，此时代表网络错误，允许重复请求
|
||
with lock:
|
||
# 负识别缓存使用独立的短 TTL：故障期间「合法 JSON 但结果为空」会被
|
||
# 记为未识别，若按完整有效期固化，故障自愈后同名仍会被判无法识别；
|
||
# 短过期让恢复后可重新识别，真不存在的条目过期后重新确认一次即可
|
||
self._set(key, {"id": 0}, ttl=get_runtime_setting('EMPTY_RESULT_CACHE_TTL'))
|
||
|
||
def save(self, force: bool = False) -> None:
|
||
"""
|
||
把本地后端仍有效且有媒体标识的缓存写入统一文件区域。
|
||
|
||
Redis 后端直接返回；保存成功后移除已发现的旧版缓存。文件操作失败仅记录日志。
|
||
|
||
:param force: True 时即使未标记变更也执行保存
|
||
"""
|
||
if self._cache.is_redis():
|
||
return
|
||
if self._file_cache is None or self._legacy_file_cache is None:
|
||
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
|
||
|
||
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 not self._dirty:
|
||
return
|
||
|
||
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=get_runtime_setting('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) -> None:
|
||
"""
|
||
实例释放前保存非 Redis 缓存。
|
||
"""
|
||
self.save()
|