diff --git a/app/api/dependencies/subscription.py b/app/api/dependencies/subscription.py index 9161e72d2..c82ea88f0 100644 --- a/app/api/dependencies/subscription.py +++ b/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, - job_id="subscribe_search", - sid=subscribe_id, - state=state, - manual=True, + _start_subscription_search_batch, + subscribe_ids, + state, + owner="api.subscribe.search", ) return SearchSubscriptionsCommand( diff --git a/app/application/subscription/search.py b/app/application/subscription/search.py index ba2580984..05a15d3c5 100644 --- a/app/application/subscription/search.py +++ b/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: - self._schedule_search(current_id, None) + if subscribe_ids: + self._schedule_search(tuple(subscribe_ids), None) return True @staticmethod diff --git a/app/chain/subscribe.py b/app/chain/subscribe.py index b5bc6e82e..887f9bb0e 100644 --- a/app/chain/subscribe.py +++ b/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: diff --git a/tests/test_subscribe_search_command.py b/tests/test_subscribe_search_command.py index c68ba80de..5fa18093b 100644 --- a/tests/test_subscribe_search_command.py +++ b/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", + ) diff --git a/tests/test_subscribe_search_state.py b/tests/test_subscribe_search_state.py index 272cf94d0..8a4f2ab94 100644 --- a/tests/test_subscribe_search_state.py +++ b/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())