diff --git a/app/chain/search/media.py b/app/chain/search/media.py index 6cbff0ddd..634d81102 100644 --- a/app/chain/search/media.py +++ b/app/chain/search/media.py @@ -3,7 +3,19 @@ import asyncio import random import time -from typing import Any, AsyncIterator, Dict, List, Optional, Tuple, cast +from dataclasses import dataclass +from typing import ( + Any, + AsyncIterator, + Awaitable, + Callable, + Dict, + Generator, + List, + Optional, + Tuple, + cast, +) from app.chain.media import MediaChain from app.chain.search.contract import _SearchOwnerBase as _SearchOwnerBase @@ -76,6 +88,98 @@ def _should_stop_keyword_search( return not search_multiple_name and bool(torrents) +@dataclass(frozen=True, slots=True) +class _KeywordSearchRequest: + """描述共享关键字状态机交给真实 I/O 边界的一次请求。""" + + keyword: str + search_count: int + + +@dataclass(frozen=True, slots=True) +class _KeywordSearchResult: + """冻结关键字状态机聚合结果及其提前停止决策。""" + + torrents: List[TorrentInfo] + stopped_early: bool + + +_KeywordSearchResolution = Generator[ + _KeywordSearchRequest, List[TorrentInfo], _KeywordSearchResult +] + + +def _keyword_search_resolution( + keywords: List[str], + search_multiple_name: bool, +) -> _KeywordSearchResolution: + """统一推进关键字顺序、结果聚合和首个有效结果短路。""" + torrents: List[TorrentInfo] = [] + for search_count, keyword in enumerate(keywords): + torrents.extend( + (yield _KeywordSearchRequest( + keyword=keyword, + search_count=search_count, + )) + ) + if _should_stop_keyword_search(search_multiple_name, torrents): + return _KeywordSearchResult(torrents=torrents, stopped_early=True) + return _KeywordSearchResult(torrents=torrents, stopped_early=False) + + +def _run_keyword_search_sync( + resolution: _KeywordSearchResolution, + execute: Callable[[_KeywordSearchRequest], List[TorrentInfo]], +) -> _KeywordSearchResult: + """通过同步 provider 驱动共享关键字状态机。""" + try: + request = next(resolution) + except StopIteration as outcome: + return cast(_KeywordSearchResult, outcome.value) + while True: + try: + request = resolution.send(execute(request)) + except StopIteration as outcome: + return cast(_KeywordSearchResult, outcome.value) + + +async def _run_keyword_search_async( + resolution: _KeywordSearchResolution, + execute: Callable[[_KeywordSearchRequest], Awaitable[List[TorrentInfo]]], +) -> _KeywordSearchResult: + """通过异步 provider 驱动共享关键字状态机。""" + try: + request = next(resolution) + except StopIteration as outcome: + return cast(_KeywordSearchResult, outcome.value) + while True: + try: + request = resolution.send(await execute(request)) + except StopIteration as outcome: + return cast(_KeywordSearchResult, outcome.value) + + +def _build_result_params( + torrents: List[TorrentInfo], + mediainfo: MediaInfo, + keyword: Optional[str], + rule_groups: Optional[List[str]], + season_episodes: Optional[Dict[int, List[int]]], + custom_words: Optional[List[str]], + filter_params: Optional[Dict[str, str]], +) -> Dict[str, Any]: + """构造同步、异步和流式结果解析共享的参数快照。""" + return { + "torrents": torrents, + "mediainfo": mediainfo, + "keyword": keyword, + "rule_groups": rule_groups, + "season_episodes": season_episodes, + "custom_words": custom_words, + "filter_params": filter_params, + } + + def _build_candidate_contexts( mediainfo: MediaInfo, torrents: List[TorrentInfo], @@ -291,46 +395,45 @@ class SearchMediaOwner(_SearchOwnerBase): # 准备搜索参数 season_episodes, keywords = self._prepare_params(mediainfo=mediainfo, keyword=keyword, no_exists=no_exists) - # 站点搜索结果 - torrents: List[TorrentInfo] = [] - # 站点搜索次数 - search_count = 0 - - # 多关键字执行搜索 - for search_word in keywords: - # 强制休眠 1-10 秒 - if search_count > 0: - logger.info(f"已搜索 {search_count} 次，强制休眠 1-10 秒 ...") + def execute_search(request: _KeywordSearchRequest) -> List[TorrentInfo]: + """执行共享状态机请求的同步站点搜索。""" + if request.search_count > 0: + logger.info( + f"已搜索 {request.search_count} 次，强制休眠 1-10 秒 ..." + ) time.sleep(random.randint(1, 10)) - - # 搜索站点 - results = ( - self._SearchChain__search_all_sites(mediainfo=mediainfo, keyword=search_word, sites=sites, area=area) + return ( + self._SearchChain__search_all_sites( + mediainfo=mediainfo, + keyword=request.keyword, + sites=sites, + area=area, + ) or [] ) - # 合并结果 - search_count += 1 - torrents.extend(results) - - # 有结果则停止 - if _should_stop_keyword_search( - self.runtime_config.search_multiple_name, torrents - ): - logger.info(f"共搜索到 {len(torrents)} 个资源，停止搜索") - break + outcome = _run_keyword_search_sync( + _keyword_search_resolution( + keywords, self.runtime_config.search_multiple_name + ), + execute_search, + ) + if outcome.stopped_early: + logger.info(f"共搜索到 {len(outcome.torrents)} 个资源，停止搜索") # 处理结果 return cast( List[Context], self._parse_result( - torrents=torrents, - mediainfo=mediainfo, - keyword=keyword, - rule_groups=rule_groups, - season_episodes=season_episodes, - custom_words=custom_words, - filter_params=filter_params, + **_build_result_params( + torrents=outcome.torrents, + mediainfo=mediainfo, + keyword=keyword, + rule_groups=rule_groups, + season_episodes=season_episodes, + custom_words=custom_words, + filter_params=filter_params, + ) ), ) @@ -392,44 +495,48 @@ class SearchMediaOwner(_SearchOwnerBase): # 准备搜索参数 season_episodes, keywords = self._prepare_params(mediainfo=mediainfo, keyword=keyword, no_exists=no_exists) - # 站点搜索结果 - torrents: List[TorrentInfo] = [] - # 站点搜索次数 - search_count = 0 - - # 多关键字执行搜索 - for search_word in keywords: - # 强制休眠 1-10 秒 - if search_count > 0: - logger.info(f"已搜索 {search_count} 次，强制休眠 1-10 秒 ...") + async def execute_search( + request: _KeywordSearchRequest, + ) -> List[TorrentInfo]: + """执行共享状态机请求的异步站点搜索。""" + if request.search_count > 0: + logger.info( + f"已搜索 {request.search_count} 次，强制休眠 1-10 秒 ..." + ) await asyncio.sleep(random.randint(1, 10)) - # 搜索站点 - torrents.extend( + return ( await self._SearchChain__async_search_all_sites( - mediainfo=mediainfo, keyword=search_word, sites=sites, area=area + mediainfo=mediainfo, + keyword=request.keyword, + sites=sites, + area=area, ) or [] ) - search_count += 1 - # 未开启多名称搜索时，有结果则停止 - if _should_stop_keyword_search( - self.runtime_config.search_multiple_name, torrents - ): - logger.info(f"共搜索到 {len(torrents)} 个资源，停止搜索") - break + + outcome = await _run_keyword_search_async( + _keyword_search_resolution( + keywords, self.runtime_config.search_multiple_name + ), + execute_search, + ) + if outcome.stopped_early: + logger.info(f"共搜索到 {len(outcome.torrents)} 个资源，停止搜索") # 处理结果 return cast( List[Context], await run_in_threadpool( self._parse_result, - torrents=torrents, - mediainfo=mediainfo, - keyword=keyword, - rule_groups=rule_groups, - season_episodes=season_episodes, - custom_words=custom_words, - filter_params=filter_params, + **_build_result_params( + torrents=outcome.torrents, + mediainfo=mediainfo, + keyword=keyword, + rule_groups=rule_groups, + season_episodes=season_episodes, + custom_words=custom_words, + filter_params=filter_params, + ), ), ) @@ -482,53 +589,69 @@ class SearchMediaOwner(_SearchOwnerBase): # 准备搜索参数 season_episodes, keywords = self._prepare_params(mediainfo=mediainfo, keyword=keyword, no_exists=no_exists) - torrents: List[TorrentInfo] = [] candidate_contexts: List[Context] = [] - search_count = 0 - - for search_word in keywords: - if search_count > 0: - logger.info(f"已搜索 {search_count} 次，强制休眠 1-10 秒 ...") - await asyncio.sleep(random.randint(1, 10)) - - async for event in self._SearchChain__async_search_all_sites_stream( - mediainfo=mediainfo, keyword=search_word, sites=sites, area=area - ): - result = event.pop("items", []) or [] - torrents.extend(result) - batch_contexts = _build_candidate_contexts(mediainfo, result) - candidate_contexts.extend(batch_contexts) - yield { - **event, - "type": "append", - "stage": "searching", - "items": [cast(Any, context).to_dict() for context in batch_contexts], - "total_items": len(candidate_contexts), - } - - search_count += 1 - if _should_stop_keyword_search( - self.runtime_config.search_multiple_name, torrents - ): - logger.info(f"共搜索到 {len(torrents)} 个资源，停止搜索") - break + resolution = _keyword_search_resolution( + keywords, self.runtime_config.search_multiple_name + ) + outcome = _KeywordSearchResult(torrents=[], stopped_early=False) + try: + request = next(resolution) + except StopIteration as completed: + outcome = cast(_KeywordSearchResult, completed.value) + else: + while True: + if request.search_count > 0: + logger.info( + f"已搜索 {request.search_count} 次，强制休眠 1-10 秒 ..." + ) + await asyncio.sleep(random.randint(1, 10)) + search_results: List[TorrentInfo] = [] + async for event in self._SearchChain__async_search_all_sites_stream( + mediainfo=mediainfo, + keyword=request.keyword, + sites=sites, + area=area, + ): + result = event.pop("items", []) or [] + search_results.extend(result) + batch_contexts = _build_candidate_contexts(mediainfo, result) + candidate_contexts.extend(batch_contexts) + yield { + **event, + "type": "append", + "stage": "searching", + "items": [ + cast(Any, context).to_dict() + for context in batch_contexts + ], + "total_items": len(candidate_contexts), + } + try: + request = resolution.send(search_results) + except StopIteration as completed: + outcome = cast(_KeywordSearchResult, completed.value) + break + if outcome.stopped_early: + logger.info(f"共搜索到 {len(outcome.torrents)} 个资源，停止搜索") yield { "type": "progress", "stage": "filtering", "value": 98, - "text": f"正在过滤匹配 {len(torrents)} 个候选资源 ...", + "text": f"正在过滤匹配 {len(outcome.torrents)} 个候选资源 ...", } contexts = await run_in_threadpool( self._parse_result, - torrents=torrents, - mediainfo=mediainfo, - keyword=keyword, - rule_groups=rule_groups, - season_episodes=season_episodes, - custom_words=custom_words, - filter_params=filter_params, + **_build_result_params( + torrents=outcome.torrents, + mediainfo=mediainfo, + keyword=keyword, + rule_groups=rule_groups, + season_episodes=season_episodes, + custom_words=custom_words, + filter_params=filter_params, + ), ) final_items = [context.to_dict() for context in contexts] yield { diff --git a/app/chain/search/title.py b/app/chain/search/title.py index 76a54a177..2625ae26c 100644 --- a/app/chain/search/title.py +++ b/app/chain/search/title.py @@ -1,5 +1,6 @@ """标题搜索入口与标题候选过滤 owner。""" +from dataclasses import dataclass from typing import Any, AsyncIterator, Callable, Dict, List, Optional, Tuple, cast from app.application.configuration import ( @@ -57,6 +58,46 @@ def _build_title_contexts( ] +@dataclass(frozen=True, slots=True) +class _TitleSearchResult: + """封装标题搜索候选解析结果及需要记录的失败原因。""" + + contexts: List[Context] + warning: Optional[str] = None + + +def _resolve_title_search_result( + title: str, + torrents: List[TorrentInfo], + rule_groups: Optional[List[str]], + mtype: Optional[MediaType], + filter_torrents: Callable[..., List[TorrentInfo]], + build_meta: Callable[[TorrentInfo, Optional[MediaType]], Any], +) -> _TitleSearchResult: + """统一标题候选为空、过滤和上下文投影的结果决策。""" + if not torrents: + return _TitleSearchResult( + contexts=[], + warning=f"{title} 未搜索到资源", + ) + filtered_torrents = filter_torrents( + torrents=torrents, + rule_groups=rule_groups, + ) + if not filtered_torrents: + return _TitleSearchResult( + contexts=[], + warning=f"{title} 没有符合过滤规则的资源", + ) + return _TitleSearchResult( + contexts=_build_title_contexts( + filtered_torrents, + mtype, + build_meta, + ) + ) + + class SearchTitleOwner(_SearchOwnerBase): """标题搜索入口与标题候选过滤 owner。""" @@ -92,25 +133,21 @@ class SearchTitleOwner(_SearchOwnerBase): else: logger.info(f"开始浏览资源，站点：{sites} ...") # 搜索 - torrents = self._SearchChain__search_all_sites(**search_params) or [] - if not torrents: - logger.warning(f"{title} 未搜索到资源") - return [] - torrents = self._filter_title_search_torrents( - torrents=torrents, + result = _resolve_title_search_result( + title=title, + torrents=self._SearchChain__search_all_sites(**search_params) or [], rule_groups=rule_groups, + mtype=mtype, + filter_torrents=self._filter_title_search_torrents, + build_meta=self._build_title_search_meta, ) - if not torrents: - logger.warning(f"{title} 没有符合过滤规则的资源") + if result.warning: + logger.warning(result.warning) return [] - # 组装上下文 - contexts = _build_title_contexts( - torrents, mtype, self._build_title_search_meta - ) # 保存到本地文件 if cache_local: - self._save_results(contexts) - return contexts + self._save_results(result.contexts) + return result.contexts async def async_search_by_title( self, @@ -144,26 +181,38 @@ class SearchTitleOwner(_SearchOwnerBase): else: logger.info(f"开始浏览资源，站点：{sites} ...") # 搜索 - torrents = await self._SearchChain__async_search_all_sites(**search_params) or [] - if not torrents: - logger.warning(f"{title} 未搜索到资源") - return [] - torrents = await run_in_threadpool( - self._filter_title_search_torrents, - torrents=torrents, - rule_groups=rule_groups, + torrents = ( + await self._SearchChain__async_search_all_sites(**search_params) or [] ) if not torrents: - logger.warning(f"{title} 没有符合过滤规则的资源") + result = _resolve_title_search_result( + title=title, + torrents=torrents, + rule_groups=rule_groups, + mtype=mtype, + filter_torrents=self._filter_title_search_torrents, + build_meta=self._build_title_search_meta, + ) + else: + result = cast( + _TitleSearchResult, + await run_in_threadpool( + _resolve_title_search_result, + title=title, + torrents=torrents, + rule_groups=rule_groups, + mtype=mtype, + filter_torrents=self._filter_title_search_torrents, + build_meta=self._build_title_search_meta, + ), + ) + if result.warning: + logger.warning(result.warning) return [] - # 组装上下文 - contexts = _build_title_contexts( - torrents, mtype, self._build_title_search_meta - ) # 保存到本地文件 if cache_local: - await self._async_save_results(contexts) - return contexts + await self._async_save_results(result.contexts) + return result.contexts async def async_search_by_title_stream( self, diff --git a/tests/test_search_sync_async_parity.py b/tests/test_search_sync_async_parity.py index 4234e3cbe..0fdb872bf 100644 --- a/tests/test_search_sync_async_parity.py +++ b/tests/test_search_sync_async_parity.py @@ -6,6 +6,7 @@ from types import SimpleNamespace from typing import Any from app.chain.search import media as media_module +from app.chain.search import title as title_module from app.chain.search.facade import SearchChain from app.domain.context import MediaInfo, TorrentInfo from app.domain.metainfo import MetaInfo @@ -177,6 +178,177 @@ def test_media_process_sync_async_share_keyword_stop_decision(monkeypatch): assert sync_result == async_result == [found] +def test_keyword_resolution_searches_all_names_when_enabled(): + """开启多名称搜索时共享状态机应完整执行并稳定聚合各关键字结果。""" + first = _torrent("First Result") + third = _torrent("Third Result") + expected = { + "first": [first], + "second": [], + "third": [third], + } + sync_keywords: list[str] = [] + async_keywords: list[str] = [] + + def execute_sync(request): + """记录同步驱动顺序并返回关键字结果。""" + sync_keywords.append(request.keyword) + return expected[request.keyword] + + async def execute_async(request): + """记录异步驱动顺序并返回关键字结果。""" + async_keywords.append(request.keyword) + return expected[request.keyword] + + keywords = ["first", "second", "third"] + sync_result = media_module._run_keyword_search_sync( + media_module._keyword_search_resolution(keywords, search_multiple_name=True), + execute_sync, + ) + async_result = asyncio.run( + media_module._run_keyword_search_async( + media_module._keyword_search_resolution( + keywords, search_multiple_name=True + ), + execute_async, + ) + ) + + assert sync_keywords == async_keywords == keywords + assert sync_result == async_result + assert sync_result.torrents == [first, third] + assert sync_result.stopped_early is False + + +def test_media_process_stream_shares_keyword_order_and_stop_decision(monkeypatch): + """流入口应复用双入口的关键字顺序、退避次数和首结果短路决策。""" + chain = _chain() + target = _media() + found = _torrent("测试电影 2026 1080p") + stream_keywords: list[str] = [] + sleeps: list[int] = [] + parsed: list[list[TorrentInfo]] = [] + + class FakeMediaChain: + """为流处理提供无需外部服务的附加信息结果。""" + + async def async_supplement_media_info(self, mediainfo): + """返回异步附加信息。""" + return mediainfo + + async def search_stream(**kwargs): + """第二个关键字输出资源，供共享状态机执行短路。""" + keyword = kwargs["keyword"] + stream_keywords.append(keyword) + yield { + "site_id": len(stream_keywords), + "items": [found] if keyword == "second" else [], + } + + async def fake_async_sleep(delay): + """记录流关键字退避但不实际等待。""" + sleeps.append(delay) + + def parse_result(**kwargs): + """记录流最终过滤前的资源顺序并返回空上下文。""" + parsed.append(list(kwargs["torrents"])) + return [] + + async def collect_events(): + """完整消费媒体搜索流。""" + return [ + event + async for event in chain.async_process_stream(mediainfo=target) + ] + + monkeypatch.setattr(media_module, "MediaChain", FakeMediaChain) + monkeypatch.setattr(media_module.random, "randint", lambda _start, _end: 1) + 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__async_search_all_sites_stream = search_stream + chain._parse_result = parse_result + + events = asyncio.run(collect_events()) + + assert stream_keywords == ["first", "second"] + assert sleeps == [1] + assert parsed == [[found]] + assert [event["type"] for event in events] == [ + "append", + "append", + "progress", + "replace", + "done", + ] + assert events[-1]["candidate_items"] == 1 + + +def test_media_process_empty_keyword_plan_skips_all_provider_io(monkeypatch): + """空关键字计划应由共享状态机直接完成，且三入口均不触发 provider。""" + chain = _chain() + target = _media() + 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 unexpected_provider(**_kwargs): + """标记空计划错误触发了同步 provider。""" + raise AssertionError("空关键字计划不应触发同步 provider") + + async def unexpected_async_provider(**_kwargs): + """标记空计划错误触发了异步 provider。""" + raise AssertionError("空关键字计划不应触发异步 provider") + + async def unexpected_stream_provider(**_kwargs): + """标记空计划错误触发了流式 provider。""" + yield {"items": []} + raise AssertionError("空关键字计划不应触发流式 provider") + + def parse_result(**kwargs): + """记录三入口交付给结果解析边界的空资源。""" + parsed.append(list(kwargs["torrents"])) + return [] + + async def collect_events(): + """完整消费空关键字媒体搜索流。""" + return [ + event + async for event in chain.async_process_stream(mediainfo=target) + ] + + monkeypatch.setattr(media_module, "MediaChain", FakeMediaChain) + chain.runtime_config = SimpleNamespace(search_multiple_name=False) + chain._copy_media_input = deepcopy + chain._prepare_params = lambda **_kwargs: (None, []) + chain._SearchChain__search_all_sites = unexpected_provider + chain._SearchChain__async_search_all_sites = unexpected_async_provider + chain._SearchChain__async_search_all_sites_stream = unexpected_stream_provider + chain._parse_result = parse_result + + assert chain.process(target) == [] + assert asyncio.run(chain.async_process(target)) == [] + events = asyncio.run(collect_events()) + + assert parsed == [[], [], []] + assert [event["type"] for event in events] == [ + "progress", + "replace", + "done", + ] + assert events[-1]["candidate_items"] == 0 + + def test_id_search_stream_failure_preserves_order_and_skips_result_cache( monkeypatch, ): @@ -283,6 +455,73 @@ def test_title_search_sync_async_share_plan_filter_and_projection(): ] == [f"mapped:{keep.title}"] +def test_title_search_sync_async_share_empty_result_decisions(monkeypatch): + """标题双入口应共享无候选与过滤为空的警告，并跳过结果缓存。""" + chain = _chain() + candidate = _torrent("Movie 2026 1080p") + provider_result: list[TorrentInfo] = [] + warnings: list[str] = [] + filter_calls: list[list[TorrentInfo]] = [] + saved_results: list[list[Any]] = [] + + def search_all_sites(**_kwargs): + """返回当前场景的同步候选快照。""" + return list(provider_result) + + async def async_search_all_sites(**_kwargs): + """返回当前场景的异步候选快照。""" + return list(provider_result) + + def filter_torrents(torrents, rule_groups=None): + """记录过滤输入并模拟候选全部被规则排除。""" + assert rule_groups == ["quality"] + filter_calls.append(list(torrents)) + return [] + + async def save_params(**_kwargs): + """模拟异步请求参数缓存。""" + + async def save_results(contexts): + """记录不应发生的异步结果缓存。""" + saved_results.append(list(contexts)) + + monkeypatch.setattr(title_module.logger, "warning", warnings.append) + 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 = lambda torrent, _mtype: MetaInfo( + title=torrent.title + ) + chain.cancel_ai_recommend = lambda: None + chain.save_last_search_params = lambda **_kwargs: None + chain.async_save_last_search_params = save_params + chain._save_results = lambda contexts: saved_results.append(list(contexts)) + chain._async_save_results = save_results + + for torrents, warning in ( + ([], "Movie 未搜索到资源"), + ([candidate], "Movie 没有符合过滤规则的资源"), + ): + provider_result[:] = torrents + warning_count = len(warnings) + assert chain.search_by_title( + title="Movie", + cache_local=True, + rule_groups=["quality"], + ) == [] + assert asyncio.run( + chain.async_search_by_title( + title="Movie", + cache_local=True, + rule_groups=["quality"], + ) + ) == [] + assert warnings[warning_count:] == [warning, warning] + + assert filter_calls == [[candidate], [candidate]] + assert saved_results == [] + + def test_title_stream_preserves_provider_and_result_order(): """标题流应按 provider 批次顺序追加，并以相同顺序发布累计完成结果。""" chain = _chain()