diff --git a/app/chain/__init__.py b/app/chain/__init__.py index 8c24c405d..dbbb4ec86 100644 --- a/app/chain/__init__.py +++ b/app/chain/__init__.py @@ -613,6 +613,29 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin, "search_torrents", site=site, keyword=keyword, mtype=mtype, page=page ) + def search_plugin_torrents( + self, + keyword: str, + mtype: Optional[MediaType] = None, + page: Optional[int] = 0, + ) -> List[TorrentInfo]: + """仅搜索插件提供的资源源，避免依赖或重复绑定站点索引器。""" + return self._module_dispatcher.execute_plugin_modules( + "search_torrents", None, site={}, keyword=keyword, mtype=mtype, page=page + ) or [] + + def search_site_torrents( + self, + site: dict, + keyword: str, + mtype: Optional[MediaType] = None, + page: Optional[int] = 0, + ) -> List[TorrentInfo]: + """仅搜索指定站点索引器；插件资源源由搜索链统一调用一次。""" + return self._module_dispatcher.execute_system_modules( + "search_torrents", None, site=site, keyword=keyword, mtype=mtype, page=page + ) or [] + def search_subtitles( self, site: dict, @@ -649,6 +672,31 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin, "async_search_torrents", site=site, keyword=keyword, mtype=mtype, page=page ) + async def async_search_plugin_torrents( + self, + keyword: str, + mtype: Optional[MediaType] = None, + page: Optional[int] = 0, + ) -> List[TorrentInfo]: + """异步搜索插件提供的资源源。""" + return await self._module_dispatcher.async_execute_plugin_modules( + "async_search_torrents", None, + site={}, keyword=keyword, mtype=mtype, page=page + ) or [] + + async def async_search_site_torrents( + self, + site: dict, + keyword: str, + mtype: Optional[MediaType] = None, + page: Optional[int] = 0, + ) -> List[TorrentInfo]: + """异步搜索指定站点索引器。""" + return await self._module_dispatcher.async_execute_system_modules( + "async_search_torrents", None, + site=site, keyword=keyword, mtype=mtype, page=page + ) or [] + async def async_search_subtitles( self, site: dict, diff --git a/app/chain/search.py b/app/chain/search.py index f985df5ff..192789d87 100644 --- a/app/chain/search.py +++ b/app/chain/search.py @@ -2321,9 +2321,15 @@ class SearchChain(ChainBase): # 检查站点索引开关 if not sites or indexer.get("id") in sites: indexer_sites.append(indexer) + + plugin_results = self.search_plugin_torrents( + keyword=keyword, + mtype=mediainfo.type if mediainfo else mtype, + page=page, + ) if not indexer_sites: - logger.warn('未开启任何有效站点，无法搜索资源') - return [] + logger.info(f'未开启有效站点，插件资源源返回 {len(plugin_results)} 条资源') + return plugin_results # 开始进度 progress = ProgressHelper(ProgressKey.Search) @@ -2339,7 +2345,7 @@ class SearchChain(ChainBase): progress.update(value=0, text=f"开始搜索，共 {len(indexer_sites)} 个站点，{len(search_pages)} 页 ...") # 结果集 - results = [] + results = list(plugin_results) # 同一站点按页顺序抓取，避免空页后仍继续请求该站点的后续页。 max_workers = min( len(indexer_sites), @@ -2356,13 +2362,13 @@ class SearchChain(ChainBase): search_keyword = mediainfo.imdb_id if area == "imdbid" and mediainfo else keyword if area == "imdbid": # 搜索IMDBID - task = executor.submit(self.search_torrents, site=site, + task = executor.submit(self.search_site_torrents, site=site, keyword=search_keyword, mtype=mediainfo.type if mediainfo else mtype, page=search_page) else: # 搜索标题 - task = executor.submit(self.search_torrents, site=site, + task = executor.submit(self.search_site_torrents, site=site, keyword=search_keyword, mtype=mediainfo.type if mediainfo else mtype, page=search_page) @@ -2438,9 +2444,15 @@ class SearchChain(ChainBase): # 检查站点索引开关 if not sites or indexer.get("id") in sites: indexer_sites.append(indexer) + + plugin_results = await self.async_search_plugin_torrents( + keyword=keyword, + mtype=mediainfo.type if mediainfo else mtype, + page=page, + ) if not indexer_sites: - logger.warn('未开启任何有效站点，无法搜索资源') - return [] + logger.info(f'未开启有效站点，插件资源源返回 {len(plugin_results)} 条资源') + return plugin_results # 开始进度（异步后端，避免同步 Redis 在事件循环上阻塞） progress = AsyncProgressHelper(ProgressKey.Search) @@ -2456,7 +2468,7 @@ class SearchChain(ChainBase): await progress.update(value=0, text=f"开始搜索，共 {len(indexer_sites)} 个站点，{len(search_pages)} 页 ...") # 结果集 - results = [] + results = list(plugin_results) semaphore = asyncio.Semaphore( self.runtime_config.search_threadpool_size or total_num ) @@ -2468,12 +2480,12 @@ class SearchChain(ChainBase): async with semaphore: if area == "imdbid": # 搜索IMDBID - return await self.async_search_torrents(site=site, + return await self.async_search_site_torrents(site=site, keyword=mediainfo.imdb_id if mediainfo else None, mtype=mediainfo.type if mediainfo else mtype, page=search_page) # 搜索标题 - return await self.async_search_torrents(site=site, + return await self.async_search_site_torrents(site=site, keyword=keyword, mtype=mediainfo.type if mediainfo else mtype, page=search_page) @@ -2562,16 +2574,37 @@ class SearchChain(ChainBase): for indexer in await SitesHelper().async_get_indexers(): if not sites or indexer.get("id") in sites: indexer_sites.append(indexer) + + plugin_results = await self.async_search_plugin_torrents( + keyword=keyword, + mtype=mediainfo.type if mediainfo else mtype, + page=page, + ) + if plugin_results: + yield { + "type": "append", + "stage": "searching", + "value": 100 if not indexer_sites else 0, + "text": f"插件资源源返回 {len(plugin_results)} 条资源", + "items": plugin_results, + "site": "插件资源源", + "site_id": None, + "page": page, + "finished": 0, + "total": len(indexer_sites), + "total_items": len(plugin_results), + } if not indexer_sites: - logger.warn('未开启任何有效站点，无法搜索资源') + logger.info(f'未开启有效站点，插件资源源返回 {len(plugin_results)} 条资源') yield { "type": "done", "stage": "searching", "value": 100, - "text": "未开启任何有效站点，无法搜索资源", + "text": f"搜索完成，共 {len(plugin_results)} 条资源", "items": [], "finished": 0, - "total": 0 + "total": 0, + "total_items": len(plugin_results), } return @@ -2604,12 +2637,12 @@ class SearchChain(ChainBase): """ async with semaphore: if area == "imdbid": - site_result = await self.async_search_torrents(site=site, + site_result = await self.async_search_site_torrents(site=site, keyword=mediainfo.imdb_id if mediainfo else None, mtype=mediainfo.type if mediainfo else mtype, page=search_page) else: - site_result = await self.async_search_torrents(site=site, + site_result = await self.async_search_site_torrents(site=site, keyword=keyword, mtype=mediainfo.type if mediainfo else mtype, page=search_page) @@ -2629,7 +2662,7 @@ class SearchChain(ChainBase): for site in indexer_sites: submit_site_page(site=site, page_index=0) - results_count = 0 + results_count = len(plugin_results) try: while tasks: if global_vars.is_system_stopped: diff --git a/tests/test_search_ai_recommend.py b/tests/test_search_ai_recommend.py index a5d3d7f74..ba976f2e0 100644 --- a/tests/test_search_ai_recommend.py +++ b/tests/test_search_ai_recommend.py @@ -55,6 +55,12 @@ class SearchChainAIRecommendTest(unittest.IsolatedAsyncioTestCase): chain.save_cache = lambda _cache, _filename: None chain.remove_cache = lambda _filename: None chain.get_search_page_size = IndexerModule.get_search_page_size + chain.search_plugin_torrents = lambda **_kwargs: [] + + async def no_plugin_results(**_kwargs): + return [] + + chain.async_search_plugin_torrents = no_plugin_results return chain async def test_start_recommend_task_restores_original_indices(self): @@ -185,7 +191,7 @@ class SearchChainAIRecommendTest(unittest.IsolatedAsyncioTestCase): for index in range(count) ] - chain.search_torrents = search_torrents + chain.search_site_torrents = search_torrents with ( patch.object(settings, "SEARCH_RESOURCE_PAGES", 4, create=True), @@ -231,7 +237,7 @@ class SearchChainAIRecommendTest(unittest.IsolatedAsyncioTestCase): for index in range(count) ] - chain.search_torrents = search_torrents + chain.search_site_torrents = search_torrents with ( patch.object(settings, "SEARCH_RESOURCE_PAGES", 3, create=True), @@ -277,7 +283,7 @@ class SearchChainAIRecommendTest(unittest.IsolatedAsyncioTestCase): for index in range(count) ] - chain.search_torrents = search_torrents + chain.search_site_torrents = search_torrents with ( patch.object(settings, "SEARCH_RESOURCE_PAGES", 3, create=True), @@ -342,7 +348,7 @@ class SearchChainAIRecommendTest(unittest.IsolatedAsyncioTestCase): for index in range(count) ] - chain.async_search_torrents = async_search_torrents + chain.async_search_site_torrents = async_search_torrents with ( patch.object(settings, "SEARCH_RESOURCE_PAGES", 4, create=True), @@ -388,7 +394,7 @@ class SearchChainAIRecommendTest(unittest.IsolatedAsyncioTestCase): for index in range(count) ] - chain.async_search_torrents = async_search_torrents + chain.async_search_site_torrents = async_search_torrents with ( patch.object(settings, "SEARCH_RESOURCE_PAGES", 3, create=True), diff --git a/tests/test_search_plugin_sources.py b/tests/test_search_plugin_sources.py new file mode 100644 index 000000000..e45aff69e --- /dev/null +++ b/tests/test_search_plugin_sources.py @@ -0,0 +1,124 @@ +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +from app.chain.search import SearchChain +from app.modules.indexer import IndexerModule +from app.runtime.config import settings + + +def make_chain() -> SearchChain: + """构造不触发完整启动流程的搜索链。""" + chain = object.__new__(SearchChain) + chain.get_search_page_size = IndexerModule.get_search_page_size + return chain + + +def test_search_returns_plugin_results_without_indexer_sites(): + """未配置 PT 站点时，插件资源仍应进入原生资源搜索。""" + chain = make_chain() + plugin_item = SimpleNamespace(title="Plugin Result", description="") + calls = [] + chain.search_plugin_torrents = lambda **kwargs: calls.append(kwargs) or [plugin_item] + + with ( + patch("app.chain.search.get_configured_system_config") as system_config_oper, + patch("app.chain.search.SitesHelper") as sites_helper, + ): + system_config_oper.return_value.get.return_value = [] + sites_helper.return_value.get_indexers.return_value = [] + results = chain._SearchChain__search_all_sites(keyword="keyword") + + assert results == [plugin_item] + assert len(calls) == 1 + assert calls[0]["keyword"] == "keyword" + + +def test_search_invokes_plugin_once_with_multiple_indexers(): + """多个 PT 站点不应导致插件资源源被重复搜索。""" + chain = make_chain() + plugin_calls = [] + site_calls = [] + chain.search_plugin_torrents = lambda **kwargs: plugin_calls.append(kwargs) or [ + SimpleNamespace(title="Plugin Result", description="") + ] + chain.search_site_torrents = lambda **kwargs: site_calls.append(kwargs) or [ + SimpleNamespace(title=f"Site {kwargs['site']['id']}", description="") + ] + + with ( + patch.object(settings, "SEARCH_RESOURCE_PAGES", 1, create=True), + patch("app.chain.search.get_configured_system_config") as system_config_oper, + patch("app.chain.search.SitesHelper") as sites_helper, + patch("app.chain.search.ProgressHelper") as progress_helper, + ): + system_config_oper.return_value.get.return_value = [1, 2] + sites_helper.return_value.get_indexers.return_value = [ + {"id": 1, "name": "站点一"}, + {"id": 2, "name": "站点二"}, + ] + progress_helper.return_value = SimpleNamespace( + start=lambda: None, update=lambda **_kwargs: None, end=lambda: None + ) + results = chain._SearchChain__search_all_sites(keyword="keyword") + + assert len(plugin_calls) == 1 + assert sorted(call["site"]["id"] for call in site_calls) == [1, 2] + assert len(results) == 3 + + +def test_async_search_returns_plugin_results_without_indexers(): + """异步搜索应支持只有插件资源源的部署方式。""" + chain = make_chain() + plugin_item = SimpleNamespace(title="Plugin Result", description="") + calls = [] + + async def plugin_search(**kwargs): + calls.append(kwargs) + return [plugin_item] + + chain.async_search_plugin_torrents = plugin_search + async def run_search(): + with ( + patch("app.chain.search.get_configured_system_config") as system_config_oper, + patch("app.chain.search.SitesHelper") as sites_helper, + ): + system_config_oper.return_value.get.return_value = [] + sites_helper.return_value.async_get_indexers = AsyncMock(return_value=[]) + return await chain._SearchChain__async_search_all_sites(keyword="keyword") + + results = asyncio.run(run_search()) + + assert results == [plugin_item] + assert len(calls) == 1 + + +def test_async_search_stream_emits_plugin_results_once_without_indexers(): + """流式搜索完成事件不应重复发送插件资源。""" + chain = make_chain() + plugin_item = SimpleNamespace(title="Plugin Result", description="") + + async def plugin_search(**_kwargs): + return [plugin_item] + + chain.async_search_plugin_torrents = plugin_search + async def collect_events(): + with ( + patch("app.chain.search.get_configured_system_config") as system_config_oper, + patch("app.chain.search.SitesHelper") as sites_helper, + ): + system_config_oper.return_value.get.return_value = [] + sites_helper.return_value.async_get_indexers = AsyncMock(return_value=[]) + return [ + event + async for event in chain._SearchChain__async_search_all_sites_stream( + keyword="keyword" + ) + ] + + events = asyncio.run(collect_events()) + + assert [event["type"] for event in events] == ["append", "done"] + assert events[0]["items"] == [plugin_item] + assert events[1]["items"] == [] + assert events[1]["total_items"] == 1