fix(site): route API connectivity checks by site protocol (#7105)

2026-10-31 12:24:39 +08:00 · 2026-10-20 15:58:27 +08:00
parent e65e0bbd7b
commit ec7e4acfba
4 changed files with 283 additions and 20 deletions
--- a/app/chain/site.py
+++ b/app/chain/site.py
@@ -9,7 +9,7 @@ from contextlib import AbstractContextManager
 from dataclasses import dataclass, replace
 from datetime import datetime
 from typing import Any, Callable, Dict, Mapping, Optional, Protocol, Tuple, Union
-from urllib.parse import urljoin
+from urllib.parse import urljoin, urlparse

 from lxml import etree

@@ -88,6 +88,7 @@ class SiteHttpPort(Protocol):
        ua: Optional[str] = None,
        proxies: Optional[Mapping[str, Any]] = None,
        timeout: int = 20,
+        json_body: Optional[Mapping[str, Any]] = None,
    ) -> AbstractContextManager[Optional[SiteResponsePort]]:
        """
        打开有界响应上下文，退出时必须关闭底层响应。
@@ -99,6 +100,7 @@ class SiteHttpPort(Protocol):
        :param ua: 本次使用的 User-Agent，None 时使用客户端默认值
        :param proxies: 协议到代理地址的映射，None 表示未指定代理
        :param timeout: 请求超时秒数，默认 20 秒
+        :param json_body: JSON 请求体，None 表示不发送 JSON
        :return: 进入后提供响应或 None 的上下文管理器；退出时关闭底层响应
        """
        ...
@@ -245,9 +247,20 @@ def _open_site_response(
    ua: Optional[str] = None,
    proxies: Optional[Mapping[str, Any]] = None,
    timeout: int = 20,
+    json_body: Optional[Mapping[str, Any]] = None,
 ) -> AbstractContextManager[Optional[SiteResponsePort]]:
    """
    通过当前 HTTP 端口打开一个必定有界关闭的响应上下文。
+
+    :param method: HTTP 请求方法
+    :param url: 完整请求地址
+    :param headers: 额外请求头
+    :param cookie: 登录 Cookie
+    :param ua: 用户代理字符串
+    :param proxies: 代理配置
+    :param timeout: 请求超时秒数
+    :param json_body: JSON 请求体，None 表示不发送 JSON
+    :return: 由调用方进入并负责退出的响应上下文
    """
    return _site_ports_snapshot().http.open(
        method=method,
@@ -257,6 +270,7 @@ def _open_site_response(
        ua=ua,
        proxies=proxies,
        timeout=timeout,
+        json_body=json_body,
    )


@@ -270,7 +284,7 @@ class SiteChain(InteractionChainMixin, ChainBase):

    def __init__(self) -> None:
        """
-        初始化站点管理处理链及特殊站点测试器
+        初始化站点管理处理链及按域名、资源协议选择的连接测试器。
        """
        super().__init__()

@@ -285,6 +299,18 @@ class SiteChain(InteractionChainMixin, ChainBase):
            "sunnypt.top": self.__sunnypt_test,
            "milkie.cc": self.__milkie_test,
        }
+        # 资源的 schema 描述资料协议，parser 描述搜索协议；二者均可识别 API 站点。
+        self.protocol_site_test = {
+            "MTorrent": self.__mteam_test,
+            "mTorrent": self.__mteam_test,
+            "HDDolby": self.__hddolby_test,
+            "RousiPro": self.__rousi_test,
+            "SunnyPT": self.__sunnypt_test,
+            "Yema": self.__yema_api_test,
+            "Milkie": self.__milkie_test,
+            "TNode": self.__zhuque_test,
+            "TNodeSpider": self.__zhuque_test,
+        }

    def refresh_userdata(self, site: dict = None) -> Optional[SiteUserData]:
        """
@@ -409,7 +435,10 @@ class SiteChain(InteractionChainMixin, ChainBase):

    def __zhuque_test(self, site: SiteSnapshot) -> Tuple[bool, str]:
        """
-        判断站点是否已经登陆：zhuique
+        提取 TNode CSRF 令牌并通过当前用户 API 验证 Cookie 会话。
+
+        :param site: TNode 协议站点配置
+        :return: 是否可用及认证或连接失败原因
        """
        token = None
        user_agent = site.ua or self.runtime_config.user_agent
@@ -463,7 +492,15 @@ class SiteChain(InteractionChainMixin, ChainBase):
        """
        user_agent = site.ua or self.runtime_config.user_agent
        domain = site_rules.extract_domain(site.url)
-        url = f"https://api.{domain}/api/member/profile"
+        indexer = SitesHelper().get_indexer(domain) or {}
+        configured_host = indexer.get("api_domain") or indexer.get("api_host")
+        api_host = urlparse(site.url).netloc
+        if configured_host:
+            value = str(configured_host).strip()
+            api_host = urlparse(value if "://" in value else f"//{value}").netloc
+        elif not api_host.startswith("api."):
+            api_host = f"api.{domain}"
+        url = f"https://{api_host}/api/member/profile"
        headers = {
            "User-Agent": user_agent,
            "Accept": "application/json, text/plain, */*",
@@ -480,7 +517,7 @@ class SiteChain(InteractionChainMixin, ChainBase):
                return False, "无法打开网站！"
            if res.status_code == 200:
                user_info = res.json() or {}
-                if user_info.get("data"):
+                if str(user_info.get("code")) == "0" and user_info.get("data"):
                    return True, "连接成功"
                return False, user_info.get("message", "鉴权已过期或无效")
            return False, f"错误：{res.status_code} {res.reason}"
@@ -492,10 +529,9 @@ class SiteChain(InteractionChainMixin, ChainBase):
        :param site: Milkie 站点配置
        :return: 是否可用及状态信息
        """
-        domain = site.domain or site_rules.extract_domain(site.url)
        with _open_site_response(
            method="GET",
-            url=f"https://{domain}/api/v1/torrents?ps=1",
+            url=f"{site.url.rstrip('/')}/api/v1/torrents?ps=1",
            headers={
                "Accept": "application/json",
                "User-Agent": site.ua or self.runtime_config.user_agent,
@@ -507,7 +543,10 @@ class SiteChain(InteractionChainMixin, ChainBase):
            if res is None:
                return False, "无法打开网站！"
            if res.status_code == 200:
-                return True, "连接成功"
+                payload = res.json()
+                if isinstance(payload, dict) and isinstance(payload.get("torrents"), list):
+                    return True, "连接成功"
+                return False, "站点返回了无效的种子数据"
            if res.status_code in (401, 403):
                return False, "API Key 无效或已过期"
            return False, f"错误：{res.status_code} {res.reason}"
@@ -522,7 +561,7 @@ class SiteChain(InteractionChainMixin, ChainBase):
        domain = site.domain or site_rules.extract_domain(site.url)
        indexer = SitesHelper().get_indexer(domain) or {}
        api_url = str(
-            indexer.get("api_url") or "https://api.sunnypt.top/api/v1/mp"
+            indexer.get("api_url") or f"https://api.{domain}/api/v1/mp"
        ).rstrip("/")
        with _open_site_response(
            method="GET",
@@ -553,8 +592,13 @@ class SiteChain(InteractionChainMixin, ChainBase):

    def __yema_test(self, site: SiteSnapshot) -> Tuple[bool, str]:
        """
-        判断站点是否已经登陆：yemapt
+        优先验证 Yema AuthKey，未配置时保留旧版 Cookie 会话探测。
+
+        :param site: YemaPT 站点配置
+        :return: 是否可用及认证或连接失败原因
        """
+        if site.apikey:
+            return self.__yema_api_test(site)
        user_agent = site.ua or self.runtime_config.user_agent
        url = f"{site.url}api/consumer/fetchSelfDetail"
        headers = {
@@ -579,11 +623,45 @@ class SiteChain(InteractionChainMixin, ChainBase):
                return False, "Cookie已过期"
            return False, f"错误：{res.status_code} {res.reason}"

+    def __yema_api_test(self, site: SiteSnapshot) -> Tuple[bool, str]:
+        """
+        通过 Yema 开放 API 验证 AuthKey，与搜索和资料解析使用相同认证方式。
+
+        :param site: Yema 协议站点配置
+        :return: 是否可用及认证或连接失败原因
+        """
+        if not site.apikey:
+            return False, "未配置 AuthKey"
+        with _open_site_response(
+            method="POST",
+            url=f"{site.url.rstrip('/')}/openApi/user/fetchBasicInfo.json",
+            headers={
+                "Authorization": site.apikey,
+                "Content-Type": "application/json",
+                "Accept": "application/json, text/plain, */*",
+                "User-Agent": site.ua or self.runtime_config.user_agent,
+            },
+            json_body={},
+            proxies=(self.runtime_config.proxy or {}) if site.proxy else {},
+            timeout=site.timeout or 15,
+        ) as res:
+            if res is None:
+                return False, "无法打开网站！"
+            if res.status_code != 200:
+                return False, f"错误：{res.status_code} {res.reason}"
+            payload = res.json() or {}
+            if payload.get("success") and isinstance(payload.get("data"), dict) and payload["data"].get("id"):
+                return True, "连接成功"
+            return False, payload.get("errorMessage") or "AuthKey 已过期或无效"
+
    def __hddolby_test(self, site: SiteSnapshot) -> Tuple[bool, str]:
        """
-        判断站点是否已经登陆：hddolby
+        通过 HDDolby 用户 API 验证 API Key，与资料解析使用相同 API 主机。
+
+        :param site: HDDolby 协议站点配置
+        :return: 是否可用及认证或连接失败原因
        """
-        url = f"{site.url}api/v1/user/data"
+        url = f"https://api.{site_rules.extract_domain(site.url)}/api/v1/user/data"
        headers = {
            "Content-Type": "application/json",
            "Accept": "application/json, text/plain, */*",
@@ -607,7 +685,10 @@ class SiteChain(InteractionChainMixin, ChainBase):

    def __rousi_test(self, site: SiteSnapshot) -> Tuple[bool, str]:
        """
-        使用 PeerGo 个人 API Key 验证 Rousi.pro 站点连接。
+        使用 PeerGo 个人 API Key 验证 RousiPro 协议站点连接。
+
+        :param site: 使用 Bearer 个人 API Key 的站点配置
+        :return: 是否可用及认证或连接失败原因
        """
        if not site.apikey:
            return False, "未配置个人 API Key"
@@ -991,8 +1072,10 @@ class SiteChain(InteractionChainMixin, ChainBase):
                state, message = special_test(site_info)
            else:
                indexer = SitesHelper().get_indexer(domain) or {}
-                if indexer.get("schema") == "MTorrent":
-                    state, message = self.__mteam_test(site_info)
+                protocol_test = self.protocol_site_test.get(str(indexer.get("schema") or "")) \
+                    or self.protocol_site_test.get(str(indexer.get("parser") or ""))
+                if protocol_test:
+                    state, message = protocol_test(site_info)
                else:
                    state, message = self.__test(
                        replace(
--- a/app/startup/composition/site.py
+++ b/app/startup/composition/site.py
@@ -210,9 +210,20 @@ class _SiteHttpAdapter:
        ua: Optional[str] = None,
        proxies: Optional[Mapping[str, Any]] = None,
        timeout: int = 20,
+        json_body: Optional[Mapping[str, Any]] = None,
    ) -> Iterator[Optional[SiteResponsePort]]:
        """
        打开响应并在所有返回或异常分支后关闭底层连接。
+
+        :param method: HTTP 请求方法
+        :param url: 完整请求地址
+        :param headers: 额外请求头
+        :param cookie: 登录 Cookie
+        :param ua: 用户代理字符串
+        :param proxies: 代理配置
+        :param timeout: 请求超时秒数
+        :param json_body: JSON 请求体，None 时保留无请求体行为
+        :yields: 在上下文有效期内提供响应或 None
        """
        request = RequestUtils(
            headers=dict(headers or {}),
@@ -221,7 +232,8 @@ class _SiteHttpAdapter:
            proxies=dict(proxies or {}),
            timeout=timeout,
        )
-        with request.response_manager(method=method, url=url) as response:
+        request_options = {} if json_body is None else {"json": dict(json_body)}
+        with request.response_manager(method=method, url=url, **request_options) as response:
            yield cast(Optional[SiteResponsePort], response)


--- a/tests/test_site_chain_ports.py
+++ b/tests/test_site_chain_ports.py
@@ -321,12 +321,14 @@ def test_initializer_rolls_back_all_ports_when_site_chain_publish_fails(
        site_module._site_ports_snapshot()


-def test_real_initializer_http_adapter_closes_response(monkeypatch) -> None:
+@pytest.mark.parametrize("json_body", [None, {}])
+def test_real_initializer_http_adapter_closes_response(monkeypatch, json_body) -> None:
    """
    组合根 HTTP adapter 必须在调用方返回后关闭 RequestUtils 响应。
    """
    response = _FakeResponse(200, text="ok")
    state = {"closed": False}
+    observed = {}

    class FakeRequest:
        """
@@ -339,10 +341,11 @@ def test_real_initializer_http_adapter_closes_response(monkeypatch) -> None:
            """

        @contextmanager
-        def response_manager(self, **_kwargs):
+        def response_manager(self, **kwargs):
            """
            退出上下文时记录响应已释放。
            """
+            observed.update(kwargs)
            try:
                yield response
            finally:
@@ -351,11 +354,15 @@ def test_real_initializer_http_adapter_closes_response(monkeypatch) -> None:
    monkeypatch.setattr(site_composition, "RequestUtils", FakeRequest)
    adapter = site_composition._SiteHttpAdapter()

-    with adapter.open(method="GET", url="https://close.test/") as opened:
+    with adapter.open(method="POST", url="https://close.test/", json_body=json_body) as opened:
        assert opened is response
        assert not state["closed"]

    assert state["closed"]
+    expected = {"method": "POST", "url": "https://close.test/"}
+    if json_body is not None:
+        expected["json"] = json_body
+    assert observed == expected


 def test_cookiecloud_adapter_closes_remote_response(monkeypatch) -> None:
--- a/tests/test_site_connectivity.py
+++ b/tests/test_site_connectivity.py
@@ -60,7 +60,7 @@ def test_mtorrent_schema_uses_profile_api(

    assert chain.test(site.url) == expected

-    indexer.assert_called_once_with("colapt.com")
+    indexer.assert_called_with("colapt.com")
    open_response.assert_called_once_with(
        method="POST",
        url="https://api.colapt.com/api/member/profile",
@@ -132,6 +132,167 @@ def test_other_schemas_keep_generic_connectivity_path(monkeypatch, schema, rende
    api_test.assert_not_called()


+@pytest.mark.parametrize("resource_key", ["schema", "parser"])
+@pytest.mark.parametrize("successful", [True, False])
+@pytest.mark.parametrize(
+    ("protocol", "method", "endpoint", "auth_header", "success_payload", "failure_payload"),
+    [
+        (
+            "MTorrent", "POST", "https://api.protocol.test/api/member/profile", "x-api-key",
+            {"code": "0", "data": {"id": "1"}}, {"code": "401", "data": {"id": "1"}},
+        ),
+        (
+            "HDDolby", "GET", "https://api.protocol.test/api/v1/user/data", "x-api-key",
+            {"status": 0, "data": [{"id": "1"}]}, {"status": 1},
+        ),
+        (
+            "RousiPro", "GET", "https://protocol.test/api/v1/profile", "Authorization",
+            {"code": 0, "data": {"id": "1"}}, {"code": 401},
+        ),
+        (
+            "SunnyPT", "GET", "https://api.protocol.test/api/v1/mp/profile", "X-API-Key",
+            {"code": 0, "data": {"download_allowed": True}}, {"code": 401},
+        ),
+        (
+            "Yema", "POST", "https://www.protocol.test/openApi/user/fetchBasicInfo.json", "Authorization",
+            {"success": True, "data": {"id": "1"}}, {"success": False, "errorMessage": "AuthKey 无效"},
+        ),
+        (
+            "Milkie", "GET", "https://www.protocol.test/api/v1/torrents?ps=1", "x-milkie-auth",
+            {"torrents": []}, {"error": "API Key 无效"},
+        ),
+    ],
+)
+def test_api_protocols_use_their_own_authentication(
+    monkeypatch, resource_key, successful, protocol, method, endpoint,
+    auth_header, success_payload, failure_payload,
+):
+    """
+    所有已适配的 API Key 协议按资源声明选择接口，并验证协议自己的成功判据。
+    """
+    chain = SiteChain()
+    site = SiteSnapshot(
+        id=1, name="协议兼容站", domain="protocol.test", url="https://www.protocol.test/",
+        apikey="test-key", cookie="unused-cookie", render=1,
+    )
+    chain.site_repository = Mock()
+    chain.site_repository.get_by_domain.return_value = site
+    # mTorrent 是搜索适配器的历史大小写，与资料 schema 不同。
+    resource_value = "mTorrent" if resource_key == "parser" and protocol == "MTorrent" else protocol
+    indexer = Mock(return_value={resource_key: resource_value})
+    monkeypatch.setattr("app.chain.site.SitesHelper", Mock(return_value=Mock(get_indexer=indexer)))
+    response = Mock(status_code=200)
+    response.json.return_value = success_payload if successful else failure_payload
+    context = MagicMock()
+    context.__enter__.return_value = response
+    open_response = Mock(return_value=context)
+    monkeypatch.setattr("app.chain.site._open_site_response", open_response)
+    generic_test = Mock(side_effect=AssertionError("API 协议不应检查首页 HTML"))
+    monkeypatch.setattr(SiteChain, "_SiteChain__test", generic_test)
+
+    state, _message = chain.test(site.url)
+
+    assert state is successful
+    open_response.assert_called_once()
+    request = open_response.call_args.kwargs
+    assert request["method"] == method
+    assert request["url"] == endpoint
+    expected_key = "Bearer test-key" if protocol == "RousiPro" else "test-key"
+    assert request["headers"][auth_header] == expected_key
+    assert "cookie" not in request
+    if protocol == "Yema":
+        assert request["json_body"] == {}
+    generic_test.assert_not_called()
+    context.__exit__.assert_called_once_with(None, None, None)
+
+
+@pytest.mark.parametrize(
+    ("protocol", "host_key", "expected_url"),
+    [
+        ("MTorrent", "api_domain", "https://custom-api.test:8443/api/member/profile"),
+        ("MTorrent", "api_host", "https://custom-api.test:8443/api/member/profile"),
+        ("SunnyPT", "api_url", "https://custom-api.test:8443/api/profile"),
+    ],
+)
+def test_api_probe_honors_resource_endpoint(monkeypatch, protocol, host_key, expected_url):
+    """
+    MTorrent 和 SunnyPT 探测与搜索使用同一资源 API 地址配置。
+    """
+    chain = SiteChain()
+    site = SiteSnapshot(id=1, name="独立 API", domain="custom.test", url="https://www.custom.test/", apikey="key")
+    chain.site_repository = Mock()
+    chain.site_repository.get_by_domain.return_value = site
+    indexer = Mock(return_value={"schema": protocol, host_key: "https://custom-api.test:8443/api"})
+    monkeypatch.setattr("app.chain.site.SitesHelper", Mock(return_value=Mock(get_indexer=indexer)))
+    context = MagicMock()
+    context.__enter__.return_value = Mock(status_code=200, json=Mock(return_value={"code": "0", "data": {"id": "1"}}))
+    open_response = Mock(return_value=context)
+    monkeypatch.setattr("app.chain.site._open_site_response", open_response)
+
+    assert chain.test(site.url) == (True, "连接成功")
+
+    assert open_response.call_args.kwargs["url"] == expected_url
+
+
+@pytest.mark.parametrize("apikey", [None, "test-auth-key"])
+def test_yema_registered_domain_prefers_authkey_and_preserves_legacy_cookie(monkeypatch, apikey):
+    """
+    Yema 主站已配置 AuthKey 时使用开放 API，否则兼容历史 Cookie 连接测试。
+    """
+    chain = SiteChain()
+    site = SiteSnapshot(
+        id=1, name="YemaPT", domain="yemapt.org", url="https://www.yemapt.org/",
+        cookie="session=test", apikey=apikey,
+    )
+    chain.site_repository = Mock()
+    chain.site_repository.get_by_domain.return_value = site
+    context = MagicMock()
+    context.__enter__.return_value = Mock(
+        status_code=200, json=Mock(return_value={"success": True, "data": {"id": "1"}}),
+    )
+    open_response = Mock(return_value=context)
+    monkeypatch.setattr("app.chain.site._open_site_response", open_response)
+
+    assert chain.test(site.url) == (True, "连接成功")
+
+    request = open_response.call_args.kwargs
+    if apikey:
+        assert request["method"] == "POST"
+        assert request["url"] == "https://www.yemapt.org/openApi/user/fetchBasicInfo.json"
+        assert request["headers"]["Authorization"] == apikey
+        assert "cookie" not in request
+    else:
+        assert request["method"] == "GET"
+        assert request["url"] == "https://www.yemapt.org/api/consumer/fetchSelfDetail"
+        assert request["cookie"] == "session=test"
+
+
+@pytest.mark.parametrize("resource_key, resource_value", [("schema", "TNode"), ("parser", "TNodeSpider")])
+def test_tnode_probe_uses_cookie_and_csrf_api(monkeypatch, resource_key, resource_value):
+    """
+    Cookie 认证的 TNode SPA 通过 CSRF 和当前用户 API 验证会话。
+    """
+    chain = SiteChain()
+    site = SiteSnapshot(id=1, name="TNode 兼容站", domain="tnode.test", url="https://tnode.test/", cookie="session=test")
+    chain.site_repository = Mock()
+    chain.site_repository.get_by_domain.return_value = site
+    indexer = Mock(return_value={resource_key: resource_value})
+    monkeypatch.setattr("app.chain.site.SitesHelper", Mock(return_value=Mock(get_indexer=indexer)))
+    page = MagicMock()
+    page.__enter__.return_value = Mock(status_code=200, text='<meta name="x-csrf-token" content="csrf-test">')
+    api = MagicMock()
+    api.__enter__.return_value = Mock(status_code=200, json=Mock(return_value={"data": {"id": "1"}}))
+    open_response = Mock(side_effect=[page, api])
+    monkeypatch.setattr("app.chain.site._open_site_response", open_response)
+
+    assert chain.test(site.url) == (True, "连接成功")
+
+    request = open_response.call_args.kwargs
+    assert request["url"] == "https://tnode.test/api/user/getInfo"
+    assert request["headers"]["X-CSRF-TOKEN"] == "csrf-test"
+    assert request["cookie"] == "session=test"
+
+
 def test_site_connectivity_records_result_without_injected_session(db, monkeypatch):
    """
    默认站点端口完成测试后应提交统计，而不是对空会话调用 execute。