fix(tmdb): 空结果缓存使用独立短 TTL，故障恢复后快速自愈，关闭 #6332

- cached 装饰器新增 empty_ttl/empty_if，空结果按独立短 TTL 写入
- TMDB request/async_request/discover 空结果快照 30 分钟过期
- TmdbCache 未识别负缓存(id=0)同步改用 30 分钟短 TTL，正缓存不变

app/modules/themoviedb/tmdb_cache.py

@@ -3,7 +3,7 @@ import traceback
 from math import ceil
 from threading import RLock
 from time import time
-from typing import Any
+from typing import Any, Optional
 
 from app.runtime.cache import FileCache, TTLCache
 from app.runtime.config import settings
@@ -11,6 +11,7 @@ from app.domain.meta.metabase import MetaBase
 from app.runtime.log import logger
 from app.schemas.types import MediaSource, MediaType
 from app.foundation.singleton import WeakSingleton
+from app.modules.themoviedb.tmdbv3api.tmdb import EMPTY_RESULT_CACHE_TTL
 
 lock = RLock()
 PERSISTENCE_VERSION = 1
@@ -94,11 +95,12 @@ class TmdbCache(metaclass=WeakSingleton):
         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)
+    def _set(self, key: str, value: dict, 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() + self.ttl
+            self._expires_at[key] = time() + ttl
             self._dirty = True
 
     def clear(self):
@@ -257,7 +259,10 @@ class TmdbCache(metaclass=WeakSingleton):
         elif info is not None:
             # None时不缓存，此时代表网络错误，允许重复请求
             with lock:
-                self._set(key, {"id": 0})
+                # 负识别缓存使用独立的短 TTL：故障期间「合法 JSON 但结果为空」会被
+                # 记为未识别，若按完整有效期固化，故障自愈后同名仍会被判无法识别；
+                # 短过期让恢复后可重新识别，真不存在的条目过期后重新确认一次即可
+                self._set(key, {"id": 0}, ttl=EMPTY_RESULT_CACHE_TTL)
 
     def save(self, force: bool = False) -> None:
         """

app/modules/themoviedb/tmdbv3api/objs/discover.py

@@ -1,5 +1,5 @@
 from app.runtime.cache import cached
-from ..tmdb import TMDb
+from ..tmdb import TMDb, EMPTY_RESULT_CACHE_TTL
 
 try:
     from urllib import urlencode
@@ -13,7 +13,7 @@ class Discover(TMDb):
         "tv": "/discover/tv"
     }
 
-    @cached(maxsize=1, ttl=43200)
+    @cached(maxsize=1, ttl=43200, empty_ttl=EMPTY_RESULT_CACHE_TTL)
     def discover_movies(self, params_tuple):
         """
         Discover movies by different types of data like average rating, number of votes, genres and certifications.
@@ -23,7 +23,7 @@ class Discover(TMDb):
         params = dict(params_tuple)
         return self._request_obj(self._urls["movies"], urlencode(params), key="results", call_cached=False)
 
-    @cached(maxsize=1, ttl=43200)
+    @cached(maxsize=1, ttl=43200, empty_ttl=EMPTY_RESULT_CACHE_TTL)
     def discover_tv_shows(self, params_tuple):
         """
         Discover TV shows by different types of data like average rating, number of votes, genres,
@@ -33,7 +33,7 @@ class Discover(TMDb):
         """
         return self._request_obj(self._urls["tv"], urlencode(params_tuple), key="results", call_cached=False)
 
-    @cached(maxsize=1, ttl=43200)
+    @cached(maxsize=1, ttl=43200, empty_ttl=EMPTY_RESULT_CACHE_TTL)
     async def async_discover_movies(self, params_tuple):
         """
         Discover movies by different types of data like average rating, number of votes, genres and certifications.（异步版本）
@@ -43,7 +43,7 @@ class Discover(TMDb):
         params = dict(params_tuple)
         return await self._async_request_obj(self._urls["movies"], urlencode(params), key="results", call_cached=False)
 
-    @cached(maxsize=1, ttl=43200)
+    @cached(maxsize=1, ttl=43200, empty_ttl=EMPTY_RESULT_CACHE_TTL)
     async def async_discover_tv_shows(self, params_tuple):
         """
         Discover TV shows by different types of data like average rating, number of votes, genres,

app/modules/themoviedb/tmdbv3api/tmdb.py

@@ -22,6 +22,11 @@ logger = logging.getLogger(__name__)
 # 故取区间内的经验值。
 RETRY_BACKOFF_SECONDS = 2
 
+# 空结果缓存的独立过期时间（秒）。TMDB 代理故障期间「合法 JSON 但 results 为空」的
+# 响应会随默认 TTL（可达数十小时）固化，故障自愈后同 key 仍持续命中空结果；空结果
+# 改用 30 分钟短 TTL，既能拦住故障窗口内的重复回源，又能在故障恢复后较快自然失效。
+EMPTY_RESULT_CACHE_TTL = 30 * 60
+
 
 def _is_business_failure_snapshot(snapshot) -> bool:
     """
@@ -37,6 +42,23 @@ def _is_business_failure_snapshot(snapshot) -> bool:
     return isinstance(json_data, dict) and json_data.get("success") is False
 
 
+def _is_empty_result_snapshot(snapshot) -> bool:
+    """
+    判断响应快照是否为空结果（列表/搜索类接口的 results 为空列表）。
+
+    这类快照结构合法但无业务内容，常由代理瞬时故障产生；不能靠 skip_none/skip_empty
+    识别（快照本身是非空字典），需单独谓词判定后按 EMPTY_RESULT_CACHE_TTL 短 TTL
+    缓存。详情类接口无 results 字段，不属于空结果。
+    """
+    if not isinstance(snapshot, dict):
+        return False
+    json_data = snapshot.get("json")
+    if not isinstance(json_data, dict):
+        return False
+    results = json_data.get("results")
+    return isinstance(results, list) and not results
+
+
 class TMDb(object):
     _RESPONSE_SNAPSHOT_MARKER = "__mp_tmdb_response_snapshot__"
 
@@ -157,7 +179,8 @@ class TMDb(object):
         self._wait_on_rate_limit = bool(wait_on_rate_limit)
 
     @cached(maxsize=settings.CONF.tmdb, ttl=settings.CONF.meta, skip_none=True,
-            skip_if=_is_business_failure_snapshot)
+            skip_if=_is_business_failure_snapshot,
+            empty_ttl=EMPTY_RESULT_CACHE_TTL, empty_if=_is_empty_result_snapshot)
     def request(self, method, url, data, json, **kwargs):
         req = self._request_once(method, url, data, json)
         if req is None and method == "GET" and self._owns_session:
@@ -182,7 +205,8 @@ class TMDb(object):
         return self._req.post_res(url, data=data, json=json)
 
     @cached(maxsize=settings.CONF.tmdb, ttl=settings.CONF.meta, skip_none=True,
-            skip_if=_is_business_failure_snapshot)
+            skip_if=_is_business_failure_snapshot,
+            empty_ttl=EMPTY_RESULT_CACHE_TTL, empty_if=_is_empty_result_snapshot)
     async def async_request(self, method, url, data, json, **kwargs):
         req = await self._async_request_once(method, url, data, json)
         if req is None:

app/runtime/cache.py

@@ -757,7 +757,8 @@ def AsyncCache(cache_type: Literal['ttl', 'lru'] = 'ttl',
 
 def cached(region: Optional[str] = None, maxsize: Optional[int] = 1024, ttl: Optional[int] = None,
            skip_none: Optional[bool] = True, skip_empty: Optional[bool] = False, shared_key: Optional[str] = None,
-           skip_if: Optional[Callable[[Any], bool]] = None):
+           skip_if: Optional[Callable[[Any], bool]] = None,
+           empty_ttl: Optional[int] = None, empty_if: Optional[Callable[[Any], bool]] = None):
     """
     自定义缓存装饰器，支持配置缓存区域的 maxsize 和每个 key 的 ttl
 
@@ -770,6 +771,12 @@ def cached(region: Optional[str] = None, maxsize: Optional[int] = 1024, ttl: Opt
     :param skip_if: 按返回值判断是否跳过缓存的谓词，返回真值时不缓存；用于
         「结构合法但业务失败」的返回值（如 TMDB 的 success=false 响应），
         这类值无法用 skip_none/skip_empty 表达
+    :param empty_ttl: 空结果的独立存活时间，单位秒；判定为空的结果改用该 TTL 写入，
+        使其比默认 ttl 更快过期（如故障期间产生的空响应），未传入时空结果沿用默认 ttl；
+        仅在 ttl 模式下生效，LRU 模式无过期概念
+    :param empty_if: 判断返回值是否为空结果的谓词，返回真值时按 empty_ttl 写入；
+        未传入时按假值判断（None, [], {}, "", set() 等视为空）；用于「结构合法但
+        内容为空」的返回值（如 TMDB 搜索结果快照中 results 为空列表）
     :return: 装饰器函数
     """
 
@@ -799,6 +806,21 @@ def cached(region: Optional[str] = None, maxsize: Optional[int] = 1024, ttl: Opt
                 return False
             return True
 
+        def get_cache_ttl(value: Any) -> Optional[int]:
+            """
+            返回写入该返回值时应使用的 TTL，空结果改用独立的短 TTL（empty_ttl）
+
+            :param value: 待写入缓存的返回值
+            :return: 实际使用的 TTL，单位秒
+            """
+            if empty_ttl is None:
+                return ttl
+            if value is None:
+                return empty_ttl
+            if empty_if is not None:
+                return empty_ttl if empty_if(value) else ttl
+            return empty_ttl if not value else ttl
+
         def is_valid_cache_value(_cache_key: str, _cached_value: Any, _cache_region: str) -> bool:
             """
             判断指定的值是否为一个有效的缓存值
@@ -893,8 +915,9 @@ def cached(region: Optional[str] = None, maxsize: Optional[int] = 1024, ttl: Opt
                 # 判断是否需要缓存
                 if not should_cache(result):
                     return result
-                # 设置缓存（如果有传入的 maxsize 和 ttl，则覆盖默认值）
-                await cache_backend.set(cache_key, result, ttl=ttl, maxsize=maxsize, region=cache_region)
+                # 设置缓存（如果有传入的 maxsize 和 ttl，则覆盖默认值；空结果使用独立的短 TTL）
+                await cache_backend.set(cache_key, result, ttl=get_cache_ttl(result), maxsize=maxsize,
+                                        region=cache_region)
                 return result
 
             async def cache_clear():
@@ -944,8 +967,8 @@ def cached(region: Optional[str] = None, maxsize: Optional[int] = 1024, ttl: Opt
                 # 判断是否需要缓存
                 if not should_cache(result):
                     return result
-                # 设置缓存（如果有传入的 maxsize 和 ttl，则覆盖默认值）
-                cache_backend.set(cache_key, result, ttl=ttl, maxsize=maxsize, region=cache_region)
+                # 设置缓存（如果有传入的 maxsize 和 ttl，则覆盖默认值；空结果使用独立的短 TTL）
+                cache_backend.set(cache_key, result, ttl=get_cache_ttl(result), maxsize=maxsize, region=cache_region)
                 return result
 
             def cache_clear():

tests/test_cache_system.py

@@ -305,6 +305,106 @@ def test_cached_zero_ttl_does_not_cache_async_result():
     assert asyncio.run(run_test()) == (1, 2)
 
 
+def test_cached_empty_ttl_expires_empty_result_sooner_sync():
+    """
+    同步 cached 的空结果应按 empty_ttl 独立过期，不受默认 ttl 影响。
+    """
+    calls = 0
+
+    @cached(region="sync_empty_ttl", ttl=600, empty_ttl=10,
+            empty_if=lambda value: not value.get("results"))
+    def load_value():
+        nonlocal calls
+        calls += 1
+        return {"results": []}
+
+    assert load_value() == {"results": []}
+    assert load_value() == {"results": []}
+    assert calls == 1
+
+    region_cache = MemoryBackend._region_caches[MemoryBackend.get_region("sync_empty_ttl")]
+    started_at = region_cache.timer()
+    region_cache.expire(time=started_at + 11)
+
+    assert load_value() == {"results": []}
+    assert calls == 2
+
+
+def test_cached_empty_ttl_keeps_default_ttl_for_non_empty_result():
+    """
+    非空结果应继续使用默认 ttl，不被 empty_ttl 缩短。
+    """
+    calls = 0
+
+    @cached(region="sync_empty_ttl_nonempty", ttl=600, empty_ttl=10,
+            empty_if=lambda value: not value.get("results"))
+    def load_value():
+        nonlocal calls
+        calls += 1
+        return {"results": [1]}
+
+    assert load_value() == {"results": [1]}
+
+    region_cache = MemoryBackend._region_caches[MemoryBackend.get_region("sync_empty_ttl_nonempty")]
+    started_at = region_cache.timer()
+    region_cache.expire(time=started_at + 11)
+
+    assert load_value() == {"results": [1]}
+    assert calls == 1
+
+
+def test_cached_empty_ttl_without_empty_if_uses_falsy_check():
+    """
+    未提供 empty_if 时按假值判断空结果，空列表按 empty_ttl 独立过期。
+    """
+    calls = 0
+
+    @cached(region="sync_empty_ttl_falsy", ttl=600, empty_ttl=10)
+    def load_value():
+        nonlocal calls
+        calls += 1
+        return []
+
+    assert load_value() == []
+    assert load_value() == []
+    assert calls == 1
+
+    region_cache = MemoryBackend._region_caches[MemoryBackend.get_region("sync_empty_ttl_falsy")]
+    started_at = region_cache.timer()
+    region_cache.expire(time=started_at + 11)
+
+    assert load_value() == []
+    assert calls == 2
+
+
+def test_cached_empty_ttl_expires_empty_result_sooner_async():
+    """
+    异步 cached 的空结果应与同步路径一致，按 empty_ttl 独立过期。
+    """
+    calls = 0
+
+    @cached(region="async_empty_ttl", ttl=600, empty_ttl=10,
+            empty_if=lambda value: not value.get("results"))
+    async def load_value():
+        nonlocal calls
+        calls += 1
+        return {"results": []}
+
+    async def run_first_round():
+        assert await load_value() == {"results": []}
+        assert await load_value() == {"results": []}
+
+    asyncio.run(run_first_round())
+    assert calls == 1
+
+    region_cache = MemoryBackend._region_caches[MemoryBackend.get_region("async_empty_ttl")]
+    started_at = region_cache.timer()
+    region_cache.expire(time=started_at + 11)
+
+    assert asyncio.run(load_value()) == {"results": []}
+    assert calls == 2
+
+
 def test_memory_backend_global_clear_is_safe_during_region_creation():
     """
     全局清理与新 region 创建应由同一把锁串行化，不能并发修改注册表。

tests/test_tmdb_cache_type_guard.py

@@ -1,7 +1,9 @@
+from time import time
 from types import SimpleNamespace
 
 from app.runtime.config import settings
 from app.modules.themoviedb.tmdb_cache import TmdbCache
+from app.modules.themoviedb.tmdbv3api.tmdb import EMPTY_RESULT_CACHE_TTL
 from app.schemas.types import MediaSource, MediaType
 
 
@@ -120,3 +122,29 @@ def test_update_caches_tv_result_for_movie_meta():
     stored = cache._cache.get(_key("电影", begin_season=None))
     assert stored["type"] == MediaType.TV
     assert stored["title"] == "某剧"
+
+
+def test_update_negative_cache_uses_short_ttl():
+    """未识别负缓存应用独立短 TTL，故障自愈后同名可较快重新识别。"""
+    meta = _build_meta(MediaType.TV)
+    cache = _build_cache()
+
+    cache.update(meta, {})
+
+    key = _key("电视剧")
+    assert cache._cache.data[key] == {"id": 0}
+    assert cache._cache.ttls[key] == EMPTY_RESULT_CACHE_TTL
+    # 持久化用的过期时间也应随短 TTL 计算，不能沿用默认有效期
+    assert cache._expires_at[key] <= time() + EMPTY_RESULT_CACHE_TTL + 5
+
+
+def test_update_positive_cache_keeps_default_ttl():
+    """识别成功的正缓存不受负缓存短 TTL 影响，仍用默认有效期。"""
+    meta = _build_meta(MediaType.TV)
+    cache = _build_cache()
+
+    cache.update(meta, {"id": 329809, "media_type": MediaType.TV,
+                        "name": "某剧", "first_air_date": "2022-01-01"})
+
+    key = _key("电视剧")
+    assert cache._cache.ttls[key] == cache.ttl

tests/test_tmdb_empty_result_cache.py

@@ -0,0 +1,113 @@
+"""
+TMDB request 层空结果快照短 TTL 缓存测试。
+
+TMDB 代理故障期间「合法 JSON 但 results 为空」的响应若随默认 TTL（可达数十小时）
+固化，故障自愈后同名搜索仍持续命中空结果。空结果快照必须仍入缓存（拦住故障窗口
+内的重复回源），但改用独立的 30 分钟短 TTL，过期后自然恢复回源。
+"""
+import asyncio
+from unittest.mock import patch
+
+from app.modules.themoviedb.tmdbv3api.tmdb import (
+    EMPTY_RESULT_CACHE_TTL,
+    TMDb,
+    _is_empty_result_snapshot,
+)
+from app.runtime.cache import MemoryBackend
+
+from tests.test_tmdb_response_cache import _FakeResponse
+
+EMPTY_PAYLOAD = {"page": 1, "results": [], "total_results": 0, "total_pages": 0}
+NOT_EMPTY_PAYLOAD = {"page": 1, "results": [{"id": 1}], "total_results": 1, "total_pages": 1}
+HEADERS = {"Content-Type": "application/json"}
+
+
+def _snapshot(payload: dict) -> dict:
+    """构造一个带快照标记的响应结构。"""
+    return {TMDb._RESPONSE_SNAPSHOT_MARKER: True, "headers": {}, "json": payload}
+
+
+def _request_region_cache():
+    """取出 TMDb.request 装饰器使用的内存缓存区实例。"""
+    return MemoryBackend._region_caches[MemoryBackend.get_region(TMDb.request.cache_region)]
+
+
+def _make_tmdb() -> TMDb:
+    """构造带测试 API Key 的 TMDb 客户端。"""
+    tmdb = TMDb()
+    tmdb.api_key = "test-key"
+    return tmdb
+
+
+def test_empty_result_cache_ttl_is_thirty_minutes():
+    """空结果缓存的独立过期时间应为 30 分钟。"""
+    assert EMPTY_RESULT_CACHE_TTL == 30 * 60
+
+
+def test_empty_result_snapshot_predicate():
+    """空结果谓词只认 results 为空列表的快照，详情与有结果的响应不算空。"""
+    assert _is_empty_result_snapshot(_snapshot(EMPTY_PAYLOAD))
+    assert not _is_empty_result_snapshot(_snapshot(NOT_EMPTY_PAYLOAD))
+    assert not _is_empty_result_snapshot(_snapshot({"id": 98865, "title": "Test"}))
+    # json 非字典或无 results 字段时不能误判为空结果
+    assert not _is_empty_result_snapshot(_snapshot("upstream error"))
+    assert not _is_empty_result_snapshot(None)
+
+
+def test_empty_result_is_cached_but_expires_with_short_ttl():
+    """空结果快照仍入缓存避免重复回源，但按短 TTL 过期后恢复回源。"""
+    tmdb = _make_tmdb()
+    url = "https://api.tmdb.test/empty-short-ttl"
+    fake = _FakeResponse(EMPTY_PAYLOAD, HEADERS)
+
+    with patch.object(TMDb, "_request_once", return_value=fake) as req:
+        tmdb.request("GET", url, None, None)
+        tmdb.request("GET", url, None, None)
+    assert req.call_count == 1
+
+    region_cache = _request_region_cache()
+    started_at = region_cache.timer()
+    region_cache.expire(time=started_at + EMPTY_RESULT_CACHE_TTL + 1)
+
+    with patch.object(TMDb, "_request_once", return_value=fake) as req:
+        tmdb.request("GET", url, None, None)
+    assert req.call_count == 1
+
+
+def test_non_empty_result_keeps_default_ttl():
+    """有结果的响应不受短 TTL 影响，过期点仍为默认元数据缓存 TTL 之后。"""
+    tmdb = _make_tmdb()
+    url = "https://api.tmdb.test/non-empty-default-ttl"
+    fake = _FakeResponse(NOT_EMPTY_PAYLOAD, HEADERS)
+
+    with patch.object(TMDb, "_request_once", return_value=fake) as req:
+        tmdb.request("GET", url, None, None)
+    assert req.call_count == 1
+
+    region_cache = _request_region_cache()
+    started_at = region_cache.timer()
+    # 推进到短 TTL 之后：非空结果不应在此刻过期
+    region_cache.expire(time=started_at + EMPTY_RESULT_CACHE_TTL + 1)
+
+    tmdb.request("GET", url, None, None)
+    assert req.call_count == 1
+
+
+def test_async_empty_result_is_cached_with_short_ttl():
+    """异步请求的空结果快照与同步路径一致，入缓存但按短 TTL 过期。"""
+    tmdb = _make_tmdb()
+    url = "https://api.tmdb.test/async-empty-short-ttl"
+    fake = _FakeResponse(EMPTY_PAYLOAD, HEADERS)
+
+    with patch.object(TMDb, "_async_request_once", return_value=fake) as req:
+        asyncio.run(tmdb.async_request("GET", url, None, None))
+        asyncio.run(tmdb.async_request("GET", url, None, None))
+    assert req.call_count == 1
+
+    region_cache = _request_region_cache()
+    started_at = region_cache.timer()
+    region_cache.expire(time=started_at + EMPTY_RESULT_CACHE_TTL + 1)
+
+    with patch.object(TMDb, "_async_request_once", return_value=fake) as req:
+        asyncio.run(tmdb.async_request("GET", url, None, None))
+    assert req.call_count == 1