diff --git a/app/modules/indexer/__init__.py b/app/modules/indexer/__init__.py index a3a81ba13..b8a4b05c1 100644 --- a/app/modules/indexer/__init__.py +++ b/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, diff --git a/app/modules/indexer/spider/mtorrent.py b/app/modules/indexer/spider/mtorrent.py index 112cce65c..3397d52ab 100644 --- a/app/modules/indexer/spider/mtorrent.py +++ b/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", "未知")}' diff --git a/tests/test_indexer_sync_async_parity.py b/tests/test_indexer_sync_async_parity.py index 634fde29a..8859bfdfe 100644 --- a/tests/test_indexer_sync_async_parity.py +++ b/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 = [] diff --git a/tests/test_mtorrent_spider.py b/tests/test_mtorrent_spider.py index 8e5bac6be..baecd9cea 100644 --- a/tests/test_mtorrent_spider.py +++ b/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}"