fix: 修复音乐搜索进度更新

app/chain/search.py

@@ -1570,6 +1570,72 @@ class SearchChain(ChainBase):
             filter_params=filter_params,
         )
 
+    async def _async_process_music_stream(
+            self,
+            mediainfo: MusicInfo,
+            keyword: Optional[str] = None,
+            sites: Optional[List[int]] = None,
+            rule_groups: Optional[List[str]] = None,
+            filter_params: Optional[Dict[str, str]] = None,
+    ) -> AsyncIterator[dict]:
+        """
+        按音乐元数据渐进式搜索资源，逐站点输出进度并在结束时返回过滤后的完整结果。
+
+        音乐候选需要同时匹配名称、艺术家和音乐分类，因此站点批次只负责推进搜索进度，
+        最终结果仍统一交给音乐上下文构造逻辑过滤、排序和去重。
+        """
+        keywords = [keyword] if keyword else SearchChain.music_site_keywords(mediainfo)
+        torrents: List[TorrentInfo] = []
+        for index, search_word in enumerate(keywords or [mediainfo.title]):
+            if index:
+                await asyncio.sleep(random.randint(1, 10))
+            keyword_matched = False
+            async for event in self.__async_search_all_sites_stream(
+                    keyword=search_word,
+                    mediainfo=mediainfo,
+                    sites=sites,
+                    mtype=MediaType.MUSIC):
+                result = event.pop("items", []) or []
+                matched_torrents = self._matching_music_torrents(result, mediainfo)
+                if matched_torrents:
+                    keyword_matched = True
+                    torrents.extend(matched_torrents)
+                yield {
+                    **event,
+                    "type": "append",
+                    "items": [],
+                    "total_items": len(torrents),
+                }
+            if keyword_matched and not settings.SEARCH_MULTIPLE_NAME:
+                break
+
+        contexts = await run_in_threadpool(
+            self._build_music_contexts,
+            torrents=torrents,
+            mediainfo=mediainfo,
+            rule_groups=rule_groups,
+            filter_params=filter_params,
+        )
+        items = [context.to_dict() for context in contexts]
+        yield {
+            "type": "replace",
+            "stage": "filtered",
+            "value": 100,
+            "text": f"过滤匹配完成，共 {len(contexts)} 个资源",
+            "items": items,
+            "total_items": len(contexts),
+            "candidate_items": len(torrents),
+        }
+        yield {
+            "type": "done",
+            "stage": "done",
+            "text": f"搜索完成，共 {len(contexts)} 个资源",
+            "items": items,
+            "total_items": len(contexts),
+            "candidate_items": len(torrents),
+            "contexts": contexts,
+        }
+
     def process(self, mediainfo: MediaInfo,
                 keyword: Optional[str] = None,
                 no_exists: Dict[int, Dict[int, NotExistMediaInfo]] = None,
@@ -1766,30 +1832,13 @@ class SearchChain(ChainBase):
         """
 
         if mediainfo.type == MediaType.MUSIC:
-            contexts = await self._async_process_music(
-                mediainfo=mediainfo,
-                keyword=keyword,
-                sites=sites,
-                rule_groups=rule_groups,
-                filter_params=filter_params,
-            )
-            items = [context.to_dict() for context in contexts]
-            yield {
-                "type": "replace",
-                "stage": "filtered",
-                "value": 100,
-                "text": f"过滤匹配完成，共 {len(contexts)} 个资源",
-                "items": items,
-                "total_items": len(contexts),
-            }
-            yield {
-                "type": "done",
-                "stage": "done",
-                "text": f"搜索完成，共 {len(contexts)} 个资源",
-                "items": items,
-                "total_items": len(contexts),
-                "contexts": contexts,
-            }
+            async for event in self._async_process_music_stream(
+                    mediainfo=mediainfo,
+                    keyword=keyword,
+                    sites=sites,
+                    rule_groups=rule_groups,
+                    filter_params=filter_params):
+                yield event
             return
 
         # 豆瓣标题处理

tests/test_music_search.py

@@ -1,9 +1,9 @@
+import asyncio
 from unittest.mock import Mock, patch
 
 from app.chain.search import SearchChain
 from app.domain.meta.metamusic import MetaMusic
-from app.domain.context import MusicInfo
-from app.schemas.context import TorrentInfo
+from app.domain.context import MusicInfo, TorrentInfo
 from app.schemas.types import MediaSource, MediaType
 
 
@@ -131,6 +131,80 @@ def test_music_search_matches_artist_from_resource_description():
     assert SearchChain._matching_music_torrents([torrent], music) == [torrent]
 
 
+def test_music_stream_reports_site_progress_before_final_results(monkeypatch):
+    """精确音乐搜索应逐站点输出进度事件，不能等待全部搜索完成后才返回。"""
+    chain = SearchChain()
+    music = MusicInfo(
+        media_source="musicbrainz",
+        media_id="recording-1",
+        title="晴天",
+        artists=["周杰伦"],
+    )
+    unrelated = TorrentInfo(
+        title="其他歌手 - 晴天 FLAC",
+        category=MediaType.MUSIC.value,
+        site_name="Site A",
+    )
+    matched = TorrentInfo(
+        title="周杰伦 - 晴天 FLAC",
+        category=MediaType.MUSIC.value,
+        site_name="Site B",
+    )
+
+    async def search_stream(**_kwargs):
+        """模拟两个站点依次完成并返回各自候选。"""
+        yield {
+            "type": "append",
+            "stage": "searching",
+            "value": 50,
+            "text": "已完成 1 / 2 个请求",
+            "items": [unrelated],
+            "finished": 1,
+            "total": 2,
+        }
+        yield {
+            "type": "append",
+            "stage": "searching",
+            "value": 100,
+            "text": "已完成 2 / 2 个请求",
+            "items": [matched],
+            "finished": 2,
+            "total": 2,
+        }
+
+    async def collect_events():
+        """收集音乐精确搜索流事件。"""
+        return [
+            event
+            async for event in chain.async_process_stream(
+                mediainfo=music,
+                sites=[1, 2],
+                rule_groups=[],
+            )
+        ]
+
+    monkeypatch.setattr(
+        chain,
+        "_SearchChain__async_search_all_sites_stream",
+        search_stream,
+    )
+    monkeypatch.setattr(
+        chain,
+        "_SearchChain__async_search_all_sites",
+        Mock(side_effect=AssertionError("音乐流式搜索不应回退到非流式站点搜索")),
+    )
+
+    with patch("app.chain.search.settings.SEARCH_MULTIPLE_NAME", False):
+        events = asyncio.run(collect_events())
+
+    assert [event["value"] for event in events[:2]] == [50, 100]
+    assert [event["finished"] for event in events[:2]] == [1, 2]
+    assert events[-2]["type"] == "replace"
+    assert events[-1]["type"] == "done"
+    assert events[-1]["total_items"] == 1
+    assert events[-1]["items"][0]["torrent_info"]["title"] == matched.title
+
+
 def test_search_by_id_routes_music_identity_to_recognize_and_process():
     """MusicBrainz 精确身份搜索应经统一识别入口识别后进入现有搜索处理链。"""
     chain = SearchChain()