fix(anilist): persist Chinese titles and use official API

app/modules/anilist/anilist.py

@@ -2,12 +2,16 @@
 AniList GraphQL 客户端，统一查询计划、中文标题补充及同步异步缓存。
 """
 
+import json
 from dataclasses import dataclass
 from datetime import date
+from math import isfinite
+from time import time
 from typing import Any, Callable, Optional, Protocol, TypeVar, cast
 
 from app.adapters.network.http import AsyncRequestUtils, RequestUtils
-from app.runtime.cache import cached
+from app.foundation.url import UrlUtils
+from app.runtime.cache import AsyncFileCache, FileCache, async_cache_mutation, cache_mutation, cached
 from app.runtime.log import logger
 from app.runtime.settings import get_runtime_setting
 
@@ -88,17 +92,21 @@ class _AniListRequestPlan:
 
 class AniListApi:
     """
-    通过中文代理或官方 GraphQL 接口查询 AniList，并补充中文标题。
+    通过官方 GraphQL 接口查询 AniList，并补充本地缓存的中文标题。
 
-    代理失败后该实例改用官方接口；中文数据集在实例内缓存。
+    中文数据集每天更新，下载失败时保留旧数据并在五分钟后重试。
     """
 
-    _base_url = "https://trace.moe/anilist/"
     _official_url = "https://graphql.anilist.co"
     _translations_url = (
         "https://raw.githubusercontent.com/soruly/anilist-chinese/"
         "master/anilist-chinese.json"
     )
+    # 上游中文数据集每天更新，磁盘和内存沿用同一有效期（秒）。
+    _translations_ttl = 24 * 60 * 60
+    _translations_retry = 5 * 60
+    _translations_region = "anilist"
+    _translations_key = "anilist-chinese.json"
     _media_summary_fields = """
         id
         idMal
@@ -245,8 +253,16 @@ class AniListApi:
             proxies=get_runtime_setting('PROXY'),
             headers=headers,
         )
-        self._proxy_available = True
         self._translations: Optional[dict[object, dict[str, Any]]] = None
+        self._translations_expires_at = 0.0
+        self._file_cache = None
+        self._async_file_cache = None
+        try:
+            cache_path = get_runtime_setting('CACHE_PATH')
+            self._file_cache = FileCache(base=cache_path, local_only=True)
+            self._async_file_cache = AsyncFileCache(base=cache_path, local_only=True)
+        except OSError as err:
+            logger.warning(f"初始化 anilist-chinese 磁盘缓存失败：{err}")
 
     @staticmethod
     def _extract_response(response: Optional[_GraphQLResponse]) -> Optional[dict[str, Any]]:
@@ -295,43 +311,28 @@ class AniListApi:
         """
         请求 GraphQL 并合并中文标题数据。
 
-        优先使用中文代理；失败后在当前实例内禁用该代理并回退官方接口。
-
         :param query: GraphQL 查询文本
         :param variables: 本次查询的变量映射
         :return: 经中文字段补充的 data 映射；请求或 GraphQL 业务失败时为 None
         """
         plan = self._request_plan(query, variables)
         payload = {"query": plan.query, "variables": plan.variables}
-        if self._proxy_available:
-            response = self._request.post_res(self._base_url, json=payload)
-            result = self._extract_response(response)
-            if result is not None:
-                return self._project_translated(result, self._translation_map())
-            self._disable_proxy(response)
         response = self._request.post_res(self._official_url, json=payload)
         result = self._extract_response(response)
+        if not result:
+            return result
         return self._project_translated(result, self._translation_map())
 
     async def _async_invoke(self, query: str, variables: dict[str, Any]) -> Optional[dict[str, Any]]:
         """
         异步请求 GraphQL 并合并中文标题数据。
 
-        优先使用中文代理；失败后在当前实例内禁用该代理并回退官方接口。
-
         :param query: GraphQL 查询文本
         :param variables: 本次查询的变量映射
         :return: 经中文字段补充的 data 映射；请求或 GraphQL 业务失败时为 None
         """
         plan = self._request_plan(query, variables)
         payload = {"query": plan.query, "variables": plan.variables}
-        if self._proxy_available:
-            response = await self._async_request.post_res(self._base_url, json=payload)
-            result = self._extract_response(response)
-            if result is not None:
-                translations = await self._async_translation_map()
-                return self._project_translated(result, translations)
-            self._disable_proxy(response)
         response = await self._async_request.post_res(self._official_url, json=payload)
         result = self._extract_response(response)
         if not result:
@@ -339,19 +340,6 @@ class AniListApi:
         translations = await self._async_translation_map()
         return self._project_translated(result, translations)
 
-    def _disable_proxy(self, response: object) -> None:
-        """
-        标记中文代理不可用，避免当前进程持续请求已失效的上游。
-
-        :param response: 中文代理响应对象
-        """
-        self._proxy_available = False
-        status_code = getattr(response, "status_code", None)
-        logger.warning(
-            f"anilist-chinese 代理不可用（HTTP {status_code}），"
-            "改用 AniList 官方接口并合并中文数据集"
-        )
-
     @staticmethod
     def _build_translation_map(items: object) -> dict[object, dict[str, Any]]:
         """
@@ -365,34 +353,137 @@ class AniListApi:
         return {
             item.get("id"): item
             for item in items
-            if isinstance(item, dict) and item.get("id")
+            if isinstance(item, dict)
+            and isinstance(item.get("id"), (int, str)) and item.get("id")
+            and isinstance(item.get("title"), str)
+            and isinstance(item.get("synonyms", []), list)
+            and all(isinstance(alias, str) for alias in item.get("synonyms", []))
         }
 
+    def _translation_routes(self) -> list[tuple[str, bool]]:
+        """
+        返回中文数据集的下载地址及直连标志，GitHub 镜像优先且不叠加部署代理。
+
+        :return: 镜像和原地址的有序列表，原地址沿用 PROXY
+        """
+        mirror = str(get_runtime_setting('GITHUB_PROXY') or '').strip()
+        routes = [(self._translations_url, False)]
+        if mirror:
+            routes.insert(0, (f"{UrlUtils.standardize_base_url(mirror)}{self._translations_url}", True))
+        return routes
+
+    @classmethod
+    def _decode_translation_cache(cls, content: Optional[bytes]) -> tuple[dict[object, dict[str, Any]], float]:
+        """
+        校验磁盘快照并读取数据和绝对过期时间，损坏快照按未命中处理。
+        """
+        if content:
+            try:
+                payload = json.loads(content)
+                if isinstance(payload, dict):
+                    expires_at = payload.get("expires_at")
+                    if isinstance(expires_at, (int, float)) and isfinite(expires_at):
+                        return cls._build_translation_map(payload.get("items")), float(expires_at)
+            except (ValueError, TypeError) as err:
+                logger.warning(f"解析 anilist-chinese 磁盘缓存失败：{err}")
+        return {}, 0.0
+
+    def _translation_snapshot(self, items: object) -> Optional[bytes]:
+        """
+        接收有效下载并生成磁盘快照；无效载荷不得覆盖内存或旧快照。
+        """
+        translations = self._build_translation_map(items)
+        if not translations:
+            return None
+        self._translations = translations
+        self._translations_expires_at = time() + self._translations_ttl
+        return json.dumps({
+            "expires_at": self._translations_expires_at,
+            "items": list(translations.values()),
+        }, ensure_ascii=False).encode('utf-8')
+
+    def _reuse_translation_cache(self, translations: dict[object, dict[str, Any]], expires_at: float) -> bool:
+        """
+        恢复磁盘缓存；内存仍有效时优先复用，过期数据仅保留为下载失败的回退。
+        """
+        if self._translations is not None and time() < self._translations_expires_at:
+            return True
+        if translations and (not self._translations or expires_at > self._translations_expires_at):
+            self._translations = translations
+            self._translations_expires_at = expires_at
+            return time() < expires_at
+        return False
+
+    def _translation_fallback(self) -> dict[object, dict[str, Any]]:
+        """
+        下载失败时保留旧中文数据，短暂退避后允许重试，不把空结果永久缓存。
+        """
+        logger.warning("加载 anilist-chinese 中文数据集失败，暂用已有数据，五分钟后重试")
+        self._translations = self._translations or {}
+        self._translations_expires_at = time() + self._translations_retry
+        return self._translations
+
     def _translation_map(self) -> dict[object, dict[str, Any]]:
         """
-        同步加载并复用 anilist-chinese 中文标题数据。
+        同步复用中文数据快照，仅在缺失或过期时下载；并发调用共用缓存写锁。
 
         :return: 中文标题数据索引
         """
-        if self._translations is None:
-            items = self._request.get_json(self._translations_url)
-            self._translations = self._build_translation_map(items)
-            if not self._translations:
-                logger.warning("加载 anilist-chinese 中文数据集失败")
-        return self._translations
+        with cache_mutation("anilist-translations"):
+            if self._reuse_translation_cache({}, 0.0):
+                return self._translations or {}
+            content = None
+            try:
+                if self._file_cache is not None:
+                    content = self._file_cache.get(self._translations_key, region=self._translations_region)
+            except OSError as err:
+                logger.warning(f"读取 anilist-chinese 磁盘缓存失败：{err}")
+            if self._reuse_translation_cache(*self._decode_translation_cache(content)):
+                return self._translations or {}
+            for url, direct in self._translation_routes():
+                request = RequestUtils(ua=get_runtime_setting('NORMAL_USER_AGENT')) if direct else self._request
+                snapshot = self._translation_snapshot(request.get_json(url))
+                if snapshot is None:
+                    continue
+                try:
+                    if self._file_cache is not None:
+                        self._file_cache.set(self._translations_key, snapshot, region=self._translations_region)
+                except OSError as err:
+                    logger.warning(f"写入 anilist-chinese 磁盘缓存失败：{err}")
+                return self._translations or {}
+            return self._translation_fallback()
 
     async def _async_translation_map(self) -> dict[object, dict[str, Any]]:
         """
-        异步加载并复用 anilist-chinese 中文标题数据。
+        异步复用同一磁盘快照和写锁，文件读写不阻塞事件循环。
 
         :return: 中文标题数据索引
         """
-        if self._translations is None:
-            items = await self._async_request.get_json(self._translations_url)
-            self._translations = self._build_translation_map(items)
-            if not self._translations:
-                logger.warning("加载 anilist-chinese 中文数据集失败")
-        return self._translations
+        async with async_cache_mutation("anilist-translations"):
+            if self._reuse_translation_cache({}, 0.0):
+                return self._translations or {}
+            content = None
+            try:
+                if self._async_file_cache is not None:
+                    content = await self._async_file_cache.get(self._translations_key, region=self._translations_region)
+            except OSError as err:
+                logger.warning(f"读取 anilist-chinese 磁盘缓存失败：{err}")
+            if self._reuse_translation_cache(*self._decode_translation_cache(content)):
+                return self._translations or {}
+            for url, direct in self._translation_routes():
+                request = AsyncRequestUtils(ua=get_runtime_setting('NORMAL_USER_AGENT')) if direct else self._async_request
+                snapshot = self._translation_snapshot(await request.get_json(url))
+                if snapshot is None:
+                    continue
+                try:
+                    if self._async_file_cache is not None:
+                        await self._async_file_cache.set(
+                            self._translations_key, snapshot, region=self._translations_region,
+                        )
+                except OSError as err:
+                    logger.warning(f"写入 anilist-chinese 磁盘缓存失败：{err}")
+                return self._translations or {}
+            return self._translation_fallback()
 
     @classmethod
     def _inject_chinese(cls, value: _Result, translations: dict[object, dict[str, Any]]) -> _Result:
@@ -567,7 +658,7 @@ class AniListApi:
 
     def _medias_by_ids(self, media_ids: list[int]) -> list[dict[str, Any]]:
         """
-        通过根级 Page.media 批量查询媒体，使中文代理能够注入标题。
+        通过根级 Page.media 批量查询媒体并补充中文标题。
 
         :param media_ids: AniList 媒体 ID 列表
         :return: 按输入顺序排列的媒体列表
@@ -583,7 +674,7 @@ class AniListApi:
 
     async def _async_medias_by_ids(self, media_ids: list[int]) -> list[dict[str, Any]]:
         """
-        异步通过根级 Page.media 批量查询媒体，使中文代理能够注入标题。
+        异步通过根级 Page.media 批量查询媒体并补充中文标题。
 
         :param media_ids: AniList 媒体 ID 列表
         :return: 按输入顺序排列的媒体列表

docs/architecture-overview.md

@@ -817,7 +817,7 @@ flowchart LR
 | 指标 | 当前值 |
 |---|---:|
 | Python 模块 | 1144 |
-| 内部导入边 | 9,887 |
+| 内部导入边 | 9,889 |
 | 非平凡 SCC | 1（29 个模块的 TMDB vendor 精确 containment） |
 | Application / Chain 具体 Adapter 直连 | 0 / 0 |
 | Direct egress | 51（精确 policy 准入，Application / Chain 均为 0） |

docs/testing.md

@@ -87,6 +87,13 @@ AppPush 的所有用户范围由 `tests/test_notification_routing_parity.py` 验
 
 ## 外部依赖：一律 mock，零真实网络
 
+AniList 直接访问官方 GraphQL 接口，不再探测 `trace.moe/anilist/`。中文数据集保存在
+`<CONFIG_DIR>/cache/anilist/anilist-chinese.json`，有效期 24 小时，同步、异步查询和重建客户端
+共用快照；配置 `GITHUB_PROXY` 时先直连镜像，失败再按 `PROXY` 访问原始 GitHub 地址。
+下载失败保留过期数据并在五分钟后重试，目录或文件读写失败时仍可使用已下载的内存数据。
+`tests/test_anilist_translations.py` 使用临时目录和下载替身验证重建客户端后的缓存复用、到期刷新、
+坏缓存、磁盘故障、镜像回退、同步异步并发去重及取消后释放锁；不据此推断真实代理耗时。
+
 高清杜比换域回归由 `tests/test_hddolby_spider.py`、`tests/test_hddolby_parser.py` 和
 `tests/test_site_connectivity.py` 验证：网页使用 `www.uhddolby.com` 时，搜索、用户资料和
 做种统计仍可按资源的 `api_url` 访问 `api.hddolby.com`，并保留站点代理和认证配置。

tests/fixtures/architecture/dependency-baseline.json

@@ -959,8 +959,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 9887,
-  "edge_sha256": "a6ce492847e6336c039524d2b4624ea88b8acdab99db9ab0825cdae99e703f66",
+  "edge_count": 9889,
+  "edge_sha256": "99db09bea0a36fa387eb552e2f97ca96ab6c289ebbab2305c7931f53fa0796f1",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -7301,6 +7301,8 @@
     "app.modules.anilist.anilist -> app.adapters",
     "app.modules.anilist.anilist -> app.adapters.network",
     "app.modules.anilist.anilist -> app.adapters.network.http",
+    "app.modules.anilist.anilist -> app.foundation",
+    "app.modules.anilist.anilist -> app.foundation.url",
     "app.modules.anilist.anilist -> app.runtime",
     "app.modules.anilist.anilist -> app.runtime.cache",
     "app.modules.anilist.anilist -> app.runtime.log",

tests/test_anilist_browse.py

@@ -1,17 +1,31 @@
 """
-验证 AniList 浏览筛选、中文代理回退及季节参数映射。
+验证 AniList 浏览筛选、官方查询的中文合并及季节参数映射。
 """
 
 import asyncio
 from datetime import date
+from pathlib import Path
 from unittest.mock import AsyncMock, Mock
 
+import pytest
+
 from app.api.endpoints import anilist as anilist_endpoint
 from app.domain.context import MediaInfo
 from app.modules.anilist import AniListModule
+from app.modules.anilist import anilist as anilist_client
 from app.modules.anilist.anilist import AniListApi
 
 
+@pytest.fixture(autouse=True)
+def isolated_translations(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    """
+    每个用例使用独立缓存目录及原始下载地址，防止中文数据跨用例串扰。
+    """
+    runtime_setting = anilist_client.get_runtime_setting
+    values = {"CACHE_PATH": tmp_path, "GITHUB_PROXY": None}
+    monkeypatch.setattr(anilist_client, "get_runtime_setting", lambda name: values[name] if name in values else runtime_setting(name))
+
+
 def _media_info(anilist_id: int = 154587) -> dict:
     """
     构造 AniList 榜单测试媒体。
@@ -37,9 +51,9 @@ def _media_info(anilist_id: int = 154587) -> dict:
     }
 
 
-def test_anilist_client_uses_chinese_proxy_and_forwards_discover_filters() -> None:
+def test_anilist_client_forwards_discover_filters() -> None:
     """
-    AniList 探索应通过中文代理并原样传递组合过滤条件。
+    AniList 探索应原样传递组合过滤条件。
     """
     client = AniListApi()
     client._invoke = Mock(return_value={"Page": {"media": [_media_info()]}})
@@ -57,7 +71,6 @@ def test_anilist_client_uses_chinese_proxy_and_forwards_discover_filters() -> No
         sort="SCORE_DESC",
     )
 
-    assert client._base_url == "https://trace.moe/anilist/"
     assert medias[0]["id"] == 154587
     variables = client._invoke.call_args.args[1]
     assert variables == {
@@ -76,11 +89,10 @@ def test_anilist_client_uses_chinese_proxy_and_forwards_discover_filters() -> No
     assert "synonyms" in client._page_query
 
 
-def test_anilist_client_falls_back_and_merges_chinese_dataset() -> None:
+def test_anilist_client_queries_official_and_merges_chinese_dataset() -> None:
     """
-    中文代理不可用时应回退官方接口并合并项目中文数据。
+    直接请求官方接口并合并项目中文数据，避免探测已失效的中文代理。
     """
-    proxy_response = Mock(status_code=403)
     official_media = _media_info()
     official_media["title"].pop("chinese")
     official_media["synonyms"] = ["Official Alias"]
@@ -90,7 +102,7 @@ def test_anilist_client_falls_back_and_merges_chinese_dataset() -> None:
     }
     client = AniListApi()
     client._request = Mock()
-    client._request.post_res.side_effect = [proxy_response, official_response]
+    client._request.post_res.return_value = official_response
     client._request.get_json.return_value = [
         {
             "id": 154587,
@@ -107,18 +119,15 @@ def test_anilist_client_falls_back_and_merges_chinese_dataset() -> None:
         "Frieren at the Funeral",
     ]
     assert [call.args[0] for call in client._request.post_res.call_args_list] == [
-        client._base_url,
         client._official_url,
     ]
     client._request.get_json.assert_called_once_with(client._translations_url)
-    assert client._proxy_available is False
 
 
-def test_anilist_async_client_falls_back_without_real_network() -> None:
+def test_anilist_async_client_queries_official_without_real_network() -> None:
     """
-    异步中文代理失败时也应使用官方接口和同一中文数据集。
+    异步查询也应直接使用官方接口和同一中文数据集。
     """
-    proxy_response = Mock(status_code=403)
     official_media = _media_info()
     official_media["title"].pop("chinese")
     official_response = Mock(status_code=200)
@@ -128,7 +137,7 @@ def test_anilist_async_client_falls_back_without_real_network() -> None:
     client = AniListApi()
     client._async_request = Mock()
     client._async_request.post_res = AsyncMock(
-        side_effect=[proxy_response, official_response]
+        return_value=official_response
     )
     client._async_request.get_json = AsyncMock(
         return_value=[
@@ -140,7 +149,6 @@ def test_anilist_async_client_falls_back_without_real_network() -> None:
 
     assert result["Page"]["media"][0]["title"]["chinese"] == "葬送的芙莉莲"
     assert [call.args[0] for call in client._async_request.post_res.call_args_list] == [
-        client._base_url,
         client._official_url,
     ]
     client._async_request.get_json.assert_awaited_once_with(client._translations_url)

tests/test_anilist_response_contract.py

@@ -1,5 +1,5 @@
 """
-AniList 的无效 GraphQL 响应不得阻止官方回退与中文合并。
+AniList 的无效 GraphQL 响应应返回失败，中文合并保留原始身份与别名顺序。
 """
 
 import asyncio
@@ -11,39 +11,31 @@ from app.modules.anilist.anilist import AniListApi
 
 
 @pytest.mark.parametrize("asynchronous", [False, True])
-@pytest.mark.parametrize("proxy_payload", [
+@pytest.mark.parametrize("payload", [
     None, [], [1], "invalid", 7, False,
     {"data": []}, {"data": [1]}, {"data": "invalid"}, {"data": 7},
     {"data": False}, {"errors": ["unavailable"], "data": {"Media": {"id": 1}}},
 ])
-def test_invalid_proxy_payload_preserves_official_fallback(asynchronous, proxy_payload):
+def test_invalid_official_payload_does_not_download_translations(asynchronous: bool, payload: object) -> None:
     """
-    无效对象、无效 data 和 GraphQL 错误都应触发既有官方查询路径。
+    无效对象、无效 data 和 GraphQL 错误不得触发数据集下载或再次查询。
     """
-    proxy = Mock(status_code=200)
-    proxy.json.return_value = proxy_payload
-    official = Mock(status_code=200)
-    media = {"id": 12, "title": {"native": "Original"}, "synonyms": ["Alias"]}
-    data = {"Media": media}
-    official.json.return_value = {"data": data}
-    request = AsyncMock(side_effect=[proxy, official]) if asynchronous else Mock(side_effect=[proxy, official])
-    translations = [{"id": 12, "title": "中文标题", "synonyms": ["Alias", "Chinese Alias"]}]
+    response = Mock(status_code=200)
+    response.json.return_value = payload
+    request = AsyncMock(return_value=response) if asynchronous else Mock(return_value=response)
     client = AniListApi()
     transport = Mock()
     transport.post_res = request
-    transport.get_json = AsyncMock(return_value=translations) if asynchronous else Mock(return_value=translations)
+    transport.get_json = AsyncMock() if asynchronous else Mock()
     if asynchronous:
         client._async_request = transport
         result = asyncio.run(client._async_invoke("query", {"id": 12}))
     else:
         client._request = transport
         result = client._invoke("query", {"id": 12})
-    assert result is data
-    assert media["title"]["chinese"] == "中文标题"
-    assert media["synonyms"] == ["Alias", "Chinese Alias"]
-    assert client._proxy_available is False
-    assert [call.args[0] for call in request.call_args_list] == [client._base_url, client._official_url]
-    assert all(call.kwargs["json"] == {"query": "query", "variables": {"id": 12}} for call in request.call_args_list)
+    assert result is None
+    request.assert_called_once_with(client._official_url, json={"query": "query", "variables": {"id": 12}})
+    transport.get_json.assert_not_called()
 
 
 @pytest.mark.parametrize("payload", [{}, {"Media": None}, {"Page": {"media": []}}])

tests/test_anilist_translations.py

@@ -0,0 +1,324 @@
+"""
+验证 AniList 中文数据集的跨实例持久化、到期刷新、镜像出口及失败回退。
+"""
+
+import asyncio
+import json
+from pathlib import Path
+from typing import Any
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+
+from app.modules.anilist import anilist as anilist_client
+from app.modules.anilist.anilist import AniListApi
+from app.runtime import cache as runtime_cache
+
+_TranslationEnvironment = tuple[dict[str, Any], list[float]]
+_Translations = list[dict[str, Any]]
+
+
+@pytest.fixture
+def translation_environment(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> _TranslationEnvironment:
+    """
+    隔离数据集缓存、下载出口和时钟，使用真实文件后端且禁止路由至 Redis。
+    """
+    values = {"CACHE_PATH": tmp_path, "GITHUB_PROXY": None}
+    runtime_setting = anilist_client.get_runtime_setting
+    monkeypatch.setattr(anilist_client, "get_runtime_setting", lambda name: values[name] if name in values else runtime_setting(name))
+    monkeypatch.setattr(runtime_cache, "_backend_type_provider", lambda: "redis")
+    now = [100_000.0]
+    monkeypatch.setattr(anilist_client, "time", lambda: now[0])
+    return values, now
+
+
+@pytest.fixture
+def translations() -> _Translations:
+    """
+    提供含数字和字符串标识的中文数据，确保 JSON 快照不会改变标识类型。
+    """
+    return [
+        {"id": 12, "title": "中文标题", "synonyms": ["中文别名"]},
+        {"id": "12", "title": "字符串标题", "synonyms": []},
+    ]
+
+
+def _download(client: AniListApi, asynchronous: bool, payload: object) -> Mock:
+    """
+    替换指定传输的下载边界，不访问真实网络。
+    """
+    request = client._async_request if asynchronous else client._request
+    request.get_json = AsyncMock(return_value=payload) if asynchronous else Mock(return_value=payload)
+    return request.get_json
+
+
+def _load(client: AniListApi, asynchronous: bool) -> dict[object, dict[str, Any]]:
+    """
+    运行同步或异步加载入口并取得最终标题索引。
+    """
+    return asyncio.run(client._async_translation_map()) if asynchronous else client._translation_map()
+
+
+def _cache_path(values: dict[str, Any]) -> Path:
+    """
+    返回当前用例对应的持久化快照路径。
+    """
+    return values["CACHE_PATH"] / "anilist" / "anilist-chinese.json"
+
+
+@pytest.mark.parametrize("first_async", [False, True])
+@pytest.mark.parametrize("second_async", [False, True])
+def test_new_client_reuses_disk_snapshot(
+    translation_environment: _TranslationEnvironment, translations: _Translations,
+    first_async: bool, second_async: bool,
+) -> None:
+    """
+    首次下载落盘，重建客户端和切换同步异步入口均应复用快照而无需重新下载。
+    """
+    values, _ = translation_environment
+    first = AniListApi()
+    download = _download(first, first_async, translations)
+    expected = _load(first, first_async)
+    assert _load(first, first_async) == expected
+    download.assert_called_once_with(first._translations_url)
+    assert expected[12]["title"] == "中文标题"
+    assert expected["12"]["title"] == "字符串标题"
+    assert json.loads(_cache_path(values).read_bytes())["items"] == translations
+
+    second = AniListApi()
+    download = _download(second, second_async, None)
+    assert _load(second, second_async) == expected
+    download.assert_not_called()
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+def test_expired_snapshot_and_memory_refresh(
+    translation_environment: _TranslationEnvironment, translations: _Translations, asynchronous: bool,
+) -> None:
+    """
+    磁盘和内存到期均应重新下载，更新后的快照能由另一个客户端立即读取。
+    """
+    values, now = translation_environment
+    path = _cache_path(values)
+    path.parent.mkdir(parents=True)
+    path.write_text(json.dumps({"expires_at": now[0] - 1, "items": [{"id": 12, "title": "旧标题"}]}))
+    client = AniListApi()
+    download = _download(client, asynchronous, translations)
+    assert _load(client, asynchronous)[12]["title"] == "中文标题"
+    download.assert_called_once()
+
+    now[0] += client._translations_ttl + 1
+    updated = [{"id": 12, "title": "更新标题", "synonyms": []}]
+    download = _download(client, asynchronous, updated)
+    assert _load(client, asynchronous)[12]["title"] == "更新标题"
+    download.assert_called_once()
+    restored = AniListApi()
+    download = _download(restored, not asynchronous, None)
+    assert _load(restored, not asynchronous)[12]["title"] == "更新标题"
+    download.assert_not_called()
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize("payload", [None, {}, [], [{"id": []}], [{"id": 12, "title": []}]])
+def test_failed_refresh_keeps_stale_snapshot_and_retries(
+    translation_environment: _TranslationEnvironment, translations: _Translations,
+    asynchronous: bool, payload: object,
+) -> None:
+    """
+    下载失败或载荷无效不得覆盖旧快照；退避后仍能恢复新中文数据。
+    """
+    values, now = translation_environment
+    path = _cache_path(values)
+    path.parent.mkdir(parents=True)
+    content = json.dumps({"expires_at": now[0] - 1, "items": translations}).encode()
+    path.write_bytes(content)
+    client = AniListApi()
+    download = _download(client, asynchronous, payload)
+    assert _load(client, asynchronous)[12]["title"] == "中文标题"
+    assert _load(client, asynchronous)[12]["title"] == "中文标题"
+    download.assert_called_once()
+    assert path.read_bytes() == content
+
+    now[0] += client._translations_retry + 1
+    download = _download(client, asynchronous, [{"id": 12, "title": "恢复标题"}])
+    assert _load(client, asynchronous)[12]["title"] == "恢复标题"
+    download.assert_called_once()
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+def test_initial_failure_is_not_cached_forever(
+    translation_environment: _TranslationEnvironment, translations: _Translations, asynchronous: bool,
+) -> None:
+    """
+    初次下载失败时不落盘空数据，五分钟后可在同一实例内重试成功。
+    """
+    values, now = translation_environment
+    client = AniListApi()
+    download = _download(client, asynchronous, None)
+    assert _load(client, asynchronous) == {}
+    assert _load(client, asynchronous) == {}
+    download.assert_called_once()
+    assert not _cache_path(values).exists()
+    now[0] += client._translations_retry + 1
+    download = _download(client, asynchronous, translations)
+    assert _load(client, asynchronous)[12]["title"] == "中文标题"
+    download.assert_called_once()
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize("content", [
+    b"{", b"\xff", b"[]", b"{}",
+    b'{"expires_at": "invalid", "items": []}',
+    b'{"expires_at": NaN, "items": []}',
+    b'{"expires_at": 999999, "items": {}}',
+])
+def test_corrupt_cache_is_redownloaded(
+    translation_environment: _TranslationEnvironment, translations: _Translations,
+    asynchronous: bool, content: bytes,
+) -> None:
+    """
+    损坏 JSON、无效编码、时间或条目结构都按缓存未命中处理。
+    """
+    values, _ = translation_environment
+    path = _cache_path(values)
+    path.parent.mkdir(parents=True)
+    path.write_bytes(content)
+    client = AniListApi()
+    download = _download(client, asynchronous, translations)
+    assert _load(client, asynchronous)[12]["title"] == "中文标题"
+    download.assert_called_once()
+    assert json.loads(path.read_bytes())["items"] == translations
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize("operation", ["get", "set", "init"])
+def test_disk_errors_preserve_downloaded_titles(
+    translation_environment: _TranslationEnvironment, translations: _Translations,
+    asynchronous: bool, operation: str, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """
+    缓存目录不可写或文件读写失败时，仍能使用成功下载的数据并在内存复用。
+    """
+    if operation == "init":
+        monkeypatch.setattr(anilist_client, "FileCache", Mock(side_effect=PermissionError("readonly")))
+    client = AniListApi()
+    if operation != "init":
+        cache = client._async_file_cache if asynchronous else client._file_cache
+        error = AsyncMock(side_effect=OSError("disk error")) if asynchronous else Mock(side_effect=OSError("disk error"))
+        monkeypatch.setattr(cache, operation, error)
+    download = _download(client, asynchronous, translations)
+    assert _load(client, asynchronous)[12]["title"] == "中文标题"
+    assert _load(client, asynchronous)[12]["title"] == "中文标题"
+    download.assert_called_once()
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize("mirror_success", [False, True])
+def test_github_mirror_precedes_proxied_raw_url(
+    translation_environment: _TranslationEnvironment, translations: _Translations,
+    asynchronous: bool, mirror_success: bool, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """
+    镜像前缀支持路径和缺失尾斜杠，镜像直连，失败后回到已配置 PROXY 的原始客户端。
+    """
+    values, _ = translation_environment
+    values["GITHUB_PROXY"] = " https://mirror.example/prefix "
+    values["PROXY"] = {"https": "socks5://127.0.0.1:1080"}
+    client = AniListApi()
+    raw = _download(client, asynchronous, translations)
+    transport = Mock()
+    payload = translations if mirror_success else {"error": "unavailable"}
+    transport.get_json = AsyncMock(return_value=payload) if asynchronous else Mock(return_value=payload)
+    constructor = Mock(return_value=transport)
+    monkeypatch.setattr(anilist_client, "AsyncRequestUtils" if asynchronous else "RequestUtils", constructor)
+    assert _load(client, asynchronous)[12]["title"] == "中文标题"
+    constructor.assert_called_once_with(ua=anilist_client.get_runtime_setting("NORMAL_USER_AGENT"))
+    transport.get_json.assert_called_once_with(f"https://mirror.example/prefix/{client._translations_url}")
+    if mirror_success:
+        raw.assert_not_called()
+    else:
+        raw.assert_called_once_with(client._translations_url)
+    request = client._async_request if asynchronous else client._request
+    assert request._proxies
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+def test_failed_write_keeps_newer_memory_on_next_refresh(
+    translation_environment: _TranslationEnvironment, asynchronous: bool, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """
+    新数据落盘失败后再遇下载失败，旧磁盘快照不得覆盖更新的内存数据。
+    """
+    values, now = translation_environment
+    path = _cache_path(values)
+    path.parent.mkdir(parents=True)
+    path.write_text(json.dumps({"expires_at": now[0] - 1, "items": [{"id": 12, "title": "旧标题"}]}))
+    client = AniListApi()
+    cache = client._async_file_cache if asynchronous else client._file_cache
+    error = AsyncMock(side_effect=OSError("readonly")) if asynchronous else Mock(side_effect=OSError("readonly"))
+    monkeypatch.setattr(cache, "set", error)
+    _download(client, asynchronous, [{"id": 12, "title": "新标题"}])
+    assert _load(client, asynchronous)[12]["title"] == "新标题"
+    now[0] += client._translations_ttl + 1
+    _download(client, asynchronous, None)
+    assert _load(client, asynchronous)[12]["title"] == "新标题"
+
+
+def test_mixed_concurrent_clients_download_once(
+    translation_environment: _TranslationEnvironment, translations: _Translations,
+) -> None:
+    """
+    同步线程和多个异步客户端共用加载锁，等待者从磁盘取得同一快照。
+    """
+    sync_client = AniListApi()
+    sync_download = _download(sync_client, False, translations)
+    clients = [AniListApi() for _ in range(6)]
+    downloads = [_download(client, True, translations) for client in clients]
+
+    async def load_all() -> list[dict[object, dict[str, Any]]]:
+        """
+        同时发起跨线程和协程加载，覆盖两种入口的缓存竞争。
+        """
+        return await asyncio.wait_for(asyncio.gather(
+            asyncio.to_thread(sync_client._translation_map),
+            *(client._async_translation_map() for client in clients),
+        ), timeout=5)
+
+    results = asyncio.run(load_all())
+    assert all(result[12]["title"] == "中文标题" for result in results)
+    assert sync_download.call_count + sum(download.call_count for download in downloads) == 1
+
+
+def test_cancelled_loader_releases_shared_lock(
+    translation_environment: _TranslationEnvironment, translations: _Translations,
+) -> None:
+    """
+    异步下载取消后释放共享锁，后续实例仍能重新下载并持久化。
+    """
+    client = AniListApi()
+    successor = AniListApi()
+    download = _download(successor, True, translations)
+
+    async def cancel_and_reload() -> dict[object, dict[str, Any]]:
+        """
+        在下载边界取消首个持锁者，并限定后续加载的完成时间。
+        """
+        entered = asyncio.Event()
+
+        async def blocked_download(_url: str) -> None:
+            """
+            挂起数据集下载直到调用方取消，避免依赖真实网络时序。
+            """
+            entered.set()
+            await asyncio.Event().wait()
+
+        client._async_request.get_json = blocked_download
+        task = asyncio.create_task(client._async_translation_map())
+        await asyncio.wait_for(entered.wait(), timeout=2)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        return await asyncio.wait_for(successor._async_translation_map(), timeout=2)
+
+    assert asyncio.run(cancel_and_reload())[12]["title"] == "中文标题"
+    download.assert_awaited_once()