fix(subscription): preserve batch search execution (#6419)

app/api/dependencies/subscription.py

@@ -95,20 +95,43 @@ def get_delete_subscriptions_by_identity_command(
     )
 
 
+def _start_subscription_search_batch(
+    subscribe_ids: tuple[int, ...] | None,
+    state: str | None,
+) -> None:
+    """把一个请求的搜索目标作为同一调度任务提交。"""
+    if subscribe_ids is None:
+        start_scheduler_job(
+            "subscribe_search",
+            sid=None,
+            state=state,
+            manual=True,
+        )
+        return
+    start_scheduler_job(
+        "subscribe_search",
+        sids=subscribe_ids,
+        state=None,
+        manual=True,
+    )
+
+
 def get_search_subscriptions_command(
     task_registry: TaskRegistry = Depends(get_background_task_registry),
     db: AsyncSession = Depends(get_async_session),
     runtime: HostRuntime = Depends(get_host_runtime),
 ) -> SearchSubscriptionsCommand:
     """组装手工订阅搜索用例，并把调度延迟到响应后的后台任务。"""
-    def schedule_search(subscribe_id: int | None, state: str | None) -> None:
+    def schedule_search(
-        """按历史参数提交订阅搜索调度任务。"""
+        subscribe_ids: tuple[int, ...] | None,
+        state: str | None,
+    ) -> None:
+        """把当前用户的搜索目标提交为一个顺序后台批次。"""
         resolve_background_task_registry(task_registry).create_sync(
-            start_scheduler_job,
+            _start_subscription_search_batch,
-            job_id="subscribe_search",
+            subscribe_ids,
-            sid=subscribe_id,
+            state,
-            state=state,
+            owner="api.subscribe.search",
-            manual=True,
         )
 
     return SearchSubscriptionsCommand(

app/application/subscription/search.py

@@ -29,7 +29,7 @@ class SubscribeSearchRepository(Protocol):
         ...
 
 
-SubscribeSearchScheduler = Callable[[int | None, str | None], None]
+SubscribeSearchScheduler = Callable[[tuple[int, ...] | None, str | None], None]
 
 
 class SearchSubscriptionsCommand:
@@ -54,7 +54,7 @@ class SearchSubscriptionsCommand:
             candidate = await self._repository.get_candidate(subscribe_id)
             if not self._can_access(candidate, actor):
                 return False
-            self._schedule_search(subscribe_id, None)
+            self._schedule_search((subscribe_id,), None)
             return True
 
         if actor.is_superuser:
@@ -65,8 +65,8 @@ class SearchSubscriptionsCommand:
             actor.username,
             "R",
         )
-        for current_id in subscribe_ids:
+        if subscribe_ids:
-            self._schedule_search(current_id, None)
+            self._schedule_search(tuple(subscribe_ids), None)
         return True
 
     @staticmethod

app/chain/subscribe.py

@@ -1512,6 +1512,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
             state: Optional[str] = 'N',
             manual: Optional[bool] = False,
             progress_callback: Optional[Callable[..., None]] = None,
+            sids: Optional[tuple[int, ...]] = None,
     ) -> None:
         """
         执行订阅搜索。
@@ -1520,6 +1521,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
         """
         return self._execute_search(
             sid=sid,
+            sids=sids,
             state=state,
             manual=manual,
             progress_callback=progress_callback,
@@ -1531,6 +1533,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
             state: Optional[str] = 'N',
             manual: Optional[bool] = False,
             progress_callback: Optional[Callable[..., None]] = None,
+            sids: Optional[tuple[int, ...]] = None,
     ) -> None:
         """
         订阅搜索
@@ -1538,6 +1541,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
         :param state: 订阅状态 N:新建, R:订阅中, P:待定, S:暂停
         :param manual: 是否手动搜索
         :param progress_callback: 定时服务进度更新回调
+        :param sids: 订阅ID集合，有值时按给定顺序处理
         :return: 更新订阅状态为R或删除订阅
         """
         lock_acquired = False
@@ -1550,9 +1554,16 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
             if sid:
                 subscribe = subscribeoper.get(sid)
                 subscribes = [subscribe] if subscribe else []
+            elif sids is not None:
+                subscribes = [
+                    subscribe
+                    for current_id in sids
+                    if (subscribe := subscribeoper.get(current_id)) is not None
+                ]
             else:
                 subscribes = subscribeoper.list(self.get_states_for_search(state))
             total_num = len(subscribes)
+            processed_subscribes = []
             if progress_callback:
                 progress_callback(
                     value=0,
@@ -1565,6 +1576,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
                 for index, subscribe in enumerate(subscribes, start=1):
                     if global_vars.is_system_stopped:
                         break
+                    processed_subscribes.append(subscribe)
                     if progress_callback:
                         progress_callback(
                             value=(index - 1) / total_num * 100 if total_num else 100,
@@ -1586,7 +1598,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
                             logger.debug(f"订阅标题：{subscribe.name} 新增小于1分钟，暂不搜索...")
                             continue
                     # 随机休眠1-5分钟
-                    if not sid and state in ['R', 'P']:
+                    if not sid and sids is None and state in ['R', 'P']:
                         sleep_time = random.randint(60, 300)
                         logger.info(f'订阅搜索随机休眠 {sleep_time} 秒 ...')
                         if progress_callback:
@@ -1757,6 +1769,13 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
                     if subscribes:
                         if sid:
                             self.messagehelper.put(f'{subscribes[0].name} 搜索完成！', title="订阅搜索", role="system")
+                        elif sids is not None:
+                            for subscribe in processed_subscribes:
+                                self.messagehelper.put(
+                                    f'{subscribe.name} 搜索完成！',
+                                    title="订阅搜索",
+                                    role="system",
+                                )
                         else:
                             self.messagehelper.put('所有订阅搜索完成！', title="订阅搜索", role="system")
                     else:

tests/test_subscribe_search_command.py

@@ -1,10 +1,15 @@
+from types import SimpleNamespace
+from unittest.mock import Mock
+
 import pytest
 
+from app.api.dependencies import subscription as subscription_dependencies
 from app.application.subscription.delete import SubscribeDeletionCandidate
 from app.application.subscription.search import (
     SearchSubscriptionsCommand,
     SubscribeSearchActor,
 )
+from app.runtime.tasks import TaskRegistry
 
 
 class _Repository:
@@ -52,7 +57,7 @@ async def test_superuser_search_all_uses_single_global_scheduler_request():
 
 @pytest.mark.asyncio
 async def test_regular_user_search_all_schedules_only_owned_subscriptions():
-    """普通用户搜索全部时逐条提交仓储已按归属过滤的订阅。"""
+    """普通用户搜索全部时把归属订阅合并为一次后台批次。"""
     scheduled = []
     command = SearchSubscriptionsCommand(
         repository=_Repository(subscribe_ids=[2, 5]),
@@ -62,7 +67,22 @@ async def test_regular_user_search_all_schedules_only_owned_subscriptions():
     assert await command.execute(
         SubscribeSearchActor(username="alice", is_superuser=False)
     ) is True
-    assert scheduled == [(2, None), (5, None)]
+    assert scheduled == [((2, 5), None)]
+
+
+@pytest.mark.asyncio
+async def test_regular_user_search_all_with_no_targets_does_not_schedule():
+    """普通用户没有可搜索订阅时不创建空后台任务。"""
+    scheduled = []
+    command = SearchSubscriptionsCommand(
+        repository=_Repository(),
+        schedule_search=lambda ids, state: scheduled.append((ids, state)),
+    )
+
+    assert await command.execute(
+        SubscribeSearchActor(username="alice", is_superuser=False)
+    ) is True
+    assert scheduled == []
 
 
 @pytest.mark.asyncio
@@ -94,4 +114,49 @@ async def test_targeted_search_schedules_accessible_subscription():
         SubscribeSearchActor(username="alice", is_superuser=False),
         subscribe_id=7,
     ) is True
-    assert scheduled == [(7, None)]
+    assert scheduled == [((7,), None)]
+
+
+def test_subscription_search_batch_uses_one_scheduler_generation(monkeypatch):
+    """一个后台批次只占用一次调度任务运行权。"""
+    calls = []
+    monkeypatch.setattr(
+        subscription_dependencies,
+        "start_scheduler_job",
+        lambda job_id, **kwargs: calls.append((job_id, kwargs)),
+    )
+
+    subscription_dependencies._start_subscription_search_batch((2, 5), None)
+
+    assert calls == [
+        (
+            "subscribe_search",
+            {"sids": (2, 5), "state": None, "manual": True},
+        ),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_search_dependency_registers_one_owned_background_batch():
+    """请求适配器只登记一个具名后台批次。"""
+    registry = TaskRegistry()
+    registry.create_sync = Mock()
+    repository = _Repository(subscribe_ids=[2, 5])
+    runtime = SimpleNamespace(
+        subscription=SimpleNamespace(repository=lambda _db: repository),
+    )
+    command = subscription_dependencies.get_search_subscriptions_command(
+        task_registry=registry,
+        db=object(),
+        runtime=runtime,
+    )
+
+    assert await command.execute(
+        SubscribeSearchActor(username="alice", is_superuser=False)
+    ) is True
+    registry.create_sync.assert_called_once_with(
+        subscription_dependencies._start_subscription_search_batch,
+        (2, 5),
+        None,
+        owner="api.subscribe.search",
+    )

tests/test_subscribe_search_state.py

@@ -104,6 +104,34 @@ def test_new_subscribe_search_marks_state_after_attempt(monkeypatch) -> None:
     assert _SubscribeOper.updates == [(31, {"state": "R"})]
 
 
+def test_targeted_batch_searches_all_ids_without_state_scan(monkeypatch) -> None:
+    """用户归属订阅批次只按指定 ID 顺序读取，不扩大为全局状态搜索。"""
+    first = _new_subscribe(datetime.now() - timedelta(minutes=2))
+    first.state = "R"
+    second = _new_subscribe(datetime.now() - timedelta(minutes=2))
+    second.id = 32
+    second.name = "测试电影 2"
+    second.state = "R"
+    subscribes = {first.id: first, second.id: second}
+    subscribe_oper = Mock()
+    subscribe_oper.get.side_effect = subscribes.get
+    monkeypatch.setattr(
+        subscribe_module,
+        "SubscribeOper",
+        lambda: subscribe_oper,
+    )
+    media_chain = Mock()
+    media_chain.recognize_media.return_value = None
+
+    with patch.object(subscribe_module, "MediaChain", return_value=media_chain):
+        chain = object.__new__(SubscribeChain)
+        chain.search(sids=(31, 32), state=None, manual=False)
+
+    assert [item.args for item in subscribe_oper.get.call_args_list] == [(31,), (32,)]
+    subscribe_oper.list.assert_not_called()
+    assert media_chain.recognize_media.call_count == 2
+
+
 def test_subscribe_search_aborts_when_lock_times_out(monkeypatch) -> None:
     """订阅搜索锁超时后必须中止，不能在无锁状态下继续访问订阅。"""
     monkeypatch.setattr(SubscribeChain, "_rlock", _TimedOutLock())