diff --git a/app/modules/anilist/anilist.py b/app/modules/anilist/anilist.py index 47307a26a..d1ba5694d 100644 --- a/app/modules/anilist/anilist.py +++ b/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: 按输入顺序排列的媒体列表 diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 6cea311a8..d66327843 100644 --- a/docs/architecture-overview.md +++ b/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） | diff --git a/docs/testing.md b/docs/testing.md index fba3b25d0..933845a5a 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -87,6 +87,13 @@ AppPush 的所有用户范围由 `tests/test_notification_routing_parity.py` 验 ## 外部依赖：一律 mock，零真实网络 +AniList 直接访问官方 GraphQL 接口，不再探测 `trace.moe/anilist/`。中文数据集保存在 +`/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`，并保留站点代理和认证配置。 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 703a4ba01..64b017d29 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/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", diff --git a/tests/test_anilist_browse.py b/tests/test_anilist_browse.py index 3b3f7235f..cbec32d0b 100644 --- a/tests/test_anilist_browse.py +++ b/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) diff --git a/tests/test_anilist_response_contract.py b/tests/test_anilist_response_contract.py index ddcf7f225..5fce39071 100644 --- a/tests/test_anilist_response_contract.py +++ b/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": []}}]) diff --git a/tests/test_anilist_translations.py b/tests/test_anilist_translations.py new file mode 100644 index 000000000..fbdd9c8d9 --- /dev/null +++ b/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()