diff --git a/app/modules/indexer/__init__.py b/app/modules/indexer/__init__.py index 4d5873404..c98f51cff 100644 --- a/app/modules/indexer/__init__.py +++ b/app/modules/indexer/__init__.py @@ -480,6 +480,20 @@ class IndexerModule(_ModuleBase): seconds=seconds, ) + @staticmethod + def __site_user_agent(site: dict[str, Any]) -> Optional[str]: + """ + 使用适配器声明的 User-Agent 规则构造下载上下文。 + + :param site: 包含搜索协议和当前用户配置的站点上下文 + :return: 专站规则处理后的 User-Agent，普通站点保留当前配置值 + """ + parser_class = SPIDER_PARSER_CLASSES.get(str(site.get("parser") or "")) + get_user_agent = getattr(parser_class, "get_user_agent", None) + if callable(get_user_agent): + return cast(str, get_user_agent(site.get("ua"))) + return cast(Optional[str], site.get("ua")) + @staticmethod def __parse_result(site: dict[str, Any], result_array: list[dict[str, Any]], seconds: int) -> List[TorrentInfo]: """ @@ -506,7 +520,7 @@ class IndexerModule(_ModuleBase): site=site.get("id"), site_name=site.get("name"), site_cookie=site.get("cookie"), - site_ua=site.get("ua"), + site_ua=IndexerModule.__site_user_agent(site), site_proxy=bool(site.get("proxy")), site_order=int(site.get("pri") or 0), site_downloader=site.get("downloader"), @@ -575,7 +589,7 @@ class IndexerModule(_ModuleBase): if not torrent.enclosure and not allow_missing: return None torrent.site_cookie = site.get("cookie") or "" - torrent.site_ua = site.get("ua") or "" + torrent.site_ua = IndexerModule.__site_user_agent(site) or "" torrent.site_proxy = bool(site.get("proxy")) torrent.site_downloader = site.get("downloader") or "" return torrent diff --git a/app/modules/indexer/parser/fluxtorrent.py b/app/modules/indexer/parser/fluxtorrent.py index 1f8b9a0a7..1760a3f79 100644 --- a/app/modules/indexer/parser/fluxtorrent.py +++ b/app/modules/indexer/parser/fluxtorrent.py @@ -5,9 +5,13 @@ import json from typing import Any, Dict, Optional, Tuple +from app.adapters.network.http import RequestUtils +from app.application.site.throttle import site_request_config from app.foundation import temporal as time_tools from app.modules.indexer.parser import SitePage, SiteParserBase, SiteSchema +from app.modules.indexer.spider.fluxtorrent import FluxTorrentSpider, get_api_response, site_request_limits from app.runtime.log import logger +from app.runtime.settings import get_runtime_setting class FluxTorrentSiteUserInfo(SiteParserBase): @@ -19,6 +23,8 @@ class FluxTorrentSiteUserInfo(SiteParserBase): schema = SiteSchema.FluxTorrent request_mode = "apikey" + # 开放 API 认证失败时保留接口错误，避免继续尝试网页模型并触发网关限流。 + allow_schema_fallback = False def _parse_site_page(self, html_text: SitePage) -> None: """ @@ -40,10 +46,36 @@ class FluxTorrentSiteUserInfo(SiteParserBase): "Authorization": f"Token {token}", "Accept": "application/json", } - if self._ua: - headers["User-Agent"] = self._ua + headers["User-Agent"] = FluxTorrentSpider.get_user_agent(self._ua) self._addition_headers = headers + def _get_page_content( + self, + url: str, + params: Optional[dict[str, object]] = None, + headers: Optional[dict[str, Optional[str]]] = None, + method: Optional[str] = None, + ) -> str: + """ + 使用只读 API 传输和共享流控读取资料，网关 503 最多退避重试两次。 + + :param url: 用户资料或补充统计接口的完整地址 + :param params: 兼容基类参数，只读接口使用查询参数 + :param headers: 兼容基类参数，认证头由开放 API 令牌构造 + :param method: 兼容基类参数，只读接口始终使用 GET + :return: 接口响应文本，无响应时返回空字符串 + """ + response = get_api_response(RequestUtils( + headers=self._addition_headers, + proxies=get_runtime_setting("PROXY") if self._proxy else None, + timeout=60, + ), url=url, limits=site_request_limits(site_request_config(self._site_url)), params=params) + try: + return response.text if response is not None else "" + finally: + if response is not None: + response.close() + def _load_api_payload(self, html_text: SitePage, operation: str) -> Optional[Dict[str, Any]]: """ 解析并校验兼容层统一响应信封。 @@ -84,6 +116,7 @@ class FluxTorrentSiteUserInfo(SiteParserBase): """ user_info = self._load_api_payload(html_text, "获取用户信息") if not user_info: + self._user_traffic_page = None return self.userid = user_info.get("id") self.username = user_info.get("username") diff --git a/app/modules/indexer/spider/fluxtorrent.py b/app/modules/indexer/spider/fluxtorrent.py index 513acdcd8..f663dcff5 100644 --- a/app/modules/indexer/spider/fluxtorrent.py +++ b/app/modules/indexer/spider/fluxtorrent.py @@ -2,9 +2,12 @@ 通过 FluxTorrent 兼容层搜索种子并转换为通用索引结果。 """ +import asyncio +import time from typing import Any, Dict, List, Optional, Tuple from app.adapters.network.http import AsyncRequestUtils, RequestUtils +from app.application.site.throttle import async_wait_for_site_request, wait_for_site_request from app.foundation import temporal as time_tools from app.runtime.log import logger from app.runtime.settings import get_runtime_setting @@ -12,6 +15,9 @@ from app.schemas.types import MediaType # FluxTorrent 兼容层的只读接口前缀。 _COMPAT_PREFIX = "/api/v1/compat/nexusphp" +# 网关偶发 503 时，只读请求最多重试两次，分别退避三秒和六秒。 +_GATEWAY_RETRY_SECONDS = 3 +_GATEWAY_RETRIES = 2 # 兼容层促销档位到下载、上传量系数的映射，与站点计费口径一致： # free 下载 0 / x2 上传 2 / x2free 0+2 / half 下载 0.5 / x2half 0.5+2 / p30 下载 0.3。 @@ -90,6 +96,74 @@ def site_api_token(site: Dict[str, Any]) -> str: return str(site.get("apikey") or site.get("token") or "").strip() +def site_request_limits(site: Dict[str, Any]) -> Dict[str, Any]: + """ + 构造兼容层请求共用的无凭据流控配置。 + + 保留用户设置的请求间隔和次数窗口，使连接检测、搜索与用户资料请求 + 共用同一站点额度。 + + :param site: 站点索引配置或已保存的站点流控快照 + :return: 只包含站点身份和流控字段的独立配置 + """ + limits = {key: site.get(key) for key in + ("id", "limit_seconds", "limit_interval", "limit_count")} + limits["domain"] = site_base_url(site) + return limits + + +def get_api_response( + request: RequestUtils, + url: str, + limits: Dict[str, Any], + params: Optional[Dict[str, Any]] = None, +) -> Any: + """ + 在共享流控下请求只读 API，网关 503 最多退避重试两次。 + + :param request: 已配置认证头和网络选项的同步传输客户端 + :param url: 只读 API 的完整地址 + :param limits: 不含凭据的站点流控配置 + :param params: 查询参数，未提供时不添加参数 + :return: 最后一次响应或 None，由调用方解析并关闭 + """ + response = None + for attempt in range(_GATEWAY_RETRIES + 1): + wait_for_site_request(limits) + response = request.get_res(url=url, params=params, allow_redirects=False) + if response is None or response.status_code != 503 or attempt == _GATEWAY_RETRIES: + break + response.close() + time.sleep(_GATEWAY_RETRY_SECONDS * (attempt + 1)) + return response + + +async def async_get_api_response( + request: AsyncRequestUtils, + url: str, + limits: Dict[str, Any], + params: Optional[Dict[str, Any]] = None, +) -> Any: + """ + 异步申请同一站点额度，网关 503 使用可取消的延迟最多重试两次。 + + :param request: 已配置认证头和网络选项的异步传输客户端 + :param url: 只读 API 的完整地址 + :param limits: 不含凭据的站点流控配置 + :param params: 查询参数，未提供时不添加参数 + :return: 最后一次响应或 None，由调用方解析并关闭 + """ + response = None + for attempt in range(_GATEWAY_RETRIES + 1): + await async_wait_for_site_request(limits) + response = await request.get_res(url=url, params=params, allow_redirects=False) + if response is None or response.status_code != 503 or attempt == _GATEWAY_RETRIES: + break + await response.aclose() + await asyncio.sleep(_GATEWAY_RETRY_SECONDS * (attempt + 1)) + return response + + class FluxTorrentSpider: """ FluxTorrent 兼容层索引器。 @@ -100,6 +174,8 @@ class FluxTorrentSpider: # 兼容层单页容量上限为 50。 _size = 50 + # 每次实际 API 请求自行申请额度，模块入口无需重复等待。 + manages_request_throttle = True def __init__(self, indexer: dict[str, Any]): """ @@ -112,8 +188,24 @@ class FluxTorrentSpider: self._site_url = site_base_url(indexer) self._api_token = site_api_token(indexer) self._proxy = get_runtime_setting("PROXY") if indexer.get("proxy") else None - self._user_agent = indexer.get("ua") or get_runtime_setting("USER_AGENT") - self._timeout = indexer.get("timeout") or 15 + self._user_agent = self.get_user_agent(indexer.get("ua")) + self._timeout = indexer.get("timeout") or 30 + self._rate_site = site_request_limits(indexer) + + @staticmethod + def get_user_agent(ua: Optional[str]) -> str: + """ + 为 API 请求和种子下载选择站点自定义或默认浏览器 User-Agent。 + + 好学网关会拒绝默认 MoviePilot 标识，返回 Nginx 503。资源合并填入的 + 宿主默认标识同样转换为浏览器标识，站点显式设置的其他值保持优先。 + + :param ua: 站点配置或资源合并后的 User-Agent + :return: 本次站点请求使用的 User-Agent + """ + if not ua or ua == get_runtime_setting("USER_AGENT"): + return str(get_runtime_setting("NORMAL_USER_AGENT")) + return ua @classmethod def get_search_page_size(cls, keyword: Optional[str] = None) -> Optional[int]: @@ -208,16 +300,16 @@ class FluxTorrentSpider: """ if not self._api_token: return False, "未配置开放 API Token（fxo_ 开头）" - with RequestUtils( + response = get_api_response(RequestUtils( headers=self._request_headers(), ua=self._user_agent, proxies=self._proxy or {}, timeout=self._timeout, - ).response_manager( - method="GET", + ), url=f"{self._base_url}/user.json", - allow_redirects=False, - ) as response: + limits=self._rate_site, + ) + try: if response is None: return False, "无法连接 FluxTorrent 兼容层接口" if response.status_code != 200: @@ -244,6 +336,9 @@ class FluxTorrentSpider: return False, str( payload.get("message") or "开放 API Token 无效或已过期" ) + finally: + if response is not None: + response.close() @staticmethod def _media_type(item: Dict[str, Any]) -> str: @@ -354,12 +449,16 @@ class FluxTorrentSpider: logger.warning(f"{self._name} 未配置开放 API 令牌") return True, [] params = self._build_params(keyword, page) - response = RequestUtils( + response = get_api_response(RequestUtils( headers=self._request_headers(), proxies=self._proxy, timeout=self._timeout, - ).get_res(url=f"{self._base_url}/torrents.json", params=params) - return self._process_search_response(response) + ), url=f"{self._base_url}/torrents.json", limits=self._rate_site, params=params) + try: + return self._process_search_response(response) + finally: + if response is not None: + response.close() async def async_search( self, @@ -377,9 +476,13 @@ class FluxTorrentSpider: logger.warning(f"{self._name} 未配置开放 API 令牌") return True, [] params = self._build_params(keyword, page) - response = await AsyncRequestUtils( + response = await async_get_api_response(AsyncRequestUtils( headers=self._request_headers(), proxies=self._proxy, timeout=self._timeout, - ).get_res(url=f"{self._base_url}/torrents.json", params=params) - return self._process_search_response(response) + ), url=f"{self._base_url}/torrents.json", limits=self._rate_site, params=params) + try: + return self._process_search_response(response) + finally: + if response is not None: + await response.aclose() diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 3ec30ab46..47e2bbac8 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -837,7 +837,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 1164 | -| 内部导入边 | 10,124 | +| 内部导入边 | 10,136 | | 非平凡 SCC | 1（29 个模块的 TMDB vendor 精确 containment） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 51（精确 policy 准入，Application / Chain 均为 0） | diff --git a/docs/site-api-integration.md b/docs/site-api-integration.md index 141ffecb9..ad4489f91 100644 --- a/docs/site-api-integration.md +++ b/docs/site-api-integration.md @@ -458,3 +458,19 @@ MP 已有 API Key 搜索及临时下载地址的实现案例，但不会自动 批量辅种接口接收的是 `info_hash`，不能将青蛙辅种插件的 `pieces_hash` 当成同一种指纹。 离线回归只使用合成凭据并替换网络边界；实际 API 验收结果需另外记录，HTTP 200 本身不能证明种子文件有效。 + +### 好学 FluxTorrent 兼容层适配 + +好学使用 `/api/v1/compat/nexusphp` 只读 JSON 兼容层，站点索引资源从 `3.0.28` 起声明 `FluxTorrent` 搜索和用户数据协议。 +登录站点后，在安全设置页面的「开放 API 令牌」中签发包含 `read` 权限的 `fxo_` 开头令牌，填入 MoviePilot 站点 **API Key**。 +API Key 为空时使用 **Token**。这枚开放 API 令牌与站点 RSS 的 passkey 分别管理。[好学安全设置](https://www.hxpt.org/my?tab=security) + +- 连接检测和用户资料使用 `GET /user.json`，读取等级、上传下载量、分享率、魔力、做种数量、做种体积和注册时间。 + 老版本接口缺少做种和注册信息时，通过 `/api/v1/plugins/ptppUserInfo` 补齐。 +- 搜索和最新列表使用 `GET /torrents.json`，页码从 1 开始，每页 50 条，关键词使用 `keyword`。 + 分类和促销倍率取兼容层字段，详情链接指向 `/torrent/{id}`，种子文件通过 `GET /download.php?id=...&apikey=...` 获取。 +- 连接检测、用户资料及同步、异步搜索共用站点设置的请求间隔和次数窗口。 + 只读 API 遇到 HTTP 503 时分别等待三秒和六秒，最多重试两次；持续失败保留失败结果。 + 认证失败直接结束本次用户解析并保留接口错误。 +- 好学网关会对默认 MoviePilot User-Agent 返回 HTTP 503，适配器使用默认浏览器标识， + 并将相同标识传递给种子下载；站点手工设置的其他 User-Agent 优先。 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 8283e9547..bc780fb73 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -962,8 +962,8 @@ "runtime_only": true } }, - "edge_count": 10124, - "edge_sha256": "3dac06b5d1ba3e44d6ed06f06c678f43767e1e8ee613cd4b7d2d09c07ccfd463", + "edge_count": 10136, + "edge_sha256": "3768338a92d10d7883a42acdb9eff190fa1cd0af9e68b7ae317c03d3ddbf3f9c", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -7906,13 +7906,22 @@ "app.modules.indexer.parser.file_list -> app.modules", "app.modules.indexer.parser.file_list -> app.modules.indexer", "app.modules.indexer.parser.file_list -> app.modules.indexer.parser", + "app.modules.indexer.parser.fluxtorrent -> app.adapters", + "app.modules.indexer.parser.fluxtorrent -> app.adapters.network", + "app.modules.indexer.parser.fluxtorrent -> app.adapters.network.http", + "app.modules.indexer.parser.fluxtorrent -> app.application", + "app.modules.indexer.parser.fluxtorrent -> app.application.site", + "app.modules.indexer.parser.fluxtorrent -> app.application.site.throttle", "app.modules.indexer.parser.fluxtorrent -> app.foundation", "app.modules.indexer.parser.fluxtorrent -> app.foundation.temporal", "app.modules.indexer.parser.fluxtorrent -> app.modules", "app.modules.indexer.parser.fluxtorrent -> app.modules.indexer", "app.modules.indexer.parser.fluxtorrent -> app.modules.indexer.parser", + "app.modules.indexer.parser.fluxtorrent -> app.modules.indexer.spider", + "app.modules.indexer.parser.fluxtorrent -> app.modules.indexer.spider.fluxtorrent", "app.modules.indexer.parser.fluxtorrent -> app.runtime", "app.modules.indexer.parser.fluxtorrent -> app.runtime.log", + "app.modules.indexer.parser.fluxtorrent -> app.runtime.settings", "app.modules.indexer.parser.gazelle -> app.foundation", "app.modules.indexer.parser.gazelle -> app.foundation.dom", "app.modules.indexer.parser.gazelle -> app.foundation.size", @@ -8110,6 +8119,9 @@ "app.modules.indexer.spider.fluxtorrent -> app.adapters", "app.modules.indexer.spider.fluxtorrent -> app.adapters.network", "app.modules.indexer.spider.fluxtorrent -> app.adapters.network.http", + "app.modules.indexer.spider.fluxtorrent -> app.application", + "app.modules.indexer.spider.fluxtorrent -> app.application.site", + "app.modules.indexer.spider.fluxtorrent -> app.application.site.throttle", "app.modules.indexer.spider.fluxtorrent -> app.foundation", "app.modules.indexer.spider.fluxtorrent -> app.foundation.temporal", "app.modules.indexer.spider.fluxtorrent -> app.runtime", diff --git a/tests/test_flux_torrent_indexer.py b/tests/test_flux_torrent_indexer.py index 264a3d81d..613bb78f6 100644 --- a/tests/test_flux_torrent_indexer.py +++ b/tests/test_flux_torrent_indexer.py @@ -4,11 +4,15 @@ import asyncio import json +from unittest.mock import AsyncMock, Mock + +import pytest import app.modules.indexer as indexer_module from app.foundation import temporal as time_tools from app.modules.indexer import IndexerModule from app.modules.indexer.parser import SiteSchema +from app.modules.indexer.parser import fluxtorrent as flux_parser_module from app.modules.indexer.parser.fluxtorrent import FluxTorrentSiteUserInfo from app.modules.indexer.spider import fluxtorrent as flux_torrent_module from app.modules.indexer.spider.fluxtorrent import FluxTorrentSpider @@ -18,6 +22,19 @@ _SITE_URL = "https://flux.example/" _API_TOKEN = "fxo_test_token" +@pytest.fixture(autouse=True) +def _isolate_request_limits(monkeypatch): + """ + 隔离共享请求等待和站点查询，回放测试不消耗真实时间或站点配置。 + """ + monkeypatch.setattr(flux_torrent_module, "wait_for_site_request", Mock()) + monkeypatch.setattr(flux_torrent_module, "async_wait_for_site_request", AsyncMock()) + monkeypatch.setattr( + flux_parser_module, "site_request_config", + lambda _url: {"id": "flux", "domain": _SITE_URL}, + ) + + class _FakeResponse: """ 提供 FluxTorrent 兼容层响应判定所需的最小 HTTP 响应契约。 @@ -31,6 +48,7 @@ class _FakeResponse: self.status_code = status_code self.reason = "OK" self.text = json.dumps(payload) if payload is not None else "" + self.closed = False def __bool__(self) -> bool: """ @@ -44,6 +62,18 @@ class _FakeResponse: """ return self._payload + def close(self): + """ + 记录同步响应释放，供重试和最终解析收尾断言。 + """ + self.closed = True + + async def aclose(self): + """ + 记录异步响应释放，避免重试延迟期间继续持有连接。 + """ + self.closed = True + def _build_indexer(**overrides) -> dict: """ @@ -142,6 +172,174 @@ def test_flux_torrent_spider_maps_compat_items(monkeypatch): }] +def test_flux_torrent_request_limits_preserve_user_settings(): + """ + 用户配置的间隔与次数限制保持有效，流控快照不携带凭据。 + """ + limits = flux_torrent_module.site_request_limits(_build_indexer( + limit_seconds=8, limit_interval=60, limit_count=5, + )) + + assert limits == { + "id": "flux", "domain": _SITE_URL.rstrip("/"), + "limit_seconds": 8, "limit_interval": 60, "limit_count": 5, + } + assert flux_torrent_module.site_request_limits(_build_indexer())["limit_seconds"] is None + assert FluxTorrentSpider.manages_request_throttle is True + + +def test_flux_torrent_search_timeout_preserves_user_setting(): + """ + 慢搜索默认等待三十秒，用户显式设置的超时时间继续优先。 + """ + assert FluxTorrentSpider(_build_indexer())._timeout == 30 + assert FluxTorrentSpider(_build_indexer(timeout=45))._timeout == 45 + + +@pytest.mark.parametrize("ua", [None, "", "MoviePilot/3.2.6", "Custom-UA"]) +def test_flux_torrent_user_agent_reaches_api_and_download_context(monkeypatch, ua): + """ + 默认标识转换为浏览器标识，资料、搜索结果和检查点恢复沿用相同规则。 + """ + monkeypatch.setattr(flux_torrent_module, "get_runtime_setting", lambda key: { + "USER_AGENT": "MoviePilot/3.2.6", "NORMAL_USER_AGENT": "Browser-UA", + }[key]) + site = _build_indexer(id=42, parser="FluxTorrent", ua=ua) + expected = "Custom-UA" if ua == "Custom-UA" else "Browser-UA" + spider = FluxTorrentSpider(site) + assert spider._request_headers()["User-Agent"] == expected + records = spider._parse_result(_torrent_items()) + torrents = IndexerModule._IndexerModule__parse_result(site, records, seconds=0) + assert torrents[0].site_ua == expected + restored = IndexerModule.restore_search_torrent(site, torrents[0].to_dict()) + assert restored.site_ua == expected + parser = FluxTorrentSiteUserInfo( + site_name="FluxTorrent", url=_SITE_URL, site_cookie=None, + apikey=_API_TOKEN, token=None, ua=ua, + ) + parser._parse_site_page({}) + assert parser._addition_headers["User-Agent"] == expected + + +def test_standard_indexer_preserves_configured_user_agent(): + """ + 普通站点搜索结果及检查点恢复保持原有 User-Agent 配置语义。 + """ + site = _build_indexer(id=42, ua="Custom-UA") + records = FluxTorrentSpider(site)._parse_result(_torrent_items()) + torrents = IndexerModule._IndexerModule__parse_result(site, records, seconds=0) + assert torrents[0].site_ua == "Custom-UA" + restored = IndexerModule.restore_search_torrent(site, torrents[0].to_dict()) + assert restored.site_ua == "Custom-UA" + + +def test_flux_torrent_connection_and_search_wait_before_each_request(monkeypatch): + """ + 连接检测、同步搜索与异步搜索应在 HTTP 调用之前申请相同的站点额度。 + """ + events = [] + expected_limits = flux_torrent_module.site_request_limits(_build_indexer()) + + def wait(limits): + """ + 记录同步请求额度申请并核对稳定站点身份。 + """ + assert limits == expected_limits + events.append("wait") + + async def async_wait(limits): + """ + 记录异步请求额度申请并核对稳定站点身份。 + """ + wait(limits) + + def search(_self, url, **_kwargs): + """ + 回放同步种子搜索响应。 + """ + events.append("connection" if url.endswith("user.json") else "search") + return _FakeResponse(_envelope({"items": _torrent_items()})) + + async def async_search(_self, **_kwargs): + """ + 回放异步种子搜索响应。 + """ + events.append("async_search") + return _FakeResponse(_envelope({"items": _torrent_items()})) + + monkeypatch.setattr(flux_torrent_module, "wait_for_site_request", wait) + monkeypatch.setattr(flux_torrent_module, "async_wait_for_site_request", async_wait) + monkeypatch.setattr(flux_torrent_module.RequestUtils, "get_res", search) + monkeypatch.setattr(flux_torrent_module.AsyncRequestUtils, "get_res", async_search) + + spider = FluxTorrentSpider(_build_indexer()) + assert spider.test_connection()[0] + assert not spider.search(keyword="Example")[0] + assert not asyncio.run(spider.async_search(keyword="Example"))[0] + assert events == ["wait", "connection", "wait", "search", "wait", "async_search"] + + +def test_flux_torrent_userdata_waits_before_profile_and_legacy_statistics(monkeypatch): + """ + 资料刷新和老版本补充统计均须经过共享流控，沿用搜索与连接检测的站点身份。 + """ + events = [] + expected_limits = flux_torrent_module.site_request_limits(_build_indexer()) + + def wait(limits): + """ + 核对资料刷新使用的站点额度与专站爬虫一致。 + """ + assert limits == expected_limits + events.append("wait") + + def get_res(_self, url, **_kwargs): + """ + 回放缺少做种字段的资料以及后续补充统计。 + """ + events.append(url.rsplit("/", 1)[-1]) + data = {"id": 42, "username": "tester"} if url.endswith("user.json") else { + "seeding": 12, "seedingSize": 1024, "joinTime": "2025-07-15T20:45:13+00:00", + } + return _FakeResponse(_envelope(data)) + + monkeypatch.setattr(flux_torrent_module, "wait_for_site_request", wait) + monkeypatch.setattr(flux_torrent_module.RequestUtils, "get_res", get_res) + parser = FluxTorrentSiteUserInfo( + site_name="FluxTorrent", url=_SITE_URL, site_cookie=None, + apikey=_API_TOKEN, token=None, ua="MoviePilot-Test", + ) + parser.parse() + + assert events == ["wait", "user.json", "wait", "ptppUserInfo"] + assert parser.seeding == 12 + assert parser.seeding_size == 1024 + assert parser.err_msg is None + + +def test_flux_torrent_auth_failure_does_not_request_other_userdata_models(monkeypatch): + """ + API 认证失败后保留失败结果，停止补充统计和网页模型回退，避免额外请求放大限流。 + """ + requests = [] + + def get_res(_self, url, **_kwargs): + """ + 记录真实资料入口发出的请求并回放认证失败。 + """ + requests.append(url) + return _FakeResponse(_envelope(None, code=2001, message="未认证"), status_code=401) + + monkeypatch.setattr(flux_torrent_module.RequestUtils, "get_res", get_res) + monkeypatch.setattr(IndexerModule, "_site_schemas", [FluxTorrentSiteUserInfo]) + module = object.__new__(IndexerModule) + result = module.refresh_userdata(_build_indexer(schema="FluxTorrent")) + + assert result.userid is None + assert result.err_msg == "未认证" + assert requests == [f"{_SITE_URL}api/v1/compat/nexusphp/user.json"] + + def test_flux_torrent_spider_maps_promotion_factors(): """ 促销档位应换算为站点计费口径的下载与上传倍率。 @@ -181,44 +379,20 @@ def test_flux_torrent_spider_maps_media_types(): assert FluxTorrentSpider._media_type({}) == MediaType.UNKNOWN.value -class _FakeResponseManager: - """ - 提供 RequestUtils.response_manager 的上下文管理器契约。 - """ - - def __init__(self, response): - """ - 保存要回放的响应。 - """ - self._response = response - - def __enter__(self): - """ - 返回回放的响应。 - """ - return self._response - - def __exit__(self, *_exc): - """ - 不吞异常。 - """ - return False - - def _patch_connection(monkeypatch, response, captured: dict = None): """ 让连接检测拿到固定响应，并记录请求地址。 """ - def fake_manager(_self, **kwargs): + def fake_get_res(_self, **kwargs): """ 回放预设响应。 """ if captured is not None: captured.update(kwargs) - return _FakeResponseManager(response) + return response monkeypatch.setattr( - flux_torrent_module.RequestUtils, "response_manager", fake_manager + flux_torrent_module.RequestUtils, "get_res", fake_get_res ) @@ -245,14 +419,14 @@ def test_flux_torrent_test_connection_requires_token(monkeypatch): """ 没有令牌时不应发起连接检测请求。 """ - def fake_manager(*_args, **_kwargs): + def fake_get_res(*_args, **_kwargs): """ 记录不应发生的请求。 """ raise AssertionError("未配置令牌时不应发起请求") monkeypatch.setattr( - flux_torrent_module.RequestUtils, "response_manager", fake_manager + flux_torrent_module.RequestUtils, "get_res", fake_get_res ) assert FluxTorrentSpider(_build_indexer(apikey="")).test_connection() == ( @@ -277,6 +451,96 @@ def test_flux_torrent_test_connection_surfaces_auth_error(monkeypatch): assert "未认证" in message +@pytest.mark.parametrize("operation", ["connection", "search", "userdata"]) +def test_flux_torrent_retries_gateway_503_once(monkeypatch, operation): + """ + 只读入口遇到网关 503 时先释放响应，延迟后重新申请额度并重试一次。 + """ + first = _FakeResponse(None, status_code=503) + second = _FakeResponse(_envelope({ + "id": 42, "username": "tester", "items": _torrent_items(), + "seeding": 0, "seeding_size": 0, "join_at": "2025-07-15T20:45:13+00:00", + })) + request = Mock(side_effect=[first, second]) + sleep = Mock() + monkeypatch.setattr(flux_torrent_module.RequestUtils, "get_res", request) + monkeypatch.setattr(flux_torrent_module.time, "sleep", sleep) + + if operation == "connection": + assert FluxTorrentSpider(_build_indexer()).test_connection()[0] + elif operation == "search": + assert not FluxTorrentSpider(_build_indexer()).search(keyword="Example")[0] + else: + parser = FluxTorrentSiteUserInfo( + site_name="FluxTorrent", url=_SITE_URL, site_cookie=None, + apikey=_API_TOKEN, token=None, ua="MoviePilot-Test", + ) + parser.parse() + assert parser.userid == 42 + assert parser.err_msg is None + + assert request.call_count == 2 + assert first.closed and second.closed + sleep.assert_called_once_with(3) + assert flux_torrent_module.wait_for_site_request.call_count == 2 + assert request.call_args.kwargs["allow_redirects"] is False + + +@pytest.mark.parametrize("status", [401, 403, 429, 503]) +def test_flux_torrent_retry_is_bounded_and_preserves_final_failure(monkeypatch, status): + """ + 认证与限流失败保持原响应，连续 503 最多重试两次并将最终失败返回调用方。 + """ + response = _FakeResponse(_envelope(None, code=2001, message="接口失败"), status_code=status) + request = Mock(return_value=response) + sleep = Mock() + monkeypatch.setattr(flux_torrent_module.RequestUtils, "get_res", request) + monkeypatch.setattr(flux_torrent_module.time, "sleep", sleep) + + assert not FluxTorrentSpider(_build_indexer()).test_connection()[0] + assert request.call_count == (3 if status == 503 else 1) + assert sleep.call_count == (2 if status == 503 else 0) + if status == 503: + assert [call.args[0] for call in sleep.call_args_list] == [3, 6] + assert response.closed + + +def test_flux_torrent_async_gateway_retry_releases_response(monkeypatch): + """ + 异步 503 重试使用可等待的延迟并释放两次响应，结果与同步解析保持一致。 + """ + first = _FakeResponse(None, status_code=503) + second = _FakeResponse(_envelope({"items": _torrent_items()})) + request = AsyncMock(side_effect=[first, second]) + sleep = AsyncMock() + monkeypatch.setattr(flux_torrent_module.AsyncRequestUtils, "get_res", request) + monkeypatch.setattr(flux_torrent_module.asyncio, "sleep", sleep) + + error, torrents = asyncio.run(FluxTorrentSpider(_build_indexer()).async_search(keyword="Example")) + + assert not error and len(torrents) == 1 + assert request.await_count == 2 + assert first.closed and second.closed + sleep.assert_awaited_once_with(3) + assert flux_torrent_module.async_wait_for_site_request.await_count == 2 + + +def test_flux_torrent_async_gateway_retry_can_be_cancelled(monkeypatch): + """ + 取消异步退避时传播取消信号，已释放失败响应且不继续发送请求。 + """ + response = _FakeResponse(None, status_code=503) + request = AsyncMock(return_value=response) + monkeypatch.setattr(flux_torrent_module.AsyncRequestUtils, "get_res", request) + monkeypatch.setattr(flux_torrent_module.asyncio, "sleep", AsyncMock(side_effect=asyncio.CancelledError)) + + with pytest.raises(asyncio.CancelledError): + asyncio.run(FluxTorrentSpider(_build_indexer()).async_search(keyword="Example")) + + assert response.closed + assert request.await_count == 1 + + def test_flux_torrent_test_connection_rejects_non_json(monkeypatch): """ 首页 HTML 等非兼容层响应不能被当成连接成功。 @@ -618,7 +882,7 @@ def test_refresh_userdata_uses_declared_flux_torrent_schema(monkeypatch): assert result.ratio == 2.0 -class _BrokenResponse: +class _BrokenResponse(_FakeResponse): """ 提供响应体不是 JSON 时的最小 HTTP 响应契约。 """