From f5870d8fb040d822a1dbfd6cf57bda5a034136e8 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 29 Aug 2026 19:14:38 +0800 Subject: [PATCH] refactor(search): unify sync async decisions --- app/chain/search/media.py | 201 ++++++++------- app/chain/search/title.py | 135 ++++++----- tests/test_search_sync_async_parity.py | 324 +++++++++++++++++++++++++ 3 files changed, 517 insertions(+), 143 deletions(-) create mode 100644 tests/test_search_sync_async_parity.py diff --git a/app/chain/search/media.py b/app/chain/search/media.py index eb36d456a..6cbff0ddd 100644 --- a/app/chain/search/media.py +++ b/app/chain/search/media.py @@ -3,7 +3,7 @@ import asyncio import random import time -from typing import Any, AsyncIterator, Dict, List, Optional, cast +from typing import Any, AsyncIterator, Dict, List, Optional, Tuple, cast from app.chain.media import MediaChain from app.chain.search.contract import _SearchOwnerBase as _SearchOwnerBase @@ -19,6 +19,80 @@ from app.schemas.types import ( ) +def _build_id_search_params( + media_source: MediaSource, + media_id: str, + mtype: Optional[MediaType], + area: Optional[str], + season: Optional[int], + sites: Optional[List[int]], + music_type: Optional[str], +) -> Tuple[Dict[str, Any], Dict[str, Any]]: + """构造同步、异步和流式 ID 搜索共享的识别与缓存参数。""" + recognition_params = { + "media_source": media_source, + "media_id": media_id, + "mtype": mtype, + "music_type": music_type, + } + cache_params = { + **recognition_params, + "area": area, + "season": season, + "sites": sites, + } + return recognition_params, cache_params + + +def _build_missing_media_map( + mediainfo: MediaInfo, + season: Optional[int], +) -> Optional[Dict[str, Dict[int, NotExistMediaInfo]]]: + """将指定季转换为统一的缺集搜索映射。""" + if season is None: + return None + resolved_source, resolved_id = resolve_media_identity(media=mediainfo) + return { + build_media_key(resolved_source, resolved_id): { + season: NotExistMediaInfo(episodes=[]) + } + } + + +def _normalize_media_search_input(mediainfo: MediaInfo) -> MediaInfo: + """归一化非 TMDB 输入标题，并保留调用方对象由外层复制的所有权约束。""" + if not mediainfo.tmdb_id: + meta = MetaInfo(title=mediainfo.title) + mediainfo.title = meta.name + mediainfo.season = cast(int, meta.begin_season) + return mediainfo + + +def _should_stop_keyword_search( + search_multiple_name: bool, + torrents: List[TorrentInfo], +) -> bool: + """统一判断首个有效关键字结果是否终止后续搜索。""" + return not search_multiple_name and bool(torrents) + + +def _build_candidate_contexts( + mediainfo: MediaInfo, + torrents: List[TorrentInfo], +) -> List[Context]: + """将流式站点候选映射为尚未精确过滤的搜索上下文。""" + return [ + Context( + meta_info=MetaInfo(title=torrent.title, subtitle=torrent.description), + media_info=mediainfo, + torrent_info=torrent, + resource_source="search", + media_info_is_target=True, + ) + for torrent in torrents + ] + + class SearchMediaOwner(_SearchOwnerBase): """精确媒体搜索与同步异步编排 owner。""" @@ -44,31 +118,24 @@ class SearchMediaOwner(_SearchOwnerBase): :param sites: 站点ID列表 :param cache_local: 是否缓存到本地 """ - if cache_local: - self.cancel_ai_recommend() - self.save_last_search_params( - media_source=media_source, - media_id=media_id, - mtype=mtype, - area=area, - season=season, - sites=sites, - music_type=music_type, - ) - # 音乐统一在 MediaChain.recognize_media 内按固定来源路由 - mediainfo = MediaChain().recognize_media( + recognition_params, cache_params = _build_id_search_params( media_source=media_source, media_id=media_id, mtype=mtype, + area=area, + season=season, + sites=sites, music_type=music_type, ) + if cache_local: + self.cancel_ai_recommend() + self.save_last_search_params(**cache_params) + # 音乐统一在 MediaChain.recognize_media 内按固定来源路由 + mediainfo = MediaChain().recognize_media(**recognition_params) if not mediainfo: logger.error(f"{self._build_search_keyword(media_source, media_id)} 媒体信息识别失败！") return [] - no_exists: Optional[Dict[str, Dict[int, NotExistMediaInfo]]] = None - if season is not None: - resolved_source, resolved_id = resolve_media_identity(media=mediainfo) - no_exists = {build_media_key(resolved_source, resolved_id): {season: NotExistMediaInfo(episodes=[])}} + no_exists = _build_missing_media_map(mediainfo, season) results = self.process(mediainfo=mediainfo, sites=sites, area=area, no_exists=no_exists) # 保存到本地文件 if cache_local: @@ -97,31 +164,24 @@ class SearchMediaOwner(_SearchOwnerBase): :param sites: 站点ID列表 :param cache_local: 是否缓存到本地 """ - if cache_local: - self.cancel_ai_recommend() - await self.async_save_last_search_params( - media_source=media_source, - media_id=media_id, - mtype=mtype, - area=area, - season=season, - sites=sites, - music_type=music_type, - ) - # 音乐统一在 MediaChain.async_recognize_media 内按固定来源路由 - mediainfo = await MediaChain().async_recognize_media( + recognition_params, cache_params = _build_id_search_params( media_source=media_source, media_id=media_id, mtype=mtype, + area=area, + season=season, + sites=sites, music_type=music_type, ) + if cache_local: + self.cancel_ai_recommend() + await self.async_save_last_search_params(**cache_params) + # 音乐统一在 MediaChain.async_recognize_media 内按固定来源路由 + mediainfo = await MediaChain().async_recognize_media(**recognition_params) if not mediainfo: logger.error(f"{self._build_search_keyword(media_source, media_id)} 媒体信息识别失败！") return [] - no_exists: Optional[Dict[str, Dict[int, NotExistMediaInfo]]] = None - if season is not None: - resolved_source, resolved_id = resolve_media_identity(media=mediainfo) - no_exists = {build_media_key(resolved_source, resolved_id): {season: NotExistMediaInfo(episodes=[])}} + no_exists = _build_missing_media_map(mediainfo, season) results = await self.async_process(mediainfo=mediainfo, sites=sites, area=area, no_exists=no_exists) # 保存到本地文件 if cache_local: @@ -142,33 +202,26 @@ class SearchMediaOwner(_SearchOwnerBase): """ 根据数据源媒体 ID 渐进式搜索资源，先返回站点原始候选，再返回过滤匹配后的最终结果 """ - if cache_local: - self.cancel_ai_recommend() - await self.async_save_last_search_params( - media_source=media_source, - media_id=media_id, - mtype=mtype, - area=area, - season=season, - sites=sites, - music_type=music_type, - ) - # 音乐统一在 MediaChain.async_recognize_media 内按固定来源路由 - mediainfo = await MediaChain().async_recognize_media( + recognition_params, cache_params = _build_id_search_params( media_source=media_source, media_id=media_id, mtype=mtype, + area=area, + season=season, + sites=sites, music_type=music_type, ) + if cache_local: + self.cancel_ai_recommend() + await self.async_save_last_search_params(**cache_params) + # 音乐统一在 MediaChain.async_recognize_media 内按固定来源路由 + mediainfo = await MediaChain().async_recognize_media(**recognition_params) if not mediainfo: logger.error(f"{self._build_search_keyword(media_source, media_id)} 媒体信息识别失败！") yield {"type": "error", "success": False, "message": "媒体信息识别失败"} return - no_exists: Optional[Dict[str, Dict[int, NotExistMediaInfo]]] = None - if season is not None: - resolved_source, resolved_id = resolve_media_identity(media=mediainfo) - no_exists = {build_media_key(resolved_source, resolved_id): {season: NotExistMediaInfo(episodes=[])}} + no_exists = _build_missing_media_map(mediainfo, season) contexts: List[Context] = [] async for event in self.async_process_stream(mediainfo=mediainfo, sites=sites, area=area, no_exists=no_exists): @@ -215,12 +268,7 @@ class SearchMediaOwner(_SearchOwnerBase): ), ) - mediainfo = self._copy_media_input(mediainfo) - # 豆瓣标题处理 - if not mediainfo.tmdb_id: - meta = MetaInfo(title=mediainfo.title) - mediainfo.title = meta.name - mediainfo.season = cast(int, meta.begin_season) + mediainfo = _normalize_media_search_input(self._copy_media_input(mediainfo)) logger.info(f"开始搜索资源，关键词：{keyword or mediainfo.title} ...") # 补充媒体信息 @@ -266,7 +314,9 @@ class SearchMediaOwner(_SearchOwnerBase): torrents.extend(results) # 有结果则停止 - if not self.runtime_config.search_multiple_name and torrents: + if _should_stop_keyword_search( + self.runtime_config.search_multiple_name, torrents + ): logger.info(f"共搜索到 {len(torrents)} 个资源，停止搜索") break @@ -319,12 +369,7 @@ class SearchMediaOwner(_SearchOwnerBase): ), ) - mediainfo = self._copy_media_input(mediainfo) - # 豆瓣标题处理 - if not mediainfo.tmdb_id: - meta = MetaInfo(title=mediainfo.title) - mediainfo.title = meta.name - mediainfo.season = cast(int, meta.begin_season) + mediainfo = _normalize_media_search_input(self._copy_media_input(mediainfo)) logger.info(f"开始搜索资源，关键词：{keyword or mediainfo.title} ...") # 补充媒体信息 @@ -367,7 +412,9 @@ class SearchMediaOwner(_SearchOwnerBase): ) search_count += 1 # 未开启多名称搜索时，有结果则停止 - if not self.runtime_config.search_multiple_name and torrents: + if _should_stop_keyword_search( + self.runtime_config.search_multiple_name, torrents + ): logger.info(f"共搜索到 {len(torrents)} 个资源，停止搜索") break @@ -412,12 +459,7 @@ class SearchMediaOwner(_SearchOwnerBase): yield event return - mediainfo = self._copy_media_input(mediainfo) - # 豆瓣标题处理 - if not mediainfo.tmdb_id: - meta = MetaInfo(title=mediainfo.title) - mediainfo.title = meta.name - mediainfo.season = cast(int, meta.begin_season) + mediainfo = _normalize_media_search_input(self._copy_media_input(mediainfo)) logger.info(f"开始渐进式搜索资源，关键词：{keyword or mediainfo.title} ...") # 补充媒体信息 @@ -454,16 +496,7 @@ class SearchMediaOwner(_SearchOwnerBase): ): result = event.pop("items", []) or [] torrents.extend(result) - batch_contexts = [ - Context( - meta_info=MetaInfo(title=torrent.title, subtitle=torrent.description), - media_info=mediainfo, - torrent_info=torrent, - resource_source="search", - media_info_is_target=True, - ) - for torrent in result - ] + batch_contexts = _build_candidate_contexts(mediainfo, result) candidate_contexts.extend(batch_contexts) yield { **event, @@ -474,7 +507,9 @@ class SearchMediaOwner(_SearchOwnerBase): } search_count += 1 - if not self.runtime_config.search_multiple_name and torrents: + if _should_stop_keyword_search( + self.runtime_config.search_multiple_name, torrents + ): logger.info(f"共搜索到 {len(torrents)} 个资源，停止搜索") break diff --git a/app/chain/search/title.py b/app/chain/search/title.py index fa95f8e4d..76a54a177 100644 --- a/app/chain/search/title.py +++ b/app/chain/search/title.py @@ -1,6 +1,6 @@ """标题搜索入口与标题候选过滤 owner。""" -from typing import Any, AsyncIterator, Callable, Dict, List, Optional, cast +from typing import Any, AsyncIterator, Callable, Dict, List, Optional, Tuple, cast from app.application.configuration import ( get_configured_system_config, @@ -17,6 +17,46 @@ from app.schemas.types import ( ) +def _build_title_search_params( + title: str, + page: Optional[int], + sites: Optional[List[int]], + mtype: Optional[MediaType], + include_empty_type: bool = False, +) -> Tuple[Dict[str, Any], Dict[str, Any]]: + """构造标题搜索共享的 provider 与缓存参数。""" + provider_params: Dict[str, Any] = { + "keyword": title, + "sites": sites, + "page": page, + } + if mtype is not None or include_empty_type: + provider_params["mtype"] = mtype + cache_params = { + "keyword": title, + "mtype": mtype, + "area": "title", + "sites": sites, + } + return provider_params, cache_params + + +def _build_title_contexts( + torrents: List[TorrentInfo], + mtype: Optional[MediaType], + build_meta: Callable[[TorrentInfo, Optional[MediaType]], Any], +) -> List[Context]: + """将过滤后的标题候选统一投影为搜索上下文。""" + return [ + Context( + meta_info=build_meta(torrent, mtype), + torrent_info=torrent, + resource_source="search", + ) + for torrent in torrents + ] + + class SearchTitleOwner(_SearchOwnerBase): """标题搜索入口与标题候选过滤 owner。""" @@ -38,27 +78,21 @@ class SearchTitleOwner(_SearchOwnerBase): :param mtype: 限定站点资源分类 :param rule_groups: 指定过滤规则组，为空时使用默认搜索过滤规则 """ + search_params, cache_params = _build_title_search_params( + title=title, + page=page, + sites=sites, + mtype=mtype, + ) if cache_local: self.cancel_ai_recommend() - self.save_last_search_params( - keyword=title, - mtype=mtype, - area="title", - sites=sites, - ) + self.save_last_search_params(**cache_params) if title: logger.info(f"开始搜索资源，关键词：{title} ...") else: logger.info(f"开始浏览资源，站点：{sites} ...") # 搜索 - search_kwargs: Dict[str, Any] = { - "keyword": title, - "sites": sites, - "page": page, - } - if mtype is not None: - search_kwargs["mtype"] = mtype - torrents = self._SearchChain__search_all_sites(**search_kwargs) or [] + torrents = self._SearchChain__search_all_sites(**search_params) or [] if not torrents: logger.warning(f"{title} 未搜索到资源") return [] @@ -70,14 +104,9 @@ class SearchTitleOwner(_SearchOwnerBase): logger.warning(f"{title} 没有符合过滤规则的资源") return [] # 组装上下文 - contexts = [ - Context( - meta_info=self._build_title_search_meta(torrent, mtype), - torrent_info=torrent, - resource_source="search", - ) - for torrent in torrents - ] + contexts = _build_title_contexts( + torrents, mtype, self._build_title_search_meta + ) # 保存到本地文件 if cache_local: self._save_results(contexts) @@ -101,27 +130,21 @@ class SearchTitleOwner(_SearchOwnerBase): :param mtype: 限定站点资源分类 :param rule_groups: 指定过滤规则组，为空时使用默认搜索过滤规则 """ + search_params, cache_params = _build_title_search_params( + title=title, + page=page, + sites=sites, + mtype=mtype, + ) if cache_local: self.cancel_ai_recommend() - await self.async_save_last_search_params( - keyword=title, - mtype=mtype, - area="title", - sites=sites, - ) + await self.async_save_last_search_params(**cache_params) if title: logger.info(f"开始搜索资源，关键词：{title} ...") else: logger.info(f"开始浏览资源，站点：{sites} ...") # 搜索 - search_kwargs: Dict[str, Any] = { - "keyword": title, - "sites": sites, - "page": page, - } - if mtype is not None: - search_kwargs["mtype"] = mtype - torrents = await self._SearchChain__async_search_all_sites(**search_kwargs) or [] + torrents = await self._SearchChain__async_search_all_sites(**search_params) or [] if not torrents: logger.warning(f"{title} 未搜索到资源") return [] @@ -134,14 +157,9 @@ class SearchTitleOwner(_SearchOwnerBase): logger.warning(f"{title} 没有符合过滤规则的资源") return [] # 组装上下文 - contexts = [ - Context( - meta_info=self._build_title_search_meta(torrent, mtype), - torrent_info=torrent, - resource_source="search", - ) - for torrent in torrents - ] + contexts = _build_title_contexts( + torrents, mtype, self._build_title_search_meta + ) # 保存到本地文件 if cache_local: await self._async_save_results(contexts) @@ -159,14 +177,16 @@ class SearchTitleOwner(_SearchOwnerBase): """ 根据标题渐进式搜索资源，不识别媒体信息，按默认搜索过滤规则返回结果 """ + search_params, cache_params = _build_title_search_params( + title=title, + page=page, + sites=sites, + mtype=mtype, + include_empty_type=True, + ) if cache_local: self.cancel_ai_recommend() - await self.async_save_last_search_params( - keyword=title, - mtype=mtype, - area="title", - sites=sites, - ) + await self.async_save_last_search_params(**cache_params) if title: logger.info(f"开始渐进式搜索资源，关键词：{title} ...") else: @@ -178,7 +198,7 @@ class SearchTitleOwner(_SearchOwnerBase): if rule_groups is None: rule_groups = get_configured_system_config().get(SystemConfigKey.SearchFilterRuleGroups) or [] async for event in self._SearchChain__async_search_all_sites_stream( - keyword=title, sites=sites, page=page, mtype=mtype + **search_params ): result = event.pop("items", []) or [] candidate_count += len(result) @@ -187,14 +207,9 @@ class SearchTitleOwner(_SearchOwnerBase): torrents=result, rule_groups=rule_groups, ) - batch_contexts = [ - Context( - meta_info=self._build_title_search_meta(torrent, mtype), - torrent_info=torrent, - resource_source="search", - ) - for torrent in result - ] + batch_contexts = _build_title_contexts( + result, mtype, self._build_title_search_meta + ) if batch_contexts: contexts.extend(batch_contexts) yield { diff --git a/tests/test_search_sync_async_parity.py b/tests/test_search_sync_async_parity.py new file mode 100644 index 000000000..4234e3cbe --- /dev/null +++ b/tests/test_search_sync_async_parity.py @@ -0,0 +1,324 @@ +"""搜索链同步、异步与流式入口的共享业务决策回归测试。""" + +import asyncio +from copy import deepcopy +from types import SimpleNamespace +from typing import Any + +from app.chain.search import media as media_module +from app.chain.search.facade import SearchChain +from app.domain.context import MediaInfo, TorrentInfo +from app.domain.metainfo import MetaInfo +from app.schemas.types import MediaSource, MediaType + + +def _media() -> MediaInfo: + """构造具备稳定来源身份和别名的测试媒体。""" + return MediaInfo( + media_source=MediaSource.TMDB, + media_id="100", + tmdb_id=100, + title="测试电影", + names=["Test Movie"], + type=MediaType.MOVIE, + year="2026", + ) + + +def _torrent(title: str) -> TorrentInfo: + """构造一条无需外部 provider 的测试资源。""" + return TorrentInfo(title=title, description="description") + + +def _chain() -> SearchChain: + """绕过运行时组合根构造可局部注入的搜索链。""" + return object.__new__(SearchChain) + + +def test_id_search_sync_async_share_request_and_missing_plan(monkeypatch): + """ID 搜索双入口应产生完全相同的识别、缓存和缺集参数。""" + chain = _chain() + target = _media() + sync_calls: list[dict[str, Any]] = [] + async_calls: list[dict[str, Any]] = [] + saved_sync: list[dict[str, Any]] = [] + saved_async: list[dict[str, Any]] = [] + result = [SimpleNamespace(name="context")] + + class FakeMediaChain: + """记录同步与异步识别参数并返回同一媒体快照。""" + + def recognize_media(self, **kwargs): + """返回同步识别结果。""" + sync_calls.append(kwargs) + return target + + async def async_recognize_media(self, **kwargs): + """返回异步识别结果。""" + async_calls.append(kwargs) + return target + + def process(**kwargs): + """记录同步处理计划。""" + sync_calls.append(kwargs) + return result + + async def async_process(**kwargs): + """记录异步处理计划。""" + async_calls.append(kwargs) + return result + + async def async_save_params(**kwargs): + """记录异步缓存参数。""" + saved_async.append(kwargs) + + async def async_save_results(_contexts): + """模拟异步结果缓存。""" + + monkeypatch.setattr(media_module, "MediaChain", FakeMediaChain) + chain.cancel_ai_recommend = lambda: None + chain.save_last_search_params = lambda **kwargs: saved_sync.append(kwargs) + chain.async_save_last_search_params = async_save_params + chain.process = process + chain.async_process = async_process + chain._save_results = lambda _contexts: None + chain._async_save_results = async_save_results + + kwargs = { + "media_source": MediaSource.TMDB, + "media_id": "100", + "mtype": MediaType.TV, + "area": "imdbid", + "season": 2, + "sites": [1, 3], + "cache_local": True, + "music_type": "recording", + } + sync_result = chain.search_by_id(**kwargs) + async_result = asyncio.run(chain.async_search_by_id(**kwargs)) + + assert sync_result is result + assert async_result is result + assert sync_calls[0] == async_calls[0] == { + "media_source": MediaSource.TMDB, + "media_id": "100", + "mtype": MediaType.TV, + "music_type": "recording", + } + assert sync_calls[1] == async_calls[1] + assert sync_calls[1]["sites"] == [1, 3] + assert sync_calls[1]["area"] == "imdbid" + missing = sync_calls[1]["no_exists"] + assert list(missing) == ["tmdb:100"] + assert list(missing["tmdb:100"]) == [2] + assert saved_sync == saved_async + + +def test_media_process_sync_async_share_keyword_stop_decision(monkeypatch): + """双入口应按相同关键字顺序搜索，并在首个有效结果后同时停止。""" + chain = _chain() + target = _media() + found = _torrent("测试电影 2026 1080p") + sync_keywords: list[str] = [] + async_keywords: list[str] = [] + sync_sleeps: list[int] = [] + async_sleeps: list[int] = [] + parsed: list[list[TorrentInfo]] = [] + + class FakeMediaChain: + """为同步与异步处理提供等价的附加信息结果。""" + + def supplement_media_info(self, mediainfo): + """返回同步附加信息。""" + return mediainfo + + async def async_supplement_media_info(self, mediainfo): + """返回异步附加信息。""" + return mediainfo + + def search_all_sites(**kwargs): + """第二个关键字开始返回资源。""" + keyword = kwargs["keyword"] + sync_keywords.append(keyword) + return [found] if keyword == "second" else [] + + async def async_search_all_sites(**kwargs): + """第二个关键字开始异步返回资源。""" + keyword = kwargs["keyword"] + async_keywords.append(keyword) + return [found] if keyword == "second" else [] + + def parse_result(**kwargs): + """记录进入统一结果处理边界的资源顺序。""" + parsed.append(list(kwargs["torrents"])) + return kwargs["torrents"] + + async def fake_async_sleep(delay): + """记录异步关键字退避但不实际等待。""" + async_sleeps.append(delay) + + monkeypatch.setattr(media_module, "MediaChain", FakeMediaChain) + monkeypatch.setattr(media_module.random, "randint", lambda _start, _end: 1) + monkeypatch.setattr(media_module.time, "sleep", sync_sleeps.append) + monkeypatch.setattr(media_module.asyncio, "sleep", fake_async_sleep) + chain.runtime_config = SimpleNamespace(search_multiple_name=False) + chain._copy_media_input = deepcopy + chain._prepare_params = lambda **_kwargs: (None, ["first", "second", "third"]) + chain._SearchChain__search_all_sites = search_all_sites + chain._SearchChain__async_search_all_sites = async_search_all_sites + chain._parse_result = parse_result + + sync_result = chain.process(target) + async_result = asyncio.run(chain.async_process(target)) + + assert sync_keywords == async_keywords == ["first", "second"] + assert sync_sleeps == async_sleeps == [1] + assert parsed == [[found], [found]] + assert sync_result == async_result == [found] + + +def test_id_search_stream_failure_preserves_order_and_skips_result_cache( + monkeypatch, +): + """识别失败时流入口应先保存请求状态，只输出错误且不保存空结果。""" + chain = _chain() + order: list[str] = [] + + class FakeMediaChain: + """模拟异步识别失败。""" + + async def async_recognize_media(self, **_kwargs): + """记录识别顺序并返回失败。""" + order.append("recognize") + return None + + async def save_params(**_kwargs): + """记录请求状态保存顺序。""" + order.append("params") + + async def save_results(_contexts): + """若失败路径错误保存结果则留下可断言记录。""" + order.append("results") + + async def collect_events(): + """完整消费识别失败的搜索流。""" + return [ + event + async for event in chain.async_search_by_id_stream( + media_source=MediaSource.TMDB, + media_id="404", + cache_local=True, + ) + ] + + monkeypatch.setattr(media_module, "MediaChain", FakeMediaChain) + chain.cancel_ai_recommend = lambda: order.append("cancel") + chain.async_save_last_search_params = save_params + chain._async_save_results = save_results + + events = asyncio.run(collect_events()) + + assert order == ["cancel", "params", "recognize"] + assert events == [ + {"type": "error", "success": False, "message": "媒体信息识别失败"} + ] + + +def test_title_search_sync_async_share_plan_filter_and_projection(): + """标题搜索双入口应共享 provider 参数、过滤决策和可替换元数据投影。""" + chain = _chain() + keep = _torrent("Movie 2026 1080p") + drop = _torrent("Movie 2026 2160p") + provider_calls: list[dict[str, Any]] = [] + filter_calls: list[list[TorrentInfo]] = [] + meta_calls: list[str] = [] + + def search_all_sites(**kwargs): + """记录同步 provider 参数。""" + provider_calls.append(kwargs) + return [keep, drop] + + async def async_search_all_sites(**kwargs): + """记录异步 provider 参数。""" + provider_calls.append(kwargs) + return [keep, drop] + + def filter_torrents(torrents, rule_groups=None): + """记录共享过滤输入并保留首条资源。""" + assert rule_groups == ["quality"] + filter_calls.append(list(torrents)) + return [keep] + + def build_meta(torrent, _mtype): + """模拟插件或测试替换的稳定元数据构造点。""" + meta_calls.append(torrent.title) + return MetaInfo(title=f"mapped:{torrent.title}") + + chain._SearchChain__search_all_sites = search_all_sites + chain._SearchChain__async_search_all_sites = async_search_all_sites + chain._filter_title_search_torrents = filter_torrents + chain._build_title_search_meta = build_meta + + sync_result = chain.search_by_title( + title="Movie", page=2, sites=[1], mtype=MediaType.MOVIE, rule_groups=["quality"] + ) + async_result = asyncio.run( + chain.async_search_by_title( + title="Movie", + page=2, + sites=[1], + mtype=MediaType.MOVIE, + rule_groups=["quality"], + ) + ) + + assert provider_calls == [ + {"keyword": "Movie", "sites": [1], "page": 2, "mtype": MediaType.MOVIE}, + {"keyword": "Movie", "sites": [1], "page": 2, "mtype": MediaType.MOVIE}, + ] + assert filter_calls == [[keep, drop], [keep, drop]] + assert meta_calls == [keep.title, keep.title] + assert [context.meta_info.title for context in sync_result] == [ + context.meta_info.title for context in async_result + ] == [f"mapped:{keep.title}"] + + +def test_title_stream_preserves_provider_and_result_order(): + """标题流应按 provider 批次顺序追加，并以相同顺序发布累计完成结果。""" + chain = _chain() + first = _torrent("Movie 2026 1080p") + second = _torrent("Movie 2026 2160p") + provider_params: list[dict[str, Any]] = [] + + async def search_stream(**kwargs): + """按固定顺序输出两个 provider 批次。""" + provider_params.append(kwargs) + yield {"type": "append", "site_id": 1, "items": [first]} + yield {"type": "append", "site_id": 2, "items": [second]} + + async def collect_events(): + """完整消费标题搜索流。""" + return [ + event + async for event in chain.async_search_by_title_stream( + title="Movie", sites=[1, 2], rule_groups=[] + ) + ] + + chain._SearchChain__async_search_all_sites_stream = search_stream + chain._filter_title_search_torrents = lambda torrents, rule_groups=None: torrents + chain._build_title_search_meta = ( + lambda torrent, _mtype: MetaInfo(title=torrent.title) + ) + + events = asyncio.run(collect_events()) + + assert provider_params == [ + {"keyword": "Movie", "sites": [1, 2], "page": 0, "mtype": None} + ] + assert [event["type"] for event in events] == ["append", "append", "done"] + assert [event["total_items"] for event in events] == [1, 2, 2] + assert [ + item["torrent_info"]["title"] for item in events[-1]["items"] + ] == [first.title, second.title] + assert events[-1]["candidate_items"] == 2