fix(indexer): use M-Team subtitle search API

app/modules/indexer/__init__.py

@@ -51,6 +51,10 @@ _SPECIALIZED_SEARCH_ARGUMENTS = {
     "RousiPro": ("keyword", "mtype", "cat", "page"),
 }
 
+_SPECIALIZED_SUBTITLE_SEARCH_ARGUMENTS = {
+    "mTorrent": ("keyword", "page", "search_type"),
+}
+
 _UNKNOWN_SEARCH_FAILURE_MESSAGE = "站点请求或页面解析失败"
 
 # NexusPHP、Gazelle 和 Unit3d 覆盖大多数站点，其余模型多为专站适配，
@@ -332,7 +336,23 @@ class IndexerModule(_ModuleBase):
             return None
 
         search_word = IndexerModule.__clear_search_text(keyword)
+        parser_name = str(site.get("parser") or "")
+        parser_class = SPIDER_PARSER_CLASSES.get(parser_name)
         if search_type == "subtitles":
+            argument_names = _SPECIALIZED_SUBTITLE_SEARCH_ARGUMENTS.get(parser_name)
+            if parser_class and argument_names:
+                available_arguments = {
+                    "keyword": search_word,
+                    "page": page,
+                    "search_type": search_type,
+                }
+                return _IndexerSearchRequest(
+                    parser_class=parser_class,
+                    arguments=MappingProxyType({
+                        name: available_arguments[name]
+                        for name in argument_names
+                    }),
+                )
             return _IndexerSearchRequest(
                 parser_class=None,
                 arguments=MappingProxyType({
@@ -343,8 +363,6 @@ class IndexerModule(_ModuleBase):
                 }),
             )
 
-        parser_name = str(site.get("parser") or "")
-        parser_class = SPIDER_PARSER_CLASSES.get(parser_name)
         argument_names = _SPECIALIZED_SEARCH_ARGUMENTS.get(parser_name)
         if parser_class and argument_names:
             available_arguments = {
@@ -379,11 +397,12 @@ class IndexerModule(_ModuleBase):
     ) -> Tuple[bool, List[dict[str, Any]], Optional[str]]:
         """通过同步解析器执行已冻结的搜索请求，返回错误标志、结果与失败原因"""
         if request.parser_class:
+            spider = request.parser_class(site)
             error_flag, result = cast(
                 Tuple[bool, List[dict[str, Any]]],
-                request.parser_class(site).search(**request.arguments),
+                spider.search(**request.arguments),
             )
-            return error_flag, result, None
+            return error_flag, result, getattr(spider, "error_detail", None)
         return IndexerModule.__spider_search(**request.arguments)
 
     @staticmethod
@@ -393,11 +412,12 @@ class IndexerModule(_ModuleBase):
     ) -> Tuple[bool, List[dict[str, Any]], Optional[str]]:
         """通过异步解析器执行已冻结的搜索请求，返回错误标志、结果与失败原因"""
         if request.parser_class:
+            spider = request.parser_class(site)
             error_flag, result = cast(
                 Tuple[bool, List[dict[str, Any]]],
-                await request.parser_class(site).async_search(**request.arguments),
+                await spider.async_search(**request.arguments),
             )
-            return error_flag, result, None
+            return error_flag, result, getattr(spider, "error_detail", None)
         return await IndexerModule.__async_spider_search(**request.arguments)
 
     @staticmethod
@@ -504,6 +524,8 @@ class IndexerModule(_ModuleBase):
         result = []
         start_time = datetime.now()
         error_flag = False
+        error: Optional[Exception] = None
+        failure_detail: Optional[str] = None
 
         request = self.__create_search_request(
             site=site,
@@ -515,16 +537,32 @@ class IndexerModule(_ModuleBase):
             return []
 
         try:
-            error_flag, result, _ = self.__execute_search(site, request)
+            error_flag, result, failure_detail = self.__execute_search(site, request)
         except Exception as err:
+            error = err
             self.__log_search_error(site, "subtitles", err)
 
-        outcome = self.__create_search_outcome(start_time, error_flag, result)
+        outcome = self.__create_search_outcome(
+            start_time,
+            error_flag,
+            result,
+            error=error,
+            failure_detail=failure_detail,
+        )
         self.__indexer_statistic(
             site=site,
             error_flag=outcome.error_flag,
             seconds=outcome.seconds,
         )
+        report_site_search_outcome(
+            attempted=True,
+            outcome=(
+                _classify_search_failure(outcome.error)
+                if outcome.error_flag or outcome.error is not None
+                else "success"
+            ),
+            error=_search_failure_message(outcome),
+        )
         return self.__parse_subtitle_result(
             site=site,
             result_array=outcome.result,
@@ -612,6 +650,8 @@ class IndexerModule(_ModuleBase):
         result = []
         start_time = datetime.now()
         error_flag = False
+        error: Optional[Exception] = None
+        failure_detail: Optional[str] = None
 
         request = self.__create_search_request(
             site=site,
@@ -623,16 +663,32 @@ class IndexerModule(_ModuleBase):
             return []
 
         try:
-            error_flag, result, _ = await self.__async_execute_search(site, request)
+            error_flag, result, failure_detail = await self.__async_execute_search(site, request)
         except Exception as err:
+            error = err
             self.__log_search_error(site, "subtitles", err)
 
-        outcome = self.__create_search_outcome(start_time, error_flag, result)
+        outcome = self.__create_search_outcome(
+            start_time,
+            error_flag,
+            result,
+            error=error,
+            failure_detail=failure_detail,
+        )
         await self.__async_indexer_statistic(
             site=site,
             error_flag=outcome.error_flag,
             seconds=outcome.seconds,
         )
+        report_site_search_outcome(
+            attempted=True,
+            outcome=(
+                _classify_search_failure(outcome.error)
+                if outcome.error_flag or outcome.error is not None
+                else "success"
+            ),
+            error=_search_failure_message(outcome),
+        )
         return self.__parse_subtitle_result(
             site=site,
             result_array=outcome.result,

app/modules/indexer/spider/mtorrent.py

@@ -11,12 +11,13 @@ from app.runtime.log import logger
 from app.schemas.types import MediaType
 from app.adapters.network.http import RequestUtils, AsyncRequestUtils
 from app.domain import site as site_rules
+from app.foundation import size as size_tools
 from app.foundation import temporal as time_tools
 
 
 class MTorrentSpider:
     """
-    mTorrent API
+    mTorrent API 搜索与字幕下载适配器。
     """
     _indexerid = None
     _domain = None
@@ -26,11 +27,13 @@ class MTorrentSpider:
     _cookie = None
     _ua = None
     _size = 100
-    _searchurl = "https://api.%s/api/torrent/search"
-    _downloadurl = "https://api.%s/api/torrent/genDlToken"
-    _subtitle_list_url = "https://api.%s/api/subtitle/list"
-    _subtitle_genlink_url = "https://api.%s/api/subtitle/genlink"
-    _subtitle_download_url ="https://api.%s/api/subtitle/dlV2?credential=%s"
+    _subtitle_size = 20
+    _api_domain = None
+    _searchurl = "https://%s/api/torrent/search"
+    _downloadurl = "https://%s/api/torrent/genDlToken"
+    _subtitle_list_url = "https://%s/api/subtitle/list"
+    _subtitle_genlink_url = "https://%s/api/subtitle/genlink"
+    _subtitle_download_url ="https://%s/api/subtitle/dlV2?credential=%s"
     _pageurl = "%sdetail/%s"
     _timeout = 15
 
@@ -45,6 +48,8 @@ class MTorrentSpider:
     _apikey = None
     # JWT Token
     _token = None
+    # 最近一次字幕搜索失败原因
+    error_detail: Optional[str] = None
 
     # 标签
     _labels = {
@@ -72,7 +77,8 @@ class MTorrentSpider:
             self._indexerid = indexer.get('id')
             self._url = indexer.get('domain')
             self._domain = site_rules.extract_domain(self._url)
-            self._searchurl = self._searchurl % self._domain
+            self._api_domain = self.__resolve_api_domain(indexer)
+            self._searchurl = self._searchurl % self._api_domain
             self._name = indexer.get('name')
             if indexer.get('proxy'):
                 self._proxy = get_runtime_setting('PROXY')
@@ -81,6 +87,31 @@ class MTorrentSpider:
             self._apikey = indexer.get('apikey')
             self._token = indexer.get('token')
             self._timeout = indexer.get('timeout') or 15
+            self.error_detail = None
+
+    @staticmethod
+    def __resolve_api_domain(indexer: dict) -> str:
+        """
+        从站点资源配置解析 M-Team API 主机。
+
+        资源可以直接提供 ``api_domain`` 或 ``api_host``；未提供时沿用
+        M-Team 的 ``api.<注册域名>`` 约定，避免把旧网页接口写死在代码中。
+        """
+        configured = indexer.get("api_domain") or indexer.get("api_host")
+        if configured:
+            value = str(configured).strip()
+            parsed = urlparse(value if "://" in value else f"//{value}")
+            host = parsed.netloc or parsed.path.split("/", 1)[0]
+            if host:
+                return host
+
+        domain_url = str(indexer.get("domain") or "").strip()
+        parsed = urlparse(domain_url if "://" in domain_url else f"//{domain_url}")
+        host = parsed.netloc or parsed.path.split("/", 1)[0]
+        if host.startswith("api."):
+            return host
+        registered_domain = site_rules.extract_domain(domain_url)
+        return f"api.{registered_domain}" if registered_domain else host
 
     def __get_params(self, keyword: str, mtype: MediaType = None, page: Optional[int] = 0) -> dict:
         """
@@ -179,10 +210,13 @@ class MTorrentSpider:
         logger.warn(f"{self._name} 搜索失败，无法连接 {self._domain}")
         return True, []
 
-    def search(self, keyword: str, mtype: MediaType = None, page: Optional[int] = 0) -> Tuple[bool, List[dict]]:
+    def search(self, keyword: str, mtype: MediaType = None, page: Optional[int] = 0,
+               search_type: Optional[str] = None) -> Tuple[bool, List[dict]]:
         """
-        搜索
+        搜索种子或字幕。
         """
+        if search_type == "subtitles":
+            return self.search_subtitles(keyword=keyword, page=page)
         # 检查ApiKey
         if not self._apikey:
             return True, []
@@ -203,10 +237,13 @@ class MTorrentSpider:
         ).post_res(url=self._searchurl, json=params)
         return self.__process_response(res)
 
-    async def async_search(self, keyword: str, mtype: MediaType = None, page: Optional[int] = 0) -> Tuple[bool, List[dict]]:
+    async def async_search(self, keyword: str, mtype: MediaType = None, page: Optional[int] = 0,
+                           search_type: Optional[str] = None) -> Tuple[bool, List[dict]]:
         """
-        搜索
+        异步搜索种子或字幕。
         """
+        if search_type == "subtitles":
+            return await self.async_search_subtitles(keyword=keyword, page=page)
         # 检查ApiKey
         if not self._apikey:
             return True, []
@@ -227,6 +264,200 @@ class MTorrentSpider:
         ).post_res(url=self._searchurl, json=params)
         return self.__process_response(res)
 
+    def __get_subtitle_params(self, keyword: str, page: Optional[int] = 0) -> dict:
+        """构造 M-Team 字幕搜索请求体。"""
+        return {
+            "keyword": keyword,
+            "pageNumber": int(page) + 1,
+            "pageSize": self._subtitle_size,
+        }
+
+    @staticmethod
+    def __get_subtitle_rows(payload: Any) -> List[dict]:
+        """兼容 M-Team 字幕接口分页数据的不同包装层级。"""
+        if not isinstance(payload, dict):
+            return []
+        data = payload.get("data")
+        if isinstance(data, list):
+            return [item for item in data if isinstance(item, dict)]
+        if isinstance(data, dict):
+            for key in ("data", "items", "records", "list"):
+                rows = data.get(key)
+                if isinstance(rows, list):
+                    return [item for item in rows if isinstance(item, dict)]
+        return []
+
+    @staticmethod
+    def __get_int(value: Any, default: int = 0) -> int:
+        """把接口中的数字字段安全转换为整数。"""
+        try:
+            return default if value is None else int(value)
+        except (TypeError, ValueError):
+            return default
+
+    def __parse_subtitle_rows(self, results: List[dict]) -> List[dict]:
+        """
+        解析 M-Team 字幕结果并为每条记录生成可直接下载的签名地址。
+        """
+        subtitles = []
+        for result in results:
+            subtitle_id = result.get("id")
+            if subtitle_id is None:
+                continue
+            torrent = result.get("torrent")
+            torrent_id = torrent.get("id") if isinstance(torrent, dict) else torrent
+            link = self.__subtitle_genlink(str(subtitle_id))
+            if not link:
+                continue
+            title = result.get("name") or result.get("filename") or str(subtitle_id)
+            file_name = result.get("filename") or result.get("fileName")
+            created = result.get("createdDate") or result.get("lastModifiedDate")
+            if isinstance(created, (int, float)) or str(created or "").isdigit():
+                pubdate = time_tools.format_timestamp(created)
+            else:
+                pubdate = time_tools.normalize_datetime(str(created)) if created else None
+            subtitles.append({
+                "title": title,
+                "description": title,
+                "enclosure": link,
+                "page_url": self._pageurl % (self._url, torrent_id) if torrent_id else None,
+                "language": result.get("lang") or result.get("language") or "",
+                "size": size_tools.parse_size(result.get("size") or 0),
+                "pubdate": pubdate,
+                "grabs": self.__get_int(result.get("hits")),
+                "uploader": result.get("author"),
+                "torrent_id": str(torrent_id) if torrent_id is not None else None,
+                "subtitle_id": str(subtitle_id),
+                "file_name": file_name,
+            })
+        return subtitles
+
+    def __process_subtitle_response(self, res: Any) -> Tuple[bool, List[dict[str, Any]]]:
+        """判定字幕搜索响应并转换为 MoviePilot 字幕字段。"""
+        self.error_detail = None
+        if res is None:
+            self.error_detail = f"无法连接 {self._api_domain or self._domain}"
+            logger.warn(f"{self._name} 字幕搜索失败，{self.error_detail}")
+            return True, []
+        if res.status_code != 200:
+            self.error_detail = f"HTTP {res.status_code}"
+            logger.warn(f"{self._name} 字幕搜索失败，错误码：{res.status_code}")
+            return True, []
+        try:
+            payload = res.json()
+        except (TypeError, ValueError):
+            self.error_detail = "响应不是有效 JSON"
+            logger.warn(f"{self._name} 字幕搜索失败，{self.error_detail}")
+            return True, []
+        if not isinstance(payload, dict) or self.__get_int(payload.get("code"), -1) != 0:
+            self.error_detail = str(payload.get("message") or "接口返回失败") if isinstance(payload, dict) else "接口返回失败"
+            logger.warn(f"{self._name} 字幕搜索失败，返回：{self.error_detail}")
+            return True, []
+        return False, self.__parse_subtitle_rows(self.__get_subtitle_rows(payload))
+
+    def search_subtitles(self, keyword: str, page: Optional[int] = 0) -> Tuple[bool, List[dict]]:
+        """通过 M-Team 字幕 API 搜索字幕。"""
+        if not self._apikey:
+            self.error_detail = "未配置 API Key"
+            return True, []
+        params = self.__get_subtitle_params(keyword, page)
+        res = RequestUtils(
+            headers={
+                "Content-Type": "application/json",
+                "User-Agent": f"{self._ua}",
+                "x-api-key": self._apikey,
+            },
+            proxies=self._proxy,
+            timeout=self._timeout,
+        ).post_res(url=f"https://{self._api_domain}/api/subtitle/search", json=params)
+        return self.__process_subtitle_response(res)
+
+    async def __async_subtitle_genlink(self, subtitle_id: str) -> Optional[str]:
+        """异步获取字幕下载链接。"""
+        url = self._subtitle_genlink_url % self._api_domain
+        res = await AsyncRequestUtils(
+            headers={
+                "Accept": "application/json, text/plain, */*",
+                "User-Agent": f"{self._ua}",
+                "x-api-key": self._apikey,
+            },
+            proxies=self._proxy,
+            timeout=self._timeout,
+        ).post_res(url, data={"id": subtitle_id})
+        if res and res.status_code == 200:
+            result = res.json()
+            if self.__get_int(result.get("code"), -1) == 0 and isinstance(result.get("data"), str):
+                return self._subtitle_download_url % (self._api_domain, result["data"])
+            return None
+        return None
+
+    async def __async_parse_subtitle_rows(self, results: List[dict]) -> List[dict]:
+        """异步解析 M-Team 字幕结果并生成下载地址。"""
+        subtitles = []
+        for result in results:
+            subtitle_id = result.get("id")
+            if subtitle_id is None:
+                continue
+            torrent = result.get("torrent")
+            torrent_id = torrent.get("id") if isinstance(torrent, dict) else torrent
+            link = await self.__async_subtitle_genlink(str(subtitle_id))
+            if not link:
+                continue
+            title = result.get("name") or result.get("filename") or str(subtitle_id)
+            file_name = result.get("filename") or result.get("fileName")
+            created = result.get("createdDate") or result.get("lastModifiedDate")
+            if isinstance(created, (int, float)) or str(created or "").isdigit():
+                pubdate = time_tools.format_timestamp(created)
+            else:
+                pubdate = time_tools.normalize_datetime(str(created)) if created else None
+            subtitles.append({
+                "title": title,
+                "description": title,
+                "enclosure": link,
+                "page_url": self._pageurl % (self._url, torrent_id) if torrent_id else None,
+                "language": result.get("lang") or result.get("language") or "",
+                "size": size_tools.parse_size(result.get("size") or 0),
+                "pubdate": pubdate,
+                "grabs": self.__get_int(result.get("hits")),
+                "uploader": result.get("author"),
+                "torrent_id": str(torrent_id) if torrent_id is not None else None,
+                "subtitle_id": str(subtitle_id),
+                "file_name": file_name,
+            })
+        return subtitles
+
+    async def async_search_subtitles(self, keyword: str, page: Optional[int] = 0) -> Tuple[bool, List[dict]]:
+        """异步通过 M-Team 字幕 API 搜索字幕。"""
+        if not self._apikey:
+            self.error_detail = "未配置 API Key"
+            return True, []
+        params = self.__get_subtitle_params(keyword, page)
+        res = await AsyncRequestUtils(
+            headers={
+                "Content-Type": "application/json",
+                "User-Agent": f"{self._ua}",
+                "x-api-key": self._apikey,
+            },
+            proxies=self._proxy,
+            timeout=self._timeout,
+        ).post_res(url=f"https://{self._api_domain}/api/subtitle/search", json=params)
+        self.error_detail = None
+        if res is None:
+            self.error_detail = f"无法连接 {self._api_domain or self._domain}"
+            return True, []
+        if res.status_code != 200:
+            self.error_detail = f"HTTP {res.status_code}"
+            return True, []
+        try:
+            payload = res.json()
+        except (TypeError, ValueError):
+            self.error_detail = "响应不是有效 JSON"
+            return True, []
+        if not isinstance(payload, dict) or self.__get_int(payload.get("code"), -1) != 0:
+            self.error_detail = str(payload.get("message") or "接口返回失败") if isinstance(payload, dict) else "接口返回失败"
+            return True, []
+        return False, await self.__async_parse_subtitle_rows(self.__get_subtitle_rows(payload))
+
     @staticmethod
     def __find_imdbid(imdb: str) -> str:
         """
@@ -272,7 +503,7 @@ class MTorrentSpider:
         """
         获取下载链接，返回base64编码的json字符串及URL
         """
-        url = self._downloadurl % self._domain
+        url = self._downloadurl % self._api_domain
         params = {
             'method': 'post',
             'cookie': False,
@@ -335,7 +566,7 @@ class MTorrentSpider:
         :return: 字幕ID
         :rtype: List[str] | None
         """
-        url = self._subtitle_list_url % self._domain
+        url = self._subtitle_list_url % self._api_domain
         # 发送请求
         res = RequestUtils(
             headers={
@@ -371,7 +602,7 @@ class MTorrentSpider:
         :return: 下载链接
         :rtype: str | None
         """
-        url = self._subtitle_genlink_url % self._domain
+        url = self._subtitle_genlink_url % self._api_domain
         # 发送请求
         res = RequestUtils(
             headers={
@@ -385,7 +616,7 @@ class MTorrentSpider:
         if res and res.status_code == 200:
             result = res.json()
             if int(result.get("code", -1)) == 0 and isinstance(result.get("data"), str):
-                return self._subtitle_download_url % (self._domain, result["data"])
+                return self._subtitle_download_url % (self._api_domain, result["data"])
             else:
                 logger.warn(
                     f'{self._name} 获取字幕下载链接失败，返回：{result.get("message", "未知")}'

tests/test_indexer_sync_async_parity.py

@@ -386,6 +386,54 @@ def test_subtitle_search_uses_one_request_and_preserves_order(monkeypatch):
     assert events[-2][0:3] == ("statistic", "Subtitle", True)
 
 
+def test_mtorrent_subtitle_search_uses_specialized_parser(monkeypatch):
+    """mTorrent 字幕搜索应走专用 API 解析器，而不是旧网页解析器。"""
+    events = []
+    calls = []
+    _install_search_observers(monkeypatch, events)
+
+    class FakeSpider:
+        """记录字幕专用解析器收到的参数。"""
+
+        error_detail = None
+
+        def __init__(self, site):
+            """保存站点配置。"""
+            self.site = site
+
+        def search(self, **kwargs):
+            """记录同步字幕搜索调用。"""
+            calls.append(("sync", kwargs))
+            events.append(("io", kwargs))
+            return False, [{"title": "subtitle"}]
+
+        async def async_search(self, **kwargs):
+            """记录异步字幕搜索调用。"""
+            calls.append(("async", kwargs))
+            events.append(("io", kwargs))
+            return False, [{"title": "subtitle"}]
+
+    monkeypatch.setitem(indexer_module.SPIDER_PARSER_CLASSES, "mTorrent", FakeSpider)
+    site = {
+        "id": 1,
+        "name": "M-Team",
+        "parser": "mTorrent",
+        "subtitles": {"search": {}},
+    }
+    module = object.__new__(IndexerModule)
+
+    sync_result = module.search_subtitles(site=site, keyword="Raw.Keyword", page=4)
+    async_result = asyncio.run(module.async_search_subtitles(site=site, keyword="Raw.Keyword", page=4))
+
+    expected_arguments = {
+        "keyword": "clean keyword",
+        "page": 4,
+        "search_type": "subtitles",
+    }
+    assert sync_result == async_result == [{"title": "subtitle"}]
+    assert calls == [("sync", expected_arguments), ("async", expected_arguments)]
+
+
 def test_disabled_subtitle_search_skips_all_follow_up_work(monkeypatch):
     """未启用字幕能力时两种入口都不得触发许可检查、I/O 或统计"""
     events = []

tests/test_mtorrent_spider.py

@@ -1,6 +1,9 @@
 # -*- coding: utf-8 -*-
 # pylint: disable=no-name-in-module
 
+import asyncio
+from types import SimpleNamespace
+
 import pytest
 
 from app.modules.indexer.spider import mtorrent as mtorrent_module
@@ -41,6 +44,16 @@ def test_movie_search_keeps_movie_categories(mteam_spider):
     assert params["categories"] == MTorrentSpider._movie_category
 
 
+def test_api_domain_can_come_from_site_resource(mteam_spider):
+    """站点资源提供 API 主机时应优先使用该配置。"""
+    indexer = {
+        **_build_indexer(),
+        "api_domain": "https://subtitle-api.example/api",
+    }
+
+    assert mteam_spider._MTorrentSpider__resolve_api_domain(indexer) == "subtitle-api.example"
+
+
 def test_parse_result_marks_music_torrents(mteam_spider):
     """音乐分区种子应标记为音乐媒体类型，供音乐搜索链路筛选。"""
     results = mteam_spider._MTorrentSpider__parse_result([
@@ -56,3 +69,101 @@ def test_parse_result_marks_music_torrents(mteam_spider):
         MediaType.MOVIE.value,
         MediaType.UNKNOWN.value,
     ]
+
+
+def test_subtitle_search_uses_api_and_generates_download_links(monkeypatch, mteam_spider):
+    """M-Team 字幕搜索应使用 JSON API，并把字幕 ID 转换为下载链接。"""
+    calls = []
+
+    class FakeRequest:
+        """记录请求适配器收到的配置。"""
+
+        def __init__(self, **kwargs):
+            """保存请求头和超时配置。"""
+            self.kwargs = kwargs
+
+        def post_res(self, url, json=None, data=None):
+            """记录请求地址与 JSON 请求体。"""
+            calls.append((self.kwargs, url, json, data))
+            return SimpleNamespace(
+                status_code=200,
+                json=lambda: {
+                    "code": 0,
+                    "data": {
+                        "data": [{
+                            "id": "subtitle-1",
+                            "name": "Kill Bill 简中",
+                            "filename": "Kill.Bill.zh.srt",
+                            "torrent": "torrent-1",
+                            "size": "12 KB",
+                            "lang": "简中",
+                            "hits": 3,
+                            "author": "uploader",
+                            "createdDate": "2026-01-01 00:00:00",
+                        }],
+                    },
+                },
+            )
+
+    monkeypatch.setattr(mtorrent_module, "RequestUtils", FakeRequest)
+    monkeypatch.setattr(
+        mteam_spider,
+        "_MTorrentSpider__subtitle_genlink",
+        lambda subtitle_id: f"https://api.m-team.io/api/subtitle/dlV2?credential={subtitle_id}",
+    )
+
+    error, subtitles = mteam_spider.search_subtitles("杀死比尔", page=2)
+
+    assert error is False
+    assert subtitles[0]["subtitle_id"] == "subtitle-1"
+    assert subtitles[0]["enclosure"].endswith("credential=subtitle-1")
+    assert subtitles[0]["size"] == 12 * 1024
+    request_kwargs, url, payload, data = calls[0]
+    assert data is None
+    assert url == "https://api.m-team.io/api/subtitle/search"
+    assert payload == {"keyword": "杀死比尔", "pageNumber": 3, "pageSize": 20}
+    assert request_kwargs["headers"]["x-api-key"] == "mteam-secret"
+
+
+def test_async_subtitle_search_uses_same_api_contract(monkeypatch, mteam_spider):
+    """M-Team 字幕搜索的异步入口应保持同步入口的地址和请求体。"""
+    calls = []
+
+    class FakeAsyncRequest:
+        """记录异步请求适配器收到的配置。"""
+
+        def __init__(self, **kwargs):
+            """保存请求头和超时配置。"""
+            self.kwargs = kwargs
+
+        async def post_res(self, url, json=None, data=None):
+            """记录异步请求地址与 JSON 请求体。"""
+            calls.append((self.kwargs, url, json, data))
+            return SimpleNamespace(
+                status_code=200,
+                json=lambda: {"code": 0, "data": {"data": []}},
+            )
+
+    monkeypatch.setattr(mtorrent_module, "AsyncRequestUtils", FakeAsyncRequest)
+
+    error, subtitles = asyncio.run(mteam_spider.async_search_subtitles("杀死比尔", page=1))
+
+    assert error is False
+    assert subtitles == []
+    request_kwargs, url, payload, data = calls[0]
+    assert data is None
+    assert url == "https://api.m-team.io/api/subtitle/search"
+    assert payload == {"keyword": "杀死比尔", "pageNumber": 2, "pageSize": 20}
+    assert request_kwargs["headers"]["x-api-key"] == "mteam-secret"
+
+
+@pytest.mark.parametrize("status_code", [401, 403, 404, 405])
+def test_subtitle_search_reports_http_errors(mteam_spider, status_code):
+    """字幕接口鉴权或路径错误应保留明确的 HTTP 状态原因。"""
+    error, subtitles = mteam_spider._MTorrentSpider__process_subtitle_response(
+        SimpleNamespace(status_code=status_code, json=lambda: {})
+    )
+
+    assert error is True
+    assert subtitles == []
+    assert mteam_spider.error_detail == f"HTTP {status_code}"