From 0a732be485e10787141b5b5da73d2d64b5834be4 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 5 Sep 2026 17:23:26 +0800 Subject: [PATCH] feat(subscribe): improve search scheduling feedback --- app/api/dependencies/subscription.py | 66 +++--- app/api/endpoints/subexecution.py | 12 +- app/api/endpoints/submaintenance.py | 68 +++++- app/application/subscription/contract.py | 8 +- app/application/subscription/execution.py | 9 +- app/application/subscription/observability.py | 144 ++++++++---- app/application/subscription/search.py | 59 +++-- app/application/subscription/sitebudget.py | 22 +- app/application/subscription/status.py | 8 +- app/chain/subscribe/contract.py | 3 +- app/chain/subscribe/match.py | 51 ++-- app/chain/subscribe/notify.py | 18 ++ app/chain/subscribe/search.py | 217 +++++++++++++----- app/db/adapters/subscription.py | 15 +- app/db/adapters/subscriptionsearch.py | 33 ++- app/db/adapters/subscriptionstatus.py | 1 + app/db/oper/subscribe.py | 10 +- app/db/oper/subscriptionsearch.py | 46 +++- app/locales/en-US.json | 43 +++- app/locales/zh-TW.json | 43 +++- app/scheduler/catalog.py | 4 +- app/schemas/exports.py | 1 + app/schemas/subscribe.py | 12 + .../subscription_governance_scale.py | 8 +- tests/test_api_background_task_registry.py | 62 ++++- tests/test_subscribe_endpoint.py | 15 +- tests/test_subscribe_search_command.py | 161 +++++++++---- tests/test_subscribe_search_state.py | 6 +- tests/test_subscription_execution_status.py | 45 +++- tests/test_subscription_match_admission.py | 32 ++- tests/test_subscription_search_governance.py | 177 ++++++++++---- tests/test_subscription_search_queue.py | 43 +++- tests/test_subscription_site_budget.py | 3 + 33 files changed, 1080 insertions(+), 365 deletions(-) diff --git a/app/api/dependencies/subscription.py b/app/api/dependencies/subscription.py index 27748d861..b9295e0bc 100644 --- a/app/api/dependencies/subscription.py +++ b/app/api/dependencies/subscription.py @@ -42,7 +42,10 @@ from app.application.subscription.mutation import ( SubscriptionMutationService, ) from app.application.subscription.query import SubscriptionQueryService -from app.application.subscription.search import SearchSubscriptionsCommand +from app.application.subscription.search import ( + SearchSubscriptionsCommand, + SubscriptionSearchSubmission, +) from app.application.subscription.status import ( SubscriptionExecutionReadRepository, SubscriptionExecutionStatusService, @@ -120,24 +123,14 @@ def get_delete_subscriptions_by_identity_command( ) -def _start_subscription_search_batch( - subscribe_ids: tuple[int, ...] | None, - state: str | None, +def _resume_submitted_subscription_search( + subscribe_ids: tuple[int, ...], ) -> 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, + "subscribe_search_queue", + limit=max(1, len(subscribe_ids)), + manual_sids=subscribe_ids, ) @@ -146,23 +139,40 @@ def get_search_subscriptions_command( db: AsyncSession = Depends(get_async_session), runtime: HostRuntime = Depends(get_host_runtime), ) -> SearchSubscriptionsCommand: - """组装手工订阅搜索用例，并把调度延迟到响应后的后台任务。""" + """组装手工搜索用例，请求内只入队，实际搜索交给后台继续。""" - def schedule_search( - subscribe_ids: tuple[int, ...] | None, - state: str | None, - ) -> None: - """把当前用户的搜索目标提交为一个顺序后台批次。""" - resolve_background_task_registry(task_registry).create_sync( - _start_subscription_search_batch, + registry = resolve_background_task_registry(task_registry) + search_repository = get_subscription_search_repository(runtime) + + async def submit_search( + subscribe_ids: tuple[int, ...], + single: bool, + ) -> SubscriptionSearchSubmission: + """在线程 owner 中完成轻量入队，并返回前端可立即跟踪的批次。""" + enqueue_task = registry.create_sync( + search_repository.enqueue, + subscription_ids=subscribe_ids, + source="manual", + priority=120 if single else 100, + owner="api.subscribe.search.enqueue", + ) + enqueued = await enqueue_task + registry.create_sync( + _resume_submitted_subscription_search, subscribe_ids, - state, - owner="api.subscribe.search", + owner="api.subscribe.search.run", + ) + return SubscriptionSearchSubmission( + batch_ids=enqueued.active_batch_ids, + target_count=len(subscribe_ids), + queued_count=enqueued.created_count, + ongoing_count=enqueued.coalesced_count, + single=single, ) return SearchSubscriptionsCommand( repository=runtime.subscription.repository(db), - schedule_search=schedule_search, + submit_search=submit_search, ) diff --git a/app/api/endpoints/subexecution.py b/app/api/endpoints/subexecution.py index 52cad52fa..12a1b7aa8 100644 --- a/app/api/endpoints/subexecution.py +++ b/app/api/endpoints/subexecution.py @@ -32,7 +32,7 @@ async def _accessible_subscription_ids( @router.get( # type: ignore[misc] "/execution/batches", - summary="查询订阅搜索批次状态", + summary="查看订阅搜索进度", response_model=List[_SchemaSubscriptionBatchStatus], ) async def list_subscription_execution_batches( @@ -52,7 +52,7 @@ async def list_subscription_execution_batches( @router.get( # type: ignore[misc] "/execution/batches/{batch_id}", - summary="查询订阅搜索批次", + summary="查看一次订阅搜索", response_model=_SchemaSubscriptionBatchStatus, ) async def get_subscription_execution_batch( @@ -68,13 +68,13 @@ async def get_subscription_execution_batch( accessible_subscription_ids=accessible_ids, ) if batch is None: - raise HTTPException(status_code=404, detail="订阅搜索批次不存在") + raise HTTPException(status_code=404, detail="没有找到这次搜索，请刷新后重试") return _SchemaSubscriptionBatchStatus.model_validate(batch) @router.put( # type: ignore[misc] "/execution/batches/{batch_id}/cancel", - summary="取消订阅搜索批次", + summary="停止一次订阅搜索", response_model=_SchemaResponse[None], ) async def cancel_subscription_execution_batch( @@ -90,9 +90,9 @@ async def cancel_subscription_execution_batch( accessible_subscription_ids=accessible_ids, ) if batch is None: - return _SchemaResponse(success=False, message="订阅搜索批次不存在") + return _SchemaResponse(success=False, message="没有找到这次搜索，请刷新后重试") cancelled = await status_service.request_cancel(batch_id) return _SchemaResponse( success=bool(cancelled), - message="" if cancelled else "订阅搜索批次已结束或无法取消", + message="" if cancelled else "这次搜索已经结束，暂时无法停止", ) diff --git a/app/api/endpoints/submaintenance.py b/app/api/endpoints/submaintenance.py index b7906797a..9023c2aa5 100644 --- a/app/api/endpoints/submaintenance.py +++ b/app/api/endpoints/submaintenance.py @@ -22,12 +22,50 @@ from app.application.subscription.mutation import ( from app.application.subscription.search import ( SearchSubscriptionsCommand, SubscribeSearchActor, + SubscriptionSearchSubmission, ) from app.schemas.response import Response +from app.schemas.subscribe import ( + SubscriptionSearchSubmission as SubscriptionSearchSubmissionSchema, +) router = ResponseAPIRouter() +def _search_submission_message(submission: SubscriptionSearchSubmission) -> str: + """把搜索安排结果转换为用户能直接理解的提示。""" + if submission.target_count == 0: + return "没有需要搜索的订阅" + if submission.queued_count == 0: + return ( + "这个订阅已经在搜索中，请稍候" + if submission.single + else f"{submission.ongoing_count} 个订阅已经在搜索中，无需重复提交" + ) + if submission.ongoing_count: + return ( + f"已安排 {submission.queued_count} 个订阅搜索，" + f"另有 {submission.ongoing_count} 个正在处理中" + ) + if submission.single: + return "已安排搜索，很快开始" + return f"已安排 {submission.queued_count} 个订阅搜索，系统会依次处理" + + +def _search_submission_schema( + submission: SubscriptionSearchSubmission, +) -> SubscriptionSearchSubmissionSchema: + """构造稳定的手工搜索响应数据。""" + return SubscriptionSearchSubmissionSchema( + batch_id=submission.batch_id, + batch_ids=list(submission.batch_ids), + target_count=submission.target_count, + queued_count=submission.queued_count, + ongoing_count=submission.ongoing_count, + single=submission.single, + ) + + @router.get( # type: ignore[misc] "/refresh", summary="刷新订阅（兼容入口）", @@ -97,38 +135,46 @@ def check_subscribes( @router.get( # type: ignore[misc] "/search", summary="搜索所有订阅（兼容入口）", - response_model=Response[None], + response_model=Response[SubscriptionSearchSubmissionSchema], include_in_schema=False, deprecated=True, ) @router.post( # type: ignore[misc] - "/search", summary="搜索所有订阅", response_model=Response[None] + "/search", + summary="搜索所有订阅", + response_model=Response[SubscriptionSearchSubmissionSchema], ) async def search_subscribes( command: SearchSubscriptionsCommand = Depends(get_search_subscriptions_command), current_user: ApiPrincipal = Depends(get_current_active_user_async), ) -> Any: """搜索当前用户可管理的全部订阅。""" - await command.execute( + submission = await command.execute( SubscribeSearchActor( username=current_user.name, is_superuser=current_user.is_superuser, ) ) - return Response(success=True) + if submission is None: + return Response(success=False, message="没有需要搜索的订阅") + return Response( + success=True, + message=_search_submission_message(submission), + data=_search_submission_schema(submission), + ) @router.get( # type: ignore[misc] "/search/{subscribe_id}", summary="搜索订阅（兼容入口）", - response_model=Response[None], + response_model=Response[SubscriptionSearchSubmissionSchema], include_in_schema=False, deprecated=True, ) @router.post( # type: ignore[misc] "/search/{subscribe_id}", summary="搜索订阅", - response_model=Response[None], + response_model=Response[SubscriptionSearchSubmissionSchema], ) async def search_subscribe( subscribe_id: int, @@ -136,13 +182,17 @@ async def search_subscribe( current_user: ApiPrincipal = Depends(get_current_active_user_async), ) -> Any: """根据订阅编号搜索一个订阅。""" - found = await command.execute( + submission = await command.execute( SubscribeSearchActor( username=current_user.name, is_superuser=current_user.is_superuser, ), subscribe_id=subscribe_id, ) - if not found: + if submission is None: return Response(success=False, message="订阅不存在") - return Response(success=True) + return Response( + success=True, + message=_search_submission_message(submission), + data=_search_submission_schema(submission), + ) diff --git a/app/application/subscription/contract.py b/app/application/subscription/contract.py index facd28a3a..871c67d94 100644 --- a/app/application/subscription/contract.py +++ b/app/application/subscription/contract.py @@ -596,8 +596,12 @@ class SubscriptionStagingPort(Protocol): """异步按媒体身份读取删除候选快照。""" ... - async def list_search_ids(self, username: str, state: str) -> builtins.list[int]: - """异步读取用户可搜索订阅主键。""" + async def list_search_ids( + self, + username: Optional[str], + state: str, + ) -> builtins.list[int]: + """异步读取用户或管理员全局范围内可搜索的订阅主键。""" ... async def stage_delete(self, subscribe_id: int) -> None: diff --git a/app/application/subscription/execution.py b/app/application/subscription/execution.py index 063919dda..72a815b15 100644 --- a/app/application/subscription/execution.py +++ b/app/application/subscription/execution.py @@ -125,11 +125,13 @@ def handle_subscription_search_deferred( deferred: SubscriptionSearchDeferred, record: Callable[..., None], ) -> None: - """把站点预算冲突重新入队，并记录为可恢复而非失败的任务结果。""" + """把站点暂时不可用的任务重新入队，而不是记录为搜索失败。""" requeued = queue.defer_task( task_id=task_id, lease_token=lease_token, available_at=deferred.retry_at, + phase="waiting_site_budget", + message="站点暂时忙，系统会自动继续搜索", ) if requeued: record("requeued", "site_budget_deferred") @@ -187,6 +189,7 @@ class SearchEnqueueResult: batch: SearchBatchSnapshot created_count: int coalesced_count: int + active_batch_ids: tuple[str, ...] class SubscriptionSearchRepository(Protocol): @@ -245,8 +248,10 @@ class SubscriptionSearchRepository(Protocol): task_id: str, lease_token: str, available_at: str, + phase: str = "waiting_site_budget", + message: Optional[str] = None, ) -> bool: - """把临时站点预算冲突任务退回队列，并设置下一次领取时间。""" + """把临时不可执行任务退回队列，并保留用户可理解的等待原因。""" ... def is_cancel_requested(self, task_id: str) -> bool: diff --git a/app/application/subscription/observability.py b/app/application/subscription/observability.py index f6da645e6..a72c180f8 100644 --- a/app/application/subscription/observability.py +++ b/app/application/subscription/observability.py @@ -16,30 +16,73 @@ from app.application.subscription.sitebudget import SubscriptionSiteBudgetMetric SearchTaskOutcome = Literal["completed", "skipped", "failed", "cancelled", "requeued"] +_SEARCH_SOURCE_NAMES = { + "manual": "手动订阅搜索", + "targeted": "指定订阅搜索", + "new": "新订阅自动搜索", + "fallback": "订阅定时检查", + "resume": "等待中的订阅搜索", + "inline": "订阅搜索", +} + +_SEARCH_STATE_NAMES = { + "completed": "已完成", + "failed": "部分订阅没有完成", + "cancelled": "已停止", + "skipped": "部分订阅这次未搜索", + "queued": "部分订阅稍后继续", + "running": "仍在处理中", + "cancelling": "正在停止", + "stopped": "部分订阅稍后继续", +} + +_MATCH_STATE_NAMES = { + "completed": "检查完成", + "failed": "部分订阅没有完成", + "cancelled": "已停止", + "skipped": "部分订阅这次未检查", + "stopped": "已停止，部分订阅这次未检查", +} + + +def _search_source_name(source: str) -> str: + """返回搜索来源对应的可读名称。""" + return _SEARCH_SOURCE_NAMES.get(source, "订阅搜索") + + +def _search_state_name(state: str) -> str: + """返回搜索结束状态对应的可读说明。""" + return _SEARCH_STATE_NAMES.get(state, "已结束") + + +def _match_state_name(state: str) -> str: + """返回订阅资源检查状态对应的可读说明。""" + return _MATCH_STATE_NAMES.get(state, "已结束") + def batch_progress_text(batch: Optional[SearchBatchSnapshot]) -> str: """把批次聚合终态转为兼容进度文案。""" if batch is None: - return "订阅搜索任务已提交" + return "搜索已安排" if batch.state == "failed": - return "订阅搜索完成，部分任务失败" + return "搜索结束，部分订阅没有完成" if batch.state == "cancelled": - return "订阅搜索已取消" + return "搜索已停止" if batch.state == "skipped": - return "订阅搜索完成，部分任务本轮已跳过" + return "搜索结束，部分订阅这次未搜索" if batch.state in {"queued", "running", "cancelling"}: - return "订阅搜索任务已排队" + return "搜索已安排，系统正在依次处理" if batch.skipped_count: - return "订阅搜索完成，部分任务本轮已跳过" - return "订阅搜索完成" + return "搜索结束，部分订阅这次未搜索" + return "搜索完成" def inline_search_result(total: int, finished: int) -> tuple[str, dict[str, int]]: """返回兼容搜索的真实终态文案与计数。""" text = ( - "订阅搜索完成" + "搜索完成" if finished == total - else "订阅搜索结束，部分订阅本轮未执行或未完成" + else "搜索结束，部分订阅这次没有完成" ) return text, {"total": total, "finished": finished} @@ -77,14 +120,14 @@ def finish_returned_search_task( task_id=task_id, lease_token=lease_token, state="completed", - error="执行截止时间晚于下载提交边界，已按实际结果完成", + error="下载已经提交，虽然搜索用时较长，结果仍然有效", ) return subscription_id, "completed", "ttl_timeout" queue.finish_task( task_id=task_id, lease_token=lease_token, state="failed", - error="订阅执行已超过协作截止时间", + error="这次搜索用时过长，已停止，可稍后重试", ) return None, "failed", "ttl_timeout" if system_stopped: @@ -93,7 +136,7 @@ def finish_returned_search_task( task_id=task_id, lease_token=lease_token, state="completed", - error="停机请求晚于下载提交边界，已按实际结果完成", + error="下载已经提交，系统停止后仍保留这次结果", ) return subscription_id, "completed", "system_stop" queue.release_task(task_id=task_id, lease_token=lease_token) @@ -104,7 +147,7 @@ def finish_returned_search_task( task_id=task_id, lease_token=lease_token, state="completed", - error="取消请求晚于下载提交边界，已按实际结果完成", + error="下载已经提交，无法撤回，已保留这次结果", ) return subscription_id, "completed", "cancelled" queue.release_task(task_id=task_id, lease_token=lease_token, cancelled=True) @@ -173,26 +216,31 @@ class MatchExecutionSummary: return "completed" def start_log(self) -> str: - """构造 Match 轮次开始日志。""" + """构造不暴露内部标记的资源检查开始日志。""" return ( - "订阅治理轮次开始: operation=match " - f"run_id={self.run_id} subscriptions={self.total} " - f"sites={self.site_count} candidates={self.candidate_count}" + f"开始检查订阅资源，共 {self.total} 个订阅、{self.candidate_count} 个资源，" + f"来自 {self.site_count} 个站点。" ) def finish_log(self) -> str: - """构造 Match 轮次结束日志，completed 表示任务处理完成而非订阅成功。""" - elapsed_ms = (time.perf_counter() - self.started_at) * 1000 - return ( - "订阅治理轮次结束: operation=match " - f"run_id={self.run_id} state={self.state} subscriptions={self.total} " - f"processed={self.finished} task_completed={self.completed} " - f"task_skipped={self.skipped} task_failed={self.failed} " - f"admission_conflicts={self.admission_conflicts} cancelled={self.cancelled} " - f"ttl_timeouts={self.ttl_timeouts} release_failures={self.release_failures} " - f"sites={self.site_count} candidates={self.candidate_count} " - f"duration_ms={elapsed_ms:.1f}" + """构造可直接阅读的资源检查结束日志。""" + elapsed_seconds = time.perf_counter() - self.started_at + message = ( + f"订阅资源检查结束：{_match_state_name(self.state)}。" + f"本次检查 {self.finished}/{self.total} 个订阅，完成 {self.completed} 个，" + f"这次未检查 {self.skipped} 个，失败 {self.failed} 个；" + f"共检查 {self.candidate_count} 个资源，来自 {self.site_count} 个站点，" + f"用时 {elapsed_seconds:.1f} 秒。" ) + if self.admission_conflicts: + message += f"其中 {self.admission_conflicts} 个订阅正在处理中，本次没有重复检查。" + if self.cancelled: + message += f"另有 {self.cancelled} 个订阅已停止。" + if self.ttl_timeouts: + message += f"另有 {self.ttl_timeouts} 个订阅检查时间过长，已停止。" + if self.release_failures: + message += f"另有 {self.release_failures} 个订阅的搜索状态没有正常恢复，系统稍后会继续检查。" + return message def as_data(self, current: Optional[int] = None) -> dict[str, int]: """构造供进度消费者读取的实际计数。""" @@ -256,15 +304,14 @@ class SearchExecutionSummary: self.ttl_timeouts += 1 def start_log(self) -> str: - """构造 Search 轮次开始日志。""" - return ( - "订阅治理轮次开始: operation=search " - f"run_id={self.run_id} batch_id={self.batch_id or '-'} source={self.source} " - f"subscriptions={self.requested} coalesced={self.coalesced}" - ) + """构造不暴露内部术语的 Search 开始日志。""" + message = f"开始{_search_source_name(self.source)}，共 {self.requested} 个订阅" + if self.coalesced: + message += f"，其中 {self.coalesced} 个已经在处理中" + return f"{message}。" def finish_log(self, batch: Optional[SearchBatchSnapshot] = None) -> str: - """构造 Search 轮次结束日志，任务完成与订阅完成保持分离。""" + """构造可直接阅读的 Search 结束日志。""" sites = self.site_metrics.snapshot() if self.round_failed: state = "failed" @@ -282,19 +329,16 @@ class SearchExecutionSummary: state = "skipped" else: state = "completed" - elapsed_ms = (time.perf_counter() - self.started_at) * 1000 - return ( - "订阅治理轮次结束: operation=search " - f"run_id={self.run_id} batch_id={self.batch_id or '-'} source={self.source} " - f"state={state} subscriptions={self.requested} processed={self.processed} " - f"task_completed={self.completed} task_skipped={self.skipped} " - f"task_failed={self.failed} task_cancelled={self.cancelled} task_requeued={self.requeued} " - f"admission_conflicts={self.admission_conflicts} cancelled={self.cancelled} " - f"ttl_timeouts={self.ttl_timeouts} consumer_conflicts={self.consumer_conflicts} " - f"round_failed={int(self.round_failed)} " - f"sites={sites.site_count} site_requests={sites.request_count} " - f"site_failures={sites.failure_count} site_cooldown_skips={sites.cooldown_skip_count} " - f"candidates={sites.candidate_count} cooldown_seconds={sites.cooldown_seconds:.1f} " - f"release_failures={self.release_failures + sites.release_failure_count} " - f"duration_ms={elapsed_ms:.1f}" + elapsed_seconds = time.perf_counter() - self.started_at + message = ( + f"{_search_source_name(self.source)}结束：{_search_state_name(state)}。" + f"本次处理 {self.processed}/{self.requested} 个，完成 {self.completed} 个，" + f"这次未搜索 {self.skipped} 个，失败 {self.failed} 个，" + f"稍后继续 {self.requeued} 个，已停止 {self.cancelled} 个；" + f"访问 {sites.site_count} 个站点，发出 {sites.request_count} 次请求，" + f"找到 {sites.candidate_count} 个资源，用时 {elapsed_seconds:.1f} 秒。" ) + unfinished_cleanup = self.release_failures + sites.release_failure_count + if unfinished_cleanup: + message += f"另有 {unfinished_cleanup} 个搜索状态没有正常恢复，系统稍后会继续处理。" + return message diff --git a/app/application/subscription/search.py b/app/application/subscription/search.py index a71821800..eb8c4dc3b 100644 --- a/app/application/subscription/search.py +++ b/app/application/subscription/search.py @@ -1,7 +1,8 @@ """手工订阅搜索应用用例。""" +from collections.abc import Awaitable, Callable from dataclasses import dataclass -from typing import Callable +from typing import Optional from app.application.subscription.contract import ( SubscribeDeletionCandidate, @@ -17,45 +18,65 @@ class SubscribeSearchActor: is_superuser: bool -SubscribeSearchScheduler = Callable[[tuple[int, ...] | None, str | None], None] +@dataclass(frozen=True, slots=True) +class SubscriptionSearchSubmission: + """一次手工搜索请求的入队结果。""" + + batch_ids: tuple[str, ...] + target_count: int + queued_count: int + ongoing_count: int + single: bool + + @property + def batch_id(self) -> Optional[str]: + """返回最适合前端立即跟踪的批次编号。""" + return self.batch_ids[0] if self.batch_ids else None + + +SubscribeSearchSubmitter = Callable[ + [tuple[int, ...], bool], + Awaitable[SubscriptionSearchSubmission], +] class SearchSubscriptionsCommand: - """按用户权限生成并提交手工订阅搜索任务。""" + """按用户权限读取目标，并把手工搜索轻量提交到持久队列。""" def __init__( self, repository: SubscriptionStagingPort, - schedule_search: SubscribeSearchScheduler, + submit_search: SubscribeSearchSubmitter, ) -> None: - """注入订阅读取端口和后台任务提交端口。""" + """注入订阅读取端口和持久队列提交端口。""" self._repository = repository - self._schedule_search = schedule_search + self._submit_search = submit_search async def execute( self, actor: SubscribeSearchActor, subscribe_id: int | None = None, - ) -> bool: - """提交单条或当前用户全部可搜索订阅，返回目标是否存在。""" + ) -> Optional[SubscriptionSearchSubmission]: + """提交单条或当前用户全部可搜索订阅，目标不可访问时返回空。""" if subscribe_id is not None: candidate = await self._repository.get_candidate(subscribe_id) if not self._can_access(candidate, actor): - return False - self._schedule_search((subscribe_id,), None) - return True - - if actor.is_superuser: - self._schedule_search(None, "R") - return True + return None + return await self._submit_search((subscribe_id,), True) subscribe_ids = await self._repository.list_search_ids( - actor.username, + None if actor.is_superuser else actor.username, "R", ) - if subscribe_ids: - self._schedule_search(tuple(subscribe_ids), None) - return True + if not subscribe_ids: + return SubscriptionSearchSubmission( + batch_ids=(), + target_count=0, + queued_count=0, + ongoing_count=0, + single=False, + ) + return await self._submit_search(tuple(subscribe_ids), False) @staticmethod def _can_access( diff --git a/app/application/subscription/sitebudget.py b/app/application/subscription/sitebudget.py index 83c8e1f46..ff3c38e14 100644 --- a/app/application/subscription/sitebudget.py +++ b/app/application/subscription/sitebudget.py @@ -27,20 +27,28 @@ class SubscriptionSearchDeferred(RuntimeError): """表示订阅搜索未失败，而是应在站点预算可用后重新入队。""" def __init__(self, *, retry_at: str, site_ids: tuple[int, ...]) -> None: - """保存队列恢复所需的时间和冲突站点，避免把临时冲突写成错误。""" - super().__init__(f"订阅搜索已延后，站点预算最早可重试：{retry_at}") + """保存队列恢复所需的时间和站点，避免把临时等待写成错误。""" + super().__init__("站点暂时忙，系统会自动继续搜索") self.retry_at = retry_at self.site_ids = site_ids class SubscriptionSiteBudgetUnavailable(RuntimeError): - """表示站点预算暂时不可用，调用方应记录为延后而非失败。""" + """表示站点暂时不可用，调用方应记录为等待而非失败。""" - def __init__(self, *, site_id: int, retry_at: str) -> None: + def __init__( + self, + *, + site_id: int, + retry_at: str, + wait_reason: Optional[str] = None, + ) -> None: """保存站点和下一次可尝试时间，供订阅队列恢复。""" - super().__init__(f"站点 {site_id} 冷却或已有在途搜索，最早可重试：{retry_at}") + message = "站点正在处理其他搜索，稍后会自动继续" if wait_reason == "busy" else "站点暂时不可用，稍后会自动重试" + super().__init__(message) self.site_id = site_id self.retry_at = retry_at + self.wait_reason = wait_reason @dataclass(frozen=True, slots=True) @@ -52,6 +60,7 @@ class SiteBudgetClaim: retry_at: str consecutive_failures: int lease_token: Optional[str] = None + wait_reason: Optional[str] = None @dataclass(frozen=True, slots=True) @@ -195,6 +204,7 @@ class SubscriptionSiteBudget: raise SubscriptionSiteBudgetUnavailable( site_id=site_id, retry_at=claim.retry_at, + wait_reason=claim.wait_reason, ) def _report_phase(self, phase: str, site_id: Optional[int]) -> None: @@ -251,4 +261,4 @@ class SubscriptionSiteBudget: def _raise_if_cancelled(self) -> None: """在创建站点租约前传播取消或停机。""" if self._stop_state.is_system_stopped or self._cancelled(): - raise SubscriptionSearchCancelled("订阅搜索已取消") + raise SubscriptionSearchCancelled("搜索已停止") diff --git a/app/application/subscription/status.py b/app/application/subscription/status.py index 4d51ca58c..79618e4ad 100644 --- a/app/application/subscription/status.py +++ b/app/application/subscription/status.py @@ -18,6 +18,7 @@ class SubscriptionExecutionStatus: batch_id: Optional[str] = None task_id: Optional[str] = None current_site_id: Optional[int] = None + next_run_at: Optional[str] = None error: Optional[str] = None can_cancel: bool = False @@ -79,6 +80,8 @@ class SubscriptionExecutionStatusService: "running", "matching", "searching", + "scheduled", + "waiting_subscription", "waiting_site_budget", "preparing", "submitting", @@ -153,8 +156,8 @@ class SubscriptionExecutionStatusService: state = phase = "cancelling" elif task.state == "running": state = phase = task.phase or "running" - elif task.state == "queued" and task.phase == "waiting_site_budget": - state = phase = "waiting_site_budget" + elif task.state == "queued" and task.phase in cls._ACTIVE_STATES: + state = phase = task.phase else: state = phase = task.state return SubscriptionExecutionStatus( @@ -164,6 +167,7 @@ class SubscriptionExecutionStatusService: batch_id=task.batch_id, task_id=task.task_id, current_site_id=task.current_site_id, + next_run_at=task.available_at if task.state == "queued" else None, updated_at=task.updated_at, error=cls._safe_error(task.last_error), can_cancel=state in cls._ACTIVE_STATES, diff --git a/app/chain/subscribe/contract.py b/app/chain/subscribe/contract.py index b23509d87..301f4203e 100644 --- a/app/chain/subscribe/contract.py +++ b/app/chain/subscribe/contract.py @@ -50,6 +50,7 @@ if TYPE_CHECKING: _SubscribeChain__is_full_season_best_version_resource: Callable[..., Any] _SubscribeChain__notify_subscribe_create_failure: Callable[..., Any] _SubscribeChain__post_subscribe_added: Callable[..., Any] + _SubscribeChain__queue_new_subscription_search: Callable[..., Any] _SubscribeChain__prepare_best_version_tv_candidate: Callable[..., Any] _SubscribeChain__prepare_subscribe_progress_fields: Callable[..., Any] _SubscribeChain__prepare_total_episode_change_fields: Callable[..., Any] @@ -71,7 +72,7 @@ if TYPE_CHECKING: _report_search_progress: Callable[..., Any] _search_music_subscribe: Callable[..., Any] _subscription_query: Callable[..., Any] - _defer_recent_subscription: Callable[..., Any] + _recent_subscription_retry_at: Callable[..., Any] _validate_music_subscribe_target: Callable[..., Any] _wait_before_scheduled_search: Callable[..., Any] add: Callable[..., Any] diff --git a/app/chain/subscribe/match.py b/app/chain/subscribe/match.py index a86a0d676..e7c8898cd 100644 --- a/app/chain/subscribe/match.py +++ b/app/chain/subscribe/match.py @@ -85,14 +85,11 @@ def _release_match_admission( subscription_id: int, summary: MatchExecutionSummary, ) -> None: - """释放 Match 订阅准入，并让 token 冲突即时可见。""" + """结束当前订阅资源检查，并记录未能恢复的状态。""" if admission.release(lease): return summary.release_failures += 1 - logger.error( - "订阅准入释放失败: operation=match " - f"subscription_id={subscription_id} run_id={summary.run_id}" - ) + logger.error(f"订阅 {subscription_id} 的搜索状态没有正常恢复，系统稍后会继续检查") def _report_match_finished( @@ -102,13 +99,13 @@ def _report_match_finished( """按实际计数发布 Match 最终进度。""" if not progress_callback: return - final_text = "订阅资源匹配完成" + final_text = "订阅资源检查完成" if summary.finished < summary.total: - final_text = "订阅资源匹配已停止，部分订阅未执行" + final_text = "订阅资源检查已停止，部分订阅这次未检查" elif summary.failed: - final_text = "订阅资源匹配完成，部分订阅失败" + final_text = "订阅资源检查结束，部分订阅没有完成" elif summary.skipped: - final_text = "订阅资源匹配完成，部分订阅跳过" + final_text = "订阅资源检查完成，部分订阅这次未检查" _report_match_progress(progress_callback, summary, value=100, text=final_text) @@ -234,7 +231,7 @@ def _prepare_subscription_match( try: meta = build_subscribe_meta(subscribe) except ValueError: - logger.error(f"订阅 {subscribe.name} 类型错误：{subscribe.type}") + logger.error(f"订阅《{subscribe.name}》的媒体类型不受支持，暂时无法检查资源") return None domains = owner.site_repository.get_domains_by_ids(subscribe.sites) if subscribe.sites else [] sub_sites = owner.get_sub_sites(subscribe) @@ -337,12 +334,12 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): 该入口保持订阅刷新、定时任务和插件调用的稳定签名，具体匹配流程由内部阶段执行。 """ if not torrents: - logger.warn("没有缓存资源，无法匹配订阅") + logger.warn("当前没有可检查的订阅资源") if progress_callback: - progress_callback(value=100, text="没有缓存资源，跳过订阅匹配") + progress_callback(value=100, text="当前没有可检查的订阅资源") return if progress_callback: - progress_callback(value=0, text="正在预处理订阅资源 ...") + progress_callback(value=0, text="正在整理订阅资源 ...") return self._run_match( torrents=torrents, progress_callback=progress_callback, @@ -360,13 +357,13 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): :param progress_callback: 订阅匹配进度更新回调 """ if not torrents: - logger.warn("没有缓存资源，无法匹配订阅") + logger.warn("当前没有可检查的订阅资源") if progress_callback: - progress_callback(value=100, text="没有缓存资源，跳过订阅匹配") + progress_callback(value=100, text="当前没有可检查的订阅资源") return if progress_callback: - progress_callback(value=0, text="正在预处理订阅资源 ...") + progress_callback(value=0, text="正在整理订阅资源 ...") lock_acquired = False summary = MatchExecutionSummary.from_candidates(torrents) @@ -390,7 +387,7 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): progress_callback, summary, value=20, - text=f"资源预处理完成，开始匹配 {total_num} 个订阅 ...", + text=f"资源整理完成，开始检查 {total_num} 个订阅 ...", ) try: for index, listed_subscribe in enumerate(subscribes, start=1): @@ -401,7 +398,7 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): progress_callback, summary, value=20 + ((index - 1) / total_num * 80 if total_num else 80), - text=(f"正在匹配订阅（{index}/{total_num}）{listed_subscribe.name} ..."), + text=(f"正在检查订阅（{index}/{total_num}）{listed_subscribe.name} ..."), current=listed_subscribe.id, ) outcome: Optional[str] = "skipped" @@ -413,7 +410,7 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): ) if lease is None: reason = "admission_conflict" - logger.debug(f"订阅 {listed_subscribe.name} 正在由其他通道处理，本轮匹配已跳过") + logger.debug(f"订阅 {listed_subscribe.name} 正在处理中，本次不再重复检查资源") else: current_subscribe = None execution_context = SubscriptionExecutionContext( @@ -424,14 +421,12 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): try: current_subscribe = self.subscription_repository.get(listed_subscribe.id) if current_subscribe is None: - logger.debug(f"订阅 {listed_subscribe.id} 已不存在，本轮匹配跳过") + logger.debug(f"订阅 {listed_subscribe.id} 已删除，本次不再检查资源") elif current_subscribe.state not in {"R", "P"}: - logger.debug( - f"订阅 {current_subscribe.name} 当前状态为 {current_subscribe.state}，本轮匹配跳过" - ) + logger.debug(f"订阅 {current_subscribe.name} 当前不需要检查资源，本次跳过") elif execution_context.should_stop(): reason = _match_stop_reason(execution_context) - logger.debug(f"订阅 {current_subscribe.name} 已取消或超时，本轮匹配跳过") + logger.debug(f"订阅 {current_subscribe.name} 的资源检查已停止") else: outcome = self._match_subscription( subscribe=current_subscribe, @@ -444,14 +439,14 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): except SubscriptionSearchCancelled as err: outcome = "skipped" reason = _match_stop_reason(execution_context) or "cancelled" - logger.debug(f"订阅 {listed_subscribe.name} 匹配已取消：{str(err)}") + logger.debug(f"订阅 {listed_subscribe.name} 的资源检查已停止：{str(err)}") except Exception as err: outcome = "failed" reason = "error" subscribe_name = ( current_subscribe.name if current_subscribe is not None else listed_subscribe.name ) - logger.error(f"订阅 {subscribe_name} 匹配失败：{str(err)}", exc_info=True) + logger.error(f"订阅 {subscribe_name} 检查资源时出错：{str(err)}", exc_info=True) finally: _release_match_admission( self._subscription_execution_admission, @@ -464,7 +459,7 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): progress_callback, summary, value=20 + (summary.finished / total_num * 80 if total_num else 80), - text=(f"订阅匹配（{index}/{total_num}）处理完成"), + text=(f"已检查订阅（{index}/{total_num}）"), ) finally: processed_torrents.clear() @@ -477,7 +472,7 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): logger.info(summary.finish_log()) if lock_acquired: self._match_lock.release() - logger.debug(f"match Lock released at {datetime.now()}") + logger.debug(f"订阅资源检查已结束：{datetime.now()}") def _match_subscription( self, diff --git a/app/chain/subscribe/notify.py b/app/chain/subscribe/notify.py index ef85d3912..d971e2b45 100644 --- a/app/chain/subscribe/notify.py +++ b/app/chain/subscribe/notify.py @@ -18,6 +18,7 @@ from app.domain.context import ( MediaInfo, ) from app.domain.meta.metabase import MetaBase +from app.runtime.execution import run_in_threadpool from app.runtime.log import logger from app.schemas.common import JsonData from app.schemas.message import Message as _SchemaMessage @@ -142,6 +143,13 @@ class SubscribeNotificationOwner(_SubscribeOwnerBase): "mediainfo": context.mediainfo.to_dict(), }, ) + try: + self._SubscribeChain__queue_new_subscription_search(subscribe_id) + except Exception as error: + logger.warning( + "订阅已保存，但自动搜索暂时没有安排成功，" + f"系统会在下一次检查时重试：{error}" + ) try: report_delivered = _subscription_share_snapshot().report_added( self._SubscribeChain__subscribe_report_payload(context) @@ -177,6 +185,16 @@ class SubscribeNotificationOwner(_SubscribeOwnerBase): "mediainfo": context.mediainfo.to_dict(), }, ) + try: + await run_in_threadpool( + self._SubscribeChain__queue_new_subscription_search, + subscribe_id, + ) + except Exception as error: + logger.warning( + "订阅已保存，但自动搜索暂时没有安排成功，" + f"系统会在下一次检查时重试：{error}" + ) try: report_delivered = await _subscription_share_snapshot().async_report_added( self._SubscribeChain__subscribe_report_payload(context) diff --git a/app/chain/subscribe/search.py b/app/chain/subscribe/search.py index 5c7cd55ce..6b5cb4563 100644 --- a/app/chain/subscribe/search.py +++ b/app/chain/subscribe/search.py @@ -53,6 +53,10 @@ from app.schemas.types import ( SystemConfigKey, ) +_NEW_SUBSCRIPTION_EDIT_SECONDS = 60 +_FOREGROUND_RETRY_SECONDS = 5 +_BACKGROUND_RETRY_SECONDS = 10 + def _ensure_execution_active( execution_context: Optional[SubscriptionExecutionContext], @@ -61,9 +65,9 @@ def _ensure_execution_active( if execution_context is None: return if execution_context.is_cancel_requested(): - raise SubscriptionSearchCancelled("订阅搜索已取消") + raise SubscriptionSearchCancelled("搜索已停止") if execution_context.is_expired(): - raise TimeoutError("订阅执行已超过协作截止时间") + raise TimeoutError("这次搜索用时过长，已停止") def _update_search_task_phase( @@ -91,7 +95,8 @@ def _search_source_and_priority( ) -> tuple[str, int]: """把兼容入口归一为持久来源和公平队列优先级。""" if manual: - return "manual", 100 + target_count = 1 if sid else len(sids or ()) + return "manual", 120 if target_count == 1 else 100 if sid or sids is not None: return "targeted", 80 if state in {"R", "P"}: @@ -99,6 +104,13 @@ def _search_source_and_priority( return "new", 50 +def _retry_at_after(seconds: int) -> str: + """返回指定秒数后的 UTC 时间，供短暂等待任务重新入队。""" + return (datetime.now(timezone.utc) + timedelta(seconds=max(1, seconds))).isoformat( + timespec="seconds" + ) + + def _search_task_available_at( source: str, subscription_ids: tuple[int, ...], @@ -179,16 +191,16 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): progress_callback: Optional[Callable[..., None]], ) -> bool: """获取 Search 或 Match 通道锁，保持各通道内部串行。""" + operation_label = {"search": "搜索", "match": "资源检查"}[operation] lock = self._match_lock if operation == "match" else self._search_queue_lock if lock.acquire(blocking=True, timeout=self._SUBSCRIPTION_EXECUTION_TTL): - logger.debug(f"{operation} lock acquired at {datetime.now()}") + logger.debug(f"订阅{operation_label}已开始：{datetime.now()}") return True - operation_label = {"search": "搜索", "match": "匹配"}[operation] progress_text = { - "search": "订阅搜索锁等待超时，已跳过本轮", - "match": "订阅匹配锁等待超时，已跳过本轮", + "search": "订阅搜索正在处理中，本次不再重复开始", + "match": "订阅资源检查正在进行，本次不再重复开始", }[operation] - logger.error(f"订阅{operation_label}锁等待超时，已中止本轮执行") + logger.error(f"订阅{operation_label}等待时间过长，本次不再重复开始") if progress_callback: progress_callback( value=100, @@ -273,6 +285,12 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): try: subscribes = self._load_search_subscriptions(sid=sid, sids=sids, state=state) total = len(subscribes) + source, _priority = _search_source_and_priority( + sid=sid, + sids=sids, + state=state, + manual=manual, + ) summary = SearchExecutionSummary(source="inline", requested=total) logger.info(summary.start_log()) if progress_callback: @@ -287,7 +305,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): summary.stopped = True break self._report_search_progress(progress_callback, subscribe, index, total) - if self._defer_recent_subscription(subscribe): + if self._recent_subscription_retry_at(subscribe, source): summary.record("skipped", "recent_subscription") continue self._wait_before_scheduled_search(sid, sids, state, progress_callback) @@ -297,7 +315,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): ttl_seconds=self._SUBSCRIPTION_EXECUTION_TTL, ) if lease is None: - logger.debug(f"订阅 {subscribe.name} 正在由其他通道处理，本轮搜索已跳过") + logger.debug(f"订阅《{subscribe.name}》正在处理，本次搜索先不重复执行") summary.record("skipped", "admission_conflict") continue execution_context = SubscriptionExecutionContext( @@ -312,7 +330,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): current = self.subscription_repository.get(subscribe.id) if current is None or current.state == "S": if current and current.state == "S": - logger.debug(f"订阅 {current.name} 已暂停，本轮搜索已跳过") + logger.debug(f"订阅《{current.name}》已暂停，本次没有搜索") continue processed_result = self._process_search_subscription( current, @@ -326,12 +344,12 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): except SubscriptionSearchCancelled: reason = "ttl_timeout" if execution_context.is_expired() else "cancelled" outcome = "failed" if reason == "ttl_timeout" else "cancelled" - logger.debug(f"订阅 {subscribe.name} 搜索已在安全边界取消") + logger.debug(f"订阅《{subscribe.name}》搜索已停止") except SubscriptionSearchDeferred as deferred: outcome = "skipped" reason = "site_budget_deferred" logger.debug( - f"订阅 {subscribe.name} 站点预算冲突，兼容搜索将在 {deferred.retry_at} 后重试" + f"订阅《{subscribe.name}》遇到站点繁忙，将在稍后自动重试：{deferred.retry_at}" ) except Exception as err: outcome = "failed" @@ -347,7 +365,8 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): ) except Exception as err: logger.error( - f"订阅 {subscribe.name} 搜索后状态重置失败：{str(err)}", + f"订阅《{subscribe.name}》搜索结束后没有恢复到正常状态，" + f"系统稍后会继续处理：{str(err)}", exc_info=True, ) finally: @@ -355,8 +374,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): if not released: summary.release_failures += 1 logger.error( - "订阅准入释放失败: operation=search " - f"subscription_id={subscribe.id} run_id={summary.run_id}" + f"订阅《{subscribe.name}》的搜索状态没有正常恢复，系统稍后会继续处理" ) summary.record(outcome, reason) self._report_search_progress( @@ -375,7 +393,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): self._search_queue_lock.release() if summary is not None: logger.info(summary.finish_log()) - logger.debug(f"search Lock released at {datetime.now()}") + logger.debug(f"订阅搜索已结束：{datetime.now()}") def _execute_queued_search( self, @@ -463,7 +481,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): ) -> set[int]: """有界消费可恢复任务；单任务失败不得阻止后续订阅。""" if not self._search_queue_lock.acquire(blocking=False): - logger.debug("订阅搜索队列已有消费者，本轮仅保留持久任务") + logger.debug("订阅搜索已经在后台进行，本次安排会接着处理") summary.consumer_conflicts += 1 return set() owner = f"subscribe-search:{uuid4().hex}" @@ -511,7 +529,7 @@ class _SubscribeSearchQueueOwner(_SubscribeSearchQueueCoordinator): ) -> Optional[int]: """执行一条已认领任务，返回实际进入搜索处理的订阅 ID。""" if task.lease_token is None: - logger.error(f"订阅搜索任务 {task.task_id} 缺少租约令牌，跳过执行") + logger.error("这次订阅搜索暂时无法开始，系统稍后会重新处理") summary.record("failed", "missing_lease") return None task_id = str(task.task_id) @@ -528,14 +546,21 @@ class _SubscribeSearchQueueOwner(_SubscribeSearchQueueCoordinator): task_id=task_id, lease_token=lease_token, state="cancelled", - error="订阅已不存在", + error="订阅已删除", ) summary.record("cancelled", "missing_subscription") return None self._report_search_progress(progress_callback, subscribe, index, limit) - if self._defer_recent_subscription(subscribe): - _skip_search_task(queue, task, "订阅仍在新增保护期，本轮搜索已跳过") - summary.record("skipped", "recent_subscription") + recent_retry_at = self._recent_subscription_retry_at(subscribe, task.source) + if recent_retry_at: + queue.defer_task( + task_id=task_id, + lease_token=lease_token, + available_at=recent_retry_at, + phase="scheduled", + message="订阅刚刚创建，保存好设置后会自动开始搜索", + ) + summary.record("requeued", "recent_subscription") return None lease = self._subscription_execution_admission.try_acquire( subscription_id=subscribe.id, @@ -543,9 +568,28 @@ class _SubscribeSearchQueueOwner(_SubscribeSearchQueueCoordinator): ttl_seconds=self._SUBSCRIPTION_EXECUTION_TTL, ) if lease is None: - logger.debug(f"订阅 {subscribe.name} 正在由其他通道处理，本轮搜索已跳过") - _skip_search_task(queue, task, "同一订阅正在由其他通道处理，本轮搜索已跳过") - summary.record("skipped", "admission_conflict") + if task.source in {"manual", "targeted", "new"}: + retry_seconds = ( + _FOREGROUND_RETRY_SECONDS + if task.source in {"manual", "targeted"} + else _BACKGROUND_RETRY_SECONDS + ) + queue.defer_task( + task_id=task_id, + lease_token=lease_token, + available_at=_retry_at_after(retry_seconds), + phase="waiting_subscription", + message="这个订阅正在处理，结束后会自动继续搜索", + ) + summary.record("requeued", "admission_conflict") + logger.debug(f"订阅《{subscribe.name}》正在处理，搜索会在结束后自动继续") + else: + _skip_search_task( + queue, + task, + "这个订阅正在处理，本次自动检查无需重复执行", + ) + summary.record("skipped", "admission_conflict") return None phase_changed = partial(_update_search_task_phase, queue, task_id, lease_token) execution_context = SubscriptionExecutionContext( @@ -563,12 +607,12 @@ class _SubscribeSearchQueueOwner(_SubscribeSearchQueueCoordinator): task_id=task_id, lease_token=lease_token, state="cancelled", - error="订阅已不存在", + error="订阅已删除", ) summary.record("cancelled", "missing_subscription") return None if current.state == "S": - _skip_search_task(queue, task, "订阅已暂停，本轮搜索已跳过") + _skip_search_task(queue, task, "订阅已暂停，这次没有搜索") summary.record("skipped", "paused") return None searchchain.configure_subscription_site_budget( @@ -605,7 +649,7 @@ class _SubscribeSearchQueueOwner(_SubscribeSearchQueueCoordinator): task_id=task_id, lease_token=lease_token, state="failed", - error="订阅执行已超过协作截止时间", + error="这次搜索用时过长，已停止，可稍后重试", ) summary.record("failed", "ttl_timeout") else: @@ -664,7 +708,11 @@ class _SubscribeSearchQueueOwner(_SubscribeSearchQueueCoordinator): try: searchchain.configure_subscription_site_budget(None) except Exception as err: - logger.error(f"订阅 {subscribe.name} 搜索站点预算清理失败：{str(err)}", exc_info=True) + logger.error( + f"订阅《{subscribe.name}》结束站点访问时遇到问题，" + f"系统稍后会继续处理：{str(err)}", + exc_info=True, + ) try: if current and current.state == "N": self._SubscribeChain__apply_subscribe_update( @@ -673,15 +721,12 @@ class _SubscribeSearchQueueOwner(_SubscribeSearchQueueCoordinator): scene="search_reset", ) except Exception as err: - logger.error(f"订阅 {subscribe.name} 搜索后状态重置失败：{str(err)}", exc_info=True) + logger.error(f"订阅《{subscribe.name}》搜索结束后未能恢复正常状态：{str(err)}", exc_info=True) finally: released = self._subscription_execution_admission.release(lease) if not released: summary.release_failures += 1 - logger.error( - "订阅准入释放失败: operation=search " - f"subscription_id={subscribe.id} run_id={summary.run_id}" - ) + logger.error(f"订阅《{subscribe.name}》的搜索状态没有正常恢复，系统稍后会继续处理") self._report_search_progress( progress_callback, subscribe, @@ -694,27 +739,50 @@ class _SubscribeSearchQueueOwner(_SubscribeSearchQueueCoordinator): self, progress_callback: Optional[Callable[..., None]] = None, limit: int = 50, + manual_sids: Optional[tuple[int, ...]] = None, ) -> None: - """短周期恢复排队或租约过期任务，不创建新的 24 小时兜底批次。""" + """短周期恢复等待任务，并可优先反馈本次手工搜索结果。""" queue: Optional[SubscriptionSearchRepository] = getattr( self, "subscription_search_repository", None ) if queue is None: return - summary = SearchExecutionSummary(source="resume", requested=max(1, limit)) - logger.info(summary.start_log()) + manual_ids = tuple(dict.fromkeys(manual_sids or ())) + subscribes = ( + self._load_search_subscriptions(sid=None, sids=manual_ids, state=None) + if manual_ids + else [] + ) + summary = SearchExecutionSummary( + source="manual" if manual_ids else "resume", + requested=len(manual_ids) if manual_ids else max(1, limit), + ) + if manual_ids: + logger.info(summary.start_log()) try: - self._drain_search_queue( + processed = self._drain_search_queue( queue=queue, - limit=max(1, limit), + limit=max(1, limit, len(manual_ids)), progress_callback=progress_callback, summary=summary, ) + if manual_ids: + processed_subscribes = [item for item in subscribes if item.id in processed] + self._notify_manual_search( + True, + manual_ids[0] if len(manual_ids) == 1 else None, + None if len(manual_ids) == 1 else manual_ids, + subscribes, + processed_subscribes, + ) except Exception: summary.round_failed = True raise finally: - logger.info(summary.finish_log()) + if not manual_ids and summary.processed: + summary.requested = summary.processed + if manual_ids or summary.processed or summary.round_failed: + logger.info(summary.finish_log()) def cancel_search_batch(self, batch_id: str) -> bool: """请求取消持久搜索批次；未注入队列时返回失败。""" @@ -750,15 +818,52 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): return cast(list[SubscriptionSnapshot], repository.list(self.get_states_for_search(state or "N"))) @staticmethod - def _defer_recent_subscription(subscribe: SubscriptionSnapshot) -> bool: - """新增一分钟内保留 N 状态，为用户编辑筛选条件留出窗口。""" - if not subscribe.date: - return False - subscribe_time = datetime.strptime(subscribe.date, "%Y-%m-%d %H:%M:%S") - if (datetime.now() - subscribe_time).total_seconds() >= 60: - return False - logger.debug(f"订阅标题：{subscribe.name} 新增小于1分钟，暂不搜索...") - return True + def _recent_subscription_retry_at( + subscribe: SubscriptionSnapshot, + source: str, + ) -> Optional[str]: + """新订阅保留一分钟编辑时间，并返回自动开始搜索的时间。""" + if source != "new" or not subscribe.date: + return None + try: + subscribe_time = datetime.strptime(subscribe.date, "%Y-%m-%d %H:%M:%S") + except ValueError: + logger.warning(f"订阅《{subscribe.name}》的添加时间无法识别，将直接开始搜索") + return None + remaining = _NEW_SUBSCRIPTION_EDIT_SECONDS - ( + datetime.now() - subscribe_time + ).total_seconds() + if remaining <= 0: + return None + retry_seconds = max(1, int(remaining) + 1) + logger.debug(f"新订阅《{subscribe.name}》将在约 {retry_seconds} 秒后自动开始搜索") + return _retry_at_after(retry_seconds) + + def _SubscribeChain__queue_new_subscription_search( + self, + subscribe_id: int, + ) -> Optional[str]: + """在订阅保存后安排自动搜索，定时扫描仍作为恢复保障。""" + queue: Optional[SubscriptionSearchRepository] = getattr( + self, "subscription_search_repository", None + ) + if queue is None: + return None + subscribe = self.subscription_repository.get(subscribe_id) + if subscribe is None or subscribe.state != "N": + return None + available_at = self._recent_subscription_retry_at(subscribe, "new") + enqueued = queue.enqueue( + subscription_ids=(subscribe_id,), + source="new", + priority=50, + available_at_by_subscription={ + subscribe_id: available_at or datetime.now(timezone.utc).isoformat(timespec="seconds") + }, + ) + if enqueued.created_count: + logger.info(f"已安排新订阅《{subscribe.name}》自动搜索，保存好设置后会自动开始") + return enqueued.active_batch_ids[0] if enqueued.active_batch_ids else None @staticmethod def _wait_before_scheduled_search( @@ -771,9 +876,9 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): if sid or sids is not None or state not in {"R", "P"}: return sleep_time = random.randint(60, 300) - logger.debug(f"订阅搜索随机休眠 {sleep_time} 秒 ...") + logger.debug(f"为了避免连续访问站点，约 {sleep_time} 秒后继续搜索") if progress_callback: - progress_callback(text=f"订阅搜索随机休眠 {sleep_time} 秒后继续 ...") + progress_callback(text=f"为了避免连续访问站点，约 {sleep_time} 秒后继续搜索 ...") time.sleep(sleep_time) def _process_search_subscription( @@ -791,7 +896,7 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): try: meta = build_subscribe_meta(subscribe) except ValueError: - logger.error(f"订阅 {subscribe.name} 类型错误：{subscribe.type}") + logger.error(f"订阅《{subscribe.name}》的媒体类型不受支持，暂时无法搜索") return subscribe mediainfo: MediaInfo = MediaChain().recognize_media( meta=meta, @@ -965,7 +1070,7 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): subscribes: list[SubscriptionSnapshot], processed: list[SubscriptionSnapshot], ) -> None: - """为手动触发的搜索发布保持旧文案的系统消息。""" + """用清晰文案反馈手动搜索的完成或等待状态。""" if not manual: return if not subscribes: @@ -974,7 +1079,7 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): message = ( f"{subscribes[0].name} 搜索完成！" if processed - else f"{subscribes[0].name} 本轮未执行或未完成，将等待下一次正常调度。" + else f"{subscribes[0].name} 已安排搜索，系统会自动继续处理。" ) self.messagehelper.put(message, title="订阅搜索", role="system") elif sids is not None: @@ -984,6 +1089,6 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): message = ( "所有订阅搜索完成！" if len(processed) == len(subscribes) - else "订阅搜索完成，部分订阅本轮未执行或未完成。" + else "订阅搜索已安排，暂未完成的项目会自动继续处理。" ) self.messagehelper.put(message, title="订阅搜索", role="system") diff --git a/app/db/adapters/subscription.py b/app/db/adapters/subscription.py index 3576641b2..6db8d42e1 100644 --- a/app/db/adapters/subscription.py +++ b/app/db/adapters/subscription.py @@ -769,9 +769,18 @@ class SessionSubscriptionRepository: candidates.append(candidate) return candidates - async def list_search_ids(self, username: str, state: str) -> builtins.list[int]: - """异步读取用户指定状态下的订阅主键。""" - return [snapshot.id for snapshot in await self.async_list_by_username(username, state)] + async def list_search_ids( + self, + username: Optional[str], + state: str, + ) -> builtins.list[int]: + """异步读取用户或管理员全局范围内指定状态的订阅主键。""" + snapshots = ( + await self.async_list_by_username(username, state) + if username is not None + else await self.async_list(state) + ) + return [snapshot.id for snapshot in snapshots] async def stage_delete(self, subscribe_id: int) -> None: """异步暂存删除订阅。""" diff --git a/app/db/adapters/subscriptionsearch.py b/app/db/adapters/subscriptionsearch.py index 0017ed492..d15cbcaaf 100644 --- a/app/db/adapters/subscriptionsearch.py +++ b/app/db/adapters/subscriptionsearch.py @@ -1,6 +1,7 @@ """订阅搜索持久队列的 SQLAlchemy 适配器。""" from collections.abc import Callable, Mapping +from datetime import datetime, timedelta, timezone from typing import Optional, TypeVar from sqlalchemy.orm import Session @@ -19,6 +20,7 @@ from app.db.oper.subscriptionsearch import SubscriptionSearchOper from app.db.uow import SqlAlchemyUnitOfWork T = TypeVar("T") +_BUSY_SITE_RETRY_SECONDS = 10 def _batch(record: SubscriptionSearchBatch) -> SearchBatchSnapshot: @@ -101,7 +103,7 @@ class TransactionalSubscriptionSearchRepository: """创建批次并返回 single-flight 合并计数。""" def operation(repository: SubscriptionSearchOper) -> SearchEnqueueResult: """在同一事务内创建批次和任务。""" - record, created, coalesced = repository.enqueue( + record, created, coalesced, active_batch_ids = repository.enqueue( subscription_ids=subscription_ids, source=source, priority=priority, @@ -111,6 +113,7 @@ class TransactionalSubscriptionSearchRepository: batch=_batch(record), created_count=created, coalesced_count=coalesced, + active_batch_ids=active_batch_ids, ) return self._write(operation) @@ -183,13 +186,17 @@ class TransactionalSubscriptionSearchRepository: task_id: str, lease_token: str, available_at: str, + phase: str = "waiting_site_budget", + message: Optional[str] = None, ) -> bool: - """以站点预算的下一次可用时间重新排队任务。""" + """按指定时间和可见原因重新排队任务。""" return self._write( lambda repository: repository.defer_task( task_id=task_id, lease_token=lease_token, available_at=available_at, + phase=phase, + message=message, ) ) @@ -225,14 +232,32 @@ class TransactionalSubscriptionSearchRepository: lease_seconds=lease_seconds, ) retry_at = record.next_allowed_at - if not acquired and record.lease_token and record.lease_expires_at: - retry_at = max(retry_at, record.lease_expires_at) + wait_reason = None + now = datetime.now(timezone.utc) + cooldown_active = bool( + record.last_outcome not in {None, "success", "skipped"} + and record.next_allowed_at > now.isoformat(timespec="seconds") + ) + lease_busy = bool( + record.lease_token + and record.lease_expires_at + and record.lease_expires_at > now.isoformat(timespec="seconds") + ) + if not acquired and cooldown_active: + wait_reason = "cooldown" + elif not acquired and lease_busy: + wait_reason = "busy" + short_retry = (now + timedelta(seconds=_BUSY_SITE_RETRY_SECONDS)).isoformat( + timespec="seconds" + ) + retry_at = min(record.lease_expires_at, short_retry) return SiteBudgetClaim( site_id=record.site_id, acquired=acquired, retry_at=retry_at, consecutive_failures=record.consecutive_failures, lease_token=record.lease_token if acquired else None, + wait_reason=wait_reason, ) return self._write(operation) diff --git a/app/db/adapters/subscriptionstatus.py b/app/db/adapters/subscriptionstatus.py index cbfe8279b..bb339a441 100644 --- a/app/db/adapters/subscriptionstatus.py +++ b/app/db/adapters/subscriptionstatus.py @@ -83,6 +83,7 @@ class SessionSubscriptionExecutionStatusRepository: """返回最近更新的批次，访问范围由应用服务依据任务校验。""" result = await self._session.execute( select(SubscriptionSearchBatch) + .where(SubscriptionSearchBatch.total_count > 0) .order_by( SubscriptionSearchBatch.updated_at.desc(), SubscriptionSearchBatch.id.desc(), diff --git a/app/db/oper/subscribe.py b/app/db/oper/subscribe.py index 17a8117f8..25c20ec92 100644 --- a/app/db/oper/subscribe.py +++ b/app/db/oper/subscribe.py @@ -556,9 +556,13 @@ class SubscribeOper(DbOper): ) return candidates - async def list_search_ids(self, username: str, state: str) -> List[int]: - """返回用户指定状态的订阅编号，不向应用用例暴露 ORM 列表。""" - subscribes = await self.async_list_by_username(username, state=state) + async def list_search_ids(self, username: Optional[str], state: str) -> List[int]: + """返回用户或管理员全局范围内指定状态的订阅编号。""" + subscribes = ( + await self.async_list_by_username(username, state=state) + if username is not None + else await self.async_list(state=state) + ) return [subscribe.id for subscribe in subscribes if subscribe.id] def get_by( diff --git a/app/db/oper/subscriptionsearch.py b/app/db/oper/subscriptionsearch.py index d35045859..726004a40 100644 --- a/app/db/oper/subscriptionsearch.py +++ b/app/db/oper/subscriptionsearch.py @@ -31,7 +31,7 @@ class SubscriptionSearchOper(DbOper): source: str, priority: int, available_at_by_subscription: Optional[Mapping[int, str]], - ) -> tuple[SubscriptionSearchBatch, int, int]: + ) -> tuple[SubscriptionSearchBatch, int, int, tuple[str, ...]]: """创建批次，并以活动键合并同一订阅的重叠搜索入口。""" if not isinstance(self._db, Session): raise RuntimeError("订阅搜索入队需要调用方提供同步 Session") @@ -49,6 +49,7 @@ class SubscriptionSearchOper(DbOper): self._db.flush() created = 0 coalesced = 0 + active_batch_ids: list[str] = [] for position, subscription_id in enumerate(dict.fromkeys(subscription_ids)): active_key = f"subscription:{subscription_id}" available_at = ( @@ -56,6 +57,11 @@ class SubscriptionSearchOper(DbOper): if available_at_by_subscription else now ) + initial_phase = ( + "scheduled" + if source == "new" and available_at > now + else "queued" + ) task = SubscriptionSearchTask( task_id=uuid4().hex, batch_id=batch.batch_id, @@ -65,7 +71,7 @@ class SubscriptionSearchOper(DbOper): priority=priority, position=position, state="queued", - phase="queued", + phase=initial_phase, available_at=available_at, created_at=now, updated_at=now, @@ -77,15 +83,31 @@ class SubscriptionSearchOper(DbOper): created += 1 except IntegrityError: coalesced += 1 + promote_queued_task = and_( + SubscriptionSearchTask.priority < priority, + SubscriptionSearchTask.state == "queued", + ) execute_dml( self._db, update(SubscriptionSearchTask) .where(SubscriptionSearchTask.active_key == active_key) .values( + source=case( + (SubscriptionSearchTask.priority < priority, source), + else_=SubscriptionSearchTask.source, + ), priority=case( (SubscriptionSearchTask.priority < priority, priority), else_=SubscriptionSearchTask.priority, ), + phase=case( + (promote_queued_task, "queued"), + else_=SubscriptionSearchTask.phase, + ), + last_error=case( + (promote_queued_task, None), + else_=SubscriptionSearchTask.last_error, + ), available_at=case( ( or_( @@ -100,11 +122,20 @@ class SubscriptionSearchOper(DbOper): ), execution_options={"synchronize_session": False}, ) + active_task = self._db.execute( + select(SubscriptionSearchTask).where( + SubscriptionSearchTask.active_key == active_key + ) + ).scalars().first() + if active_task is not None: + active_batch_ids.append(active_task.batch_id) batch.total_count = created if created == 0: batch.state = "completed" batch.finished_at = now - return batch, created, coalesced + else: + active_batch_ids.insert(0, batch.batch_id) + return batch, created, coalesced, tuple(dict.fromkeys(active_batch_ids)) def claim_next(self, *, owner: str, lease_seconds: int) -> Optional[SubscriptionSearchTask]: """使用 CAS 认领最高优先级任务，过期 running 任务可被恢复。""" @@ -139,6 +170,7 @@ class SubscriptionSearchOper(DbOper): ) .order_by( case( + (SubscriptionSearchTask.priority >= 100, 2), (SubscriptionSearchTask.created_at <= fairness_before, 1), else_=0, ).desc(), @@ -369,8 +401,10 @@ class SubscriptionSearchOper(DbOper): task_id: str, lease_token: str, available_at: str, + phase: str, + message: Optional[str], ) -> bool: - """释放当前租约并在站点预算时间到达后恢复同一任务。""" + """释放当前租约并在指定时间后按可见原因恢复同一任务。""" if not isinstance(self._db, Session): raise RuntimeError("订阅搜索延后需要调用方提供同步 Session") task = self._db.execute( @@ -400,7 +434,7 @@ class SubscriptionSearchOper(DbOper): ) .values( state="queued", - phase="waiting_site_budget", + phase=phase, current_site_id=None, lease_owner=None, lease_token=None, @@ -408,7 +442,7 @@ class SubscriptionSearchOper(DbOper): available_at=available_at, updated_at=now, finished_at=None, - last_error=None, + last_error=message, ), execution_options={"synchronize_session": False}, ) diff --git a/app/locales/en-US.json b/app/locales/en-US.json index b8c435c13..5268d15de 100644 --- a/app/locales/en-US.json +++ b/app/locales/en-US.json @@ -137,6 +137,17 @@ "当前管理用户缺少可审计身份": "The current administrator does not have an auditable identity", "人工复核任务不存在": "The manual review task does not exist", "任务添加失败": "Failed to add task", + "没有需要搜索的订阅": "No subscriptions need searching", + "这个订阅已经在搜索中，请稍候": "This subscription is already being searched. Please wait", + "已安排搜索，很快开始": "Search scheduled and will start soon", + "当前没有可检查的订阅资源": "No subscription resources are currently available to check", + "正在整理订阅资源 ...": "Organizing subscription resources ...", + "订阅资源检查完成": "Subscription resource check complete", + "订阅资源检查已停止，部分订阅这次未检查": "Subscription resource check stopped; some subscriptions were not checked this time", + "订阅资源检查结束，部分订阅没有完成": "Subscription resource check finished; some subscriptions did not complete", + "订阅资源检查完成，部分订阅这次未检查": "Subscription resource check complete; some subscriptions were not checked this time", + "订阅搜索正在处理中，本次不再重复开始": "A subscription search is already in progress. This run will not start again", + "订阅资源检查正在进行，本次不再重复开始": "A subscription resource check is already in progress. This run will not start again", "无法识别媒体信息": "Unable to recognize media information", "未识别到媒体信息": "Unable to recognize media information", "未识别到音乐信息": "Unable to recognize music information", @@ -382,8 +393,8 @@ "微信 ClawBot 通知未启用或配置尚未保存，请先保存并启用当前渠道": "WeChat ClawBot notification is not enabled or the configuration has not been saved. Please save and enable this channel first", "请输入至少一个有效的站点 ID": "Enter at least one valid site ID", "所有订阅搜索完成": "All subscription searches are complete", - "订阅搜索批次不存在": "The subscription search batch does not exist", - "订阅搜索批次已结束或无法取消": "The subscription search batch has ended or cannot be cancelled", + "没有找到这次搜索，请刷新后重试": "This search could not be found. Refresh and try again", + "这次搜索已经结束，暂时无法停止": "This search has already ended and cannot be stopped", "订阅搜索锁等待超时，已跳过本轮": "Subscription search lock timed out, this round was skipped", "订阅匹配锁等待超时，已跳过本轮": "Subscription matching lock timed out, this round was skipped", "请输入订阅 ID，多个 ID 用空格分隔，或输入 all": "Enter subscription IDs separated by spaces, or enter all", @@ -765,6 +776,18 @@ "source": "已完成 {count} 个订阅搜索", "target": "Completed searches for {count} subscriptions" }, + { + "source": "{count} 个订阅已经在搜索中，无需重复提交", + "target": "{count} subscriptions are already being searched. No need to submit again" + }, + { + "source": "已安排 {queued} 个订阅搜索，另有 {ongoing} 个正在处理中", + "target": "Scheduled searches for {queued} subscriptions; {ongoing} are already being processed" + }, + { + "source": "已安排 {count} 个订阅搜索，系统会依次处理", + "target": "Scheduled searches for {count} subscriptions. They will be processed in order" + }, { "source": "请输入至少一个有效的订阅 ID", "target": "Enter at least one valid subscription ID" @@ -1421,6 +1444,10 @@ "source": "订阅搜索（{index}/{total}）处理完成", "target": "Subscription search ({index}/{total}) completed" }, + { + "source": "为了避免连续访问站点，约 {seconds} 秒后继续搜索 ...", + "target": "To avoid repeated site requests, search will continue in about {seconds} seconds ..." + }, { "source": "开始刷新订阅，共 {count} 个订阅 ...", "target": "Starting subscription refresh, {count} subscriptions ..." @@ -1433,6 +1460,18 @@ "source": "正在匹配订阅（{index}/{total}）{name} ...", "target": "Matching subscription ({index}/{total}) {name} ..." }, + { + "source": "资源整理完成，开始检查 {count} 个订阅 ...", + "target": "Resource organization complete. Checking {count} subscriptions ..." + }, + { + "source": "正在检查订阅（{index}/{total}）{name} ...", + "target": "Checking subscription ({index}/{total}) {name} ..." + }, + { + "source": "已检查订阅（{index}/{total}）", + "target": "Checked subscription ({index}/{total})" + }, { "source": "开始更新订阅元数据，共 {count} 个订阅 ...", "target": "Starting subscription metadata update, {count} subscriptions ..." diff --git a/app/locales/zh-TW.json b/app/locales/zh-TW.json index db8be27a8..6865fa371 100644 --- a/app/locales/zh-TW.json +++ b/app/locales/zh-TW.json @@ -129,6 +129,17 @@ "当前管理用户缺少可审计身份": "目前管理使用者缺少可稽核身分", "人工复核任务不存在": "人工複核任務不存在", "任务添加失败": "任務新增失敗", + "没有需要搜索的订阅": "沒有需要搜尋的訂閱", + "这个订阅已经在搜索中，请稍候": "這個訂閱已在搜尋中，請稍候", + "已安排搜索，很快开始": "已安排搜尋，很快開始", + "当前没有可检查的订阅资源": "目前沒有可檢查的訂閱資源", + "正在整理订阅资源 ...": "正在整理訂閱資源 ...", + "订阅资源检查完成": "訂閱資源檢查完成", + "订阅资源检查已停止，部分订阅这次未检查": "訂閱資源檢查已停止，部分訂閱這次未檢查", + "订阅资源检查结束，部分订阅没有完成": "訂閱資源檢查結束，部分訂閱沒有完成", + "订阅资源检查完成，部分订阅这次未检查": "訂閱資源檢查完成，部分訂閱這次未檢查", + "订阅搜索正在处理中，本次不再重复开始": "訂閱搜尋正在處理中，本次不再重複開始", + "订阅资源检查正在进行，本次不再重复开始": "訂閱資源檢查正在進行，本次不再重複開始", "无法识别媒体信息": "無法識別媒體資訊", "未识别到媒体信息": "未識別到媒體資訊", "未识别到音乐信息": "未識別到音樂資訊", @@ -374,8 +385,8 @@ "微信 ClawBot 通知未启用或配置尚未保存，请先保存并启用当前渠道": "微信 ClawBot 通知未啟用或設定尚未儲存，請先儲存並啟用目前渠道", "请输入至少一个有效的站点 ID": "請輸入至少一個有效的站點 ID", "所有订阅搜索完成": "所有訂閱搜尋完成", - "订阅搜索批次不存在": "訂閱搜尋批次不存在", - "订阅搜索批次已结束或无法取消": "訂閱搜尋批次已結束或無法取消", + "没有找到这次搜索，请刷新后重试": "找不到這次搜尋，請重新整理後再試", + "这次搜索已经结束，暂时无法停止": "這次搜尋已經結束，暫時無法停止", "请输入订阅 ID，多个 ID 用空格分隔，或输入 all": "請輸入訂閱 ID，多個 ID 以空格分隔，或輸入 all", "请输入至少一个有效的订阅 ID": "請輸入至少一個有效的訂閱 ID", "格式错误，请输入：cookie [2fa_code/secret]": "格式錯誤，請輸入：cookie [2fa_code/secret]", @@ -751,6 +762,18 @@ "source": "已完成 {count} 个订阅搜索", "target": "已完成 {count} 個訂閱搜尋" }, + { + "source": "{count} 个订阅已经在搜索中，无需重复提交", + "target": "{count} 個訂閱已在搜尋中，無需重複提交" + }, + { + "source": "已安排 {queued} 个订阅搜索，另有 {ongoing} 个正在处理中", + "target": "已安排 {queued} 個訂閱搜尋，另有 {ongoing} 個正在處理中" + }, + { + "source": "已安排 {count} 个订阅搜索，系统会依次处理", + "target": "已安排 {count} 個訂閱搜尋，系統會依序處理" + }, { "source": "请输入至少一个有效的订阅 ID", "target": "請輸入至少一個有效的訂閱 ID" @@ -1407,6 +1430,10 @@ "source": "订阅搜索（{index}/{total}）处理完成", "target": "訂閱搜尋（{index}/{total}）處理完成" }, + { + "source": "为了避免连续访问站点，约 {seconds} 秒后继续搜索 ...", + "target": "為了避免連續存取站點，約 {seconds} 秒後繼續搜尋 ..." + }, { "source": "开始刷新订阅，共 {count} 个订阅 ...", "target": "開始重新整理訂閱，共 {count} 個訂閱 ..." @@ -1419,6 +1446,18 @@ "source": "正在匹配订阅（{index}/{total}）{name} ...", "target": "正在匹配訂閱（{index}/{total}）{name} ..." }, + { + "source": "资源整理完成，开始检查 {count} 个订阅 ...", + "target": "資源整理完成，開始檢查 {count} 個訂閱 ..." + }, + { + "source": "正在检查订阅（{index}/{total}）{name} ...", + "target": "正在檢查訂閱（{index}/{total}）{name} ..." + }, + { + "source": "已检查订阅（{index}/{total}）", + "target": "已檢查訂閱（{index}/{total}）" + }, { "source": "开始更新订阅元数据，共 {count} 个订阅 ...", "target": "開始更新訂閱元資料，共 {count} 個訂閱 ..." diff --git a/app/scheduler/catalog.py b/app/scheduler/catalog.py index d68fa607b..866e7c681 100644 --- a/app/scheduler/catalog.py +++ b/app/scheduler/catalog.py @@ -154,8 +154,8 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase): "interval", id="subscribe_search_queue", name="恢复订阅搜索队列", - minutes=1, - next_run_time=datetime.now(pytz.timezone(config.timezone)) + timedelta(seconds=10), + seconds=10, + next_run_time=datetime.now(pytz.timezone(config.timezone)) + timedelta(seconds=5), kwargs={"job_id": "subscribe_search_queue"}, ) diff --git a/app/schemas/exports.py b/app/schemas/exports.py index b2300ec44..c4a01e9f1 100644 --- a/app/schemas/exports.py +++ b/app/schemas/exports.py @@ -463,6 +463,7 @@ SCHEMA_EXPORTS = { 'SubscriptionBatchStatus': ('app.schemas.subscribe', 'SubscriptionBatchStatus'), 'SubscriptionExecutionStatus': ('app.schemas.subscribe', 'SubscriptionExecutionStatus'), 'SubscriptionMessage': ('app.schemas.message', 'SubscriptionMessage'), + 'SubscriptionSearchSubmission': ('app.schemas.subscribe', 'SubscriptionSearchSubmission'), 'SubtitleDownloadData': ('app.schemas.download', 'SubtitleDownloadData'), 'SubtitleInfo': ('app.schemas.search', 'SubtitleInfo'), 'SystemEnvironmentUpdateData': ('app.schemas.system', 'SystemEnvironmentUpdateData'), diff --git a/app/schemas/subscribe.py b/app/schemas/subscribe.py index 333af40a6..935c78132 100644 --- a/app/schemas/subscribe.py +++ b/app/schemas/subscribe.py @@ -55,6 +55,7 @@ class SubscriptionExecutionStatus(BaseModel): # type: ignore[misc] batch_id: Optional[str] = None task_id: Optional[str] = None current_site_id: Optional[int] = None + next_run_at: Optional[str] = None error: Optional[str] = None can_cancel: bool = False @@ -84,6 +85,17 @@ class SubscriptionBatchStatus(BaseModel): # type: ignore[misc] model_config = ConfigDict(from_attributes=True) +class SubscriptionSearchSubmission(BaseModel): # type: ignore[misc] + """手工订阅搜索已安排后的轻量跟踪信息。""" + + batch_id: Optional[str] = None + batch_ids: List[str] = Field(default_factory=list) + target_count: int = 0 + queued_count: int = 0 + ongoing_count: int = 0 + single: bool = False + + class Subscribe(OptionalMediaIdentityMixin, BaseModel): """订阅输入与响应模型，媒体身份必须为空对或完整有效对。""" diff --git a/scripts/validation/subscription_governance_scale.py b/scripts/validation/subscription_governance_scale.py index 99679d8f7..b8d0843cc 100644 --- a/scripts/validation/subscription_governance_scale.py +++ b/scripts/validation/subscription_governance_scale.py @@ -799,7 +799,7 @@ def _run_match_execution_case(case: ScaleCase) -> dict[str, Any]: @staticmethod def info(message: Any, *_args: Any, **_kwargs: Any) -> None: - """保存结构化批次摘要。""" + """保存开始和结束摘要。""" info_logs.append(str(message)) class _ScaleMediaChain: @@ -958,8 +958,10 @@ def _run_match_execution_case(case: ScaleCase) -> dict[str, Any]: "info_log_count": len(info_logs), "info_log_bounded": bool( len(info_logs) == 2 - and info_logs[0].startswith("订阅治理轮次开始: operation=match ") - and info_logs[1].startswith("订阅治理轮次结束: operation=match ") + and info_logs[0].startswith("开始检查订阅资源，共 ") + and info_logs[1].startswith("订阅资源检查结束：") + and "operation=" not in "\n".join(info_logs) + and "run_id=" not in "\n".join(info_logs) ), "first_settlement_local_ms": round(settlement_ms[0], 3), "subscription_settlement_local_p50_ms": _percentile(settlement_ms, 50), diff --git a/tests/test_api_background_task_registry.py b/tests/test_api_background_task_registry.py index 3affb8e0e..40e7ad7cc 100644 --- a/tests/test_api_background_task_registry.py +++ b/tests/test_api_background_task_registry.py @@ -72,6 +72,34 @@ class _RunningTaskRegistry(TaskRegistry): ) +class _CompletedSearchTaskRegistry(_TaskRegistry): + """返回已完成结果，验证手工搜索先保存安排再启动后台处理。""" + + def create_sync(self, function, *args, owner: str, **kwargs) -> asyncio.Future: + """记录同步任务，并为测试提供可等待的安排结果。""" + super().create_sync(function, *args, owner=owner, **kwargs) + future = asyncio.get_running_loop().create_future() + result = None + if owner == "api.subscribe.search.enqueue": + result = SimpleNamespace( + active_batch_ids=("batch-1", "batch-2"), + created_count=2, + coalesced_count=0, + ) + future.set_result(result) + return future + + +class _SubscriptionSearchTargets: + """为手工搜索命令提供当前用户可访问的订阅编号。""" + + async def list_search_ids(self, username, state) -> list[int]: + """返回稳定目标，并校验超级用户读取全部运行中订阅。""" + assert username is None + assert state == "R" + return [11, 12] + + class _ProtocolManager: """提供兼容协议流结束时需要的最小 AgentManager 接口。""" @@ -206,10 +234,14 @@ def test_seerr_subscribe_uses_task_registry(monkeypatch) -> None: def test_manual_subscription_search_uses_task_registry() -> None: """手工订阅搜索命令应以稳定 owner 提交顺序搜索批次。""" - registry = _TaskRegistry() - repository = object() + registry = _CompletedSearchTaskRegistry() + repository = _SubscriptionSearchTargets() + search_repository = SimpleNamespace(enqueue=lambda **_kwargs: None) runtime = SimpleNamespace( - subscription=SimpleNamespace(repository=lambda _db: repository) + subscription=SimpleNamespace( + repository=lambda _db: repository, + search_repository=search_repository, + ) ) command = subscription_dependencies.get_search_subscriptions_command( task_registry=registry, @@ -221,12 +253,24 @@ def test_manual_subscription_search_uses_task_registry() -> None: command.execute(SubscribeSearchActor(username="admin", is_superuser=True)) ) - function, args, kwargs, owner = registry.calls[0] - assert found is True - assert function is subscription_dependencies._start_subscription_search_batch - assert args == (None, "R") - assert kwargs == {} - assert owner == "api.subscribe.search" + enqueue_function, enqueue_args, enqueue_kwargs, enqueue_owner = registry.calls[0] + run_function, run_args, run_kwargs, run_owner = registry.calls[1] + assert found is not None + assert found.batch_ids == ("batch-1", "batch-2") + assert found.queued_count == 2 + assert found.ongoing_count == 0 + assert enqueue_function is search_repository.enqueue + assert enqueue_args == () + assert enqueue_kwargs == { + "subscription_ids": (11, 12), + "source": "manual", + "priority": 100, + } + assert enqueue_owner == "api.subscribe.search.enqueue" + assert run_function is subscription_dependencies._resume_submitted_subscription_search + assert run_args == ((11, 12),) + assert run_kwargs == {} + assert run_owner == "api.subscribe.search.run" def test_history_ai_redo_uses_task_registry() -> None: diff --git a/tests/test_subscribe_endpoint.py b/tests/test_subscribe_endpoint.py index 17c7a1a17..984dbd8db 100644 --- a/tests/test_subscribe_endpoint.py +++ b/tests/test_subscribe_endpoint.py @@ -653,8 +653,19 @@ class TestSubscribeEndpoint: 普通用户批量搜索把用户身份交给应用命令。 """ from app.api.endpoints.subscribe import search_subscribes + from app.application.subscription.search import SubscriptionSearchSubmission - command = SimpleNamespace(execute=AsyncMock(return_value=True)) + command = SimpleNamespace( + execute=AsyncMock( + return_value=SubscriptionSearchSubmission( + batch_ids=("batch-1",), + target_count=2, + queued_count=1, + ongoing_count=1, + single=False, + ) + ) + ) response = asyncio.run( search_subscribes( command=command, @@ -663,6 +674,8 @@ class TestSubscribeEndpoint: ) assert response.success + assert response.message == "已安排 1 个订阅搜索，另有 1 个正在处理中" + assert response.data.queued_count == 1 command.execute.assert_awaited_once() actor = command.execute.await_args.args[0] assert actor.username == "alice" diff --git a/tests/test_subscribe_search_command.py b/tests/test_subscribe_search_command.py index 5fa18093b..96f610571 100644 --- a/tests/test_subscribe_search_command.py +++ b/tests/test_subscribe_search_command.py @@ -1,3 +1,4 @@ +import asyncio from types import SimpleNamespace from unittest.mock import Mock @@ -8,6 +9,7 @@ from app.application.subscription.delete import SubscribeDeletionCandidate from app.application.subscription.search import ( SearchSubscriptionsCommand, SubscribeSearchActor, + SubscriptionSearchSubmission, ) from app.runtime.tasks import TaskRegistry @@ -19,15 +21,15 @@ class _Repository: """保存预设单条候选和批量编号。""" self.candidate = candidate self.subscribe_ids = subscribe_ids or [] + self.list_calls = [] async def get_candidate(self, _subscribe_id): """返回预设订阅候选。""" return self.candidate async def list_search_ids(self, username, state): - """校验普通用户搜索状态并返回预设编号。""" - assert username == "alice" - assert state == "R" + """记录访问范围并返回预设编号。""" + self.list_calls.append((username, state)) return self.subscribe_ids @@ -40,85 +42,117 @@ def _candidate(username): ) +def _submitter(calls): + """构造记录目标并返回稳定安排结果的异步提交器。""" + async def submit(subscribe_ids, single): + """记录一次轻量入队请求。""" + calls.append((subscribe_ids, single)) + return SubscriptionSearchSubmission( + batch_ids=("batch-1",), + target_count=len(subscribe_ids), + queued_count=len(subscribe_ids), + ongoing_count=0, + single=single, + ) + + return submit + + @pytest.mark.asyncio -async def test_superuser_search_all_uses_single_global_scheduler_request(): - """管理员搜索全部订阅时保持一次 state=R 的全局调度语义。""" - scheduled = [] +async def test_superuser_search_all_queues_every_active_subscription(): + """管理员搜索全部订阅时读取全局活动订阅并一次提交。""" + submitted = [] + repository = _Repository(subscribe_ids=[2, 5]) command = SearchSubscriptionsCommand( - repository=_Repository(), - schedule_search=lambda sid, state: scheduled.append((sid, state)), + repository=repository, + submit_search=_submitter(submitted), ) - assert await command.execute( + result = await command.execute( SubscribeSearchActor(username="admin", is_superuser=True) - ) is True - assert scheduled == [(None, "R")] + ) + + assert result is not None + assert result.target_count == 2 + assert repository.list_calls == [(None, "R")] + assert submitted == [((2, 5), False)] @pytest.mark.asyncio async def test_regular_user_search_all_schedules_only_owned_subscriptions(): """普通用户搜索全部时把归属订阅合并为一次后台批次。""" - scheduled = [] + submitted = [] + repository = _Repository(subscribe_ids=[2, 5]) command = SearchSubscriptionsCommand( - repository=_Repository(subscribe_ids=[2, 5]), - schedule_search=lambda sid, state: scheduled.append((sid, state)), + repository=repository, + submit_search=_submitter(submitted), ) - assert await command.execute( + result = await command.execute( SubscribeSearchActor(username="alice", is_superuser=False) - ) is True - assert scheduled == [((2, 5), None)] + ) + + assert result is not None + assert repository.list_calls == [("alice", "R")] + assert submitted == [((2, 5), False)] @pytest.mark.asyncio async def test_regular_user_search_all_with_no_targets_does_not_schedule(): """普通用户没有可搜索订阅时不创建空后台任务。""" - scheduled = [] + submitted = [] command = SearchSubscriptionsCommand( repository=_Repository(), - schedule_search=lambda ids, state: scheduled.append((ids, state)), + submit_search=_submitter(submitted), ) - assert await command.execute( + result = await command.execute( SubscribeSearchActor(username="alice", is_superuser=False) - ) is True - assert scheduled == [] + ) + + assert result is not None + assert result.target_count == 0 + assert result.batch_ids == () + assert submitted == [] @pytest.mark.asyncio async def test_targeted_search_rejects_missing_or_other_users_subscription(): """单条搜索不得泄漏订阅是否属于其他普通用户。""" - scheduled = [] + submitted = [] command = SearchSubscriptionsCommand( repository=_Repository(candidate=_candidate("bob")), - schedule_search=lambda sid, state: scheduled.append((sid, state)), + submit_search=_submitter(submitted), ) assert await command.execute( SubscribeSearchActor(username="alice", is_superuser=False), subscribe_id=7, - ) is False - assert scheduled == [] + ) is None + assert submitted == [] @pytest.mark.asyncio async def test_targeted_search_schedules_accessible_subscription(): """归属用户搜索单条订阅时提交历史兼容参数。""" - scheduled = [] + submitted = [] command = SearchSubscriptionsCommand( repository=_Repository(candidate=_candidate("alice")), - schedule_search=lambda sid, state: scheduled.append((sid, state)), + submit_search=_submitter(submitted), ) - assert await command.execute( + result = await command.execute( SubscribeSearchActor(username="alice", is_superuser=False), subscribe_id=7, - ) is True - assert scheduled == [((7,), None)] + ) + + assert result is not None + assert result.single is True + assert submitted == [((7,), True)] -def test_subscription_search_batch_uses_one_scheduler_generation(monkeypatch): - """一个后台批次只占用一次调度任务运行权。""" +def test_submitted_search_wakes_short_cycle_queue(monkeypatch): + """手工搜索入队后立即唤醒短周期恢复任务。""" calls = [] monkeypatch.setattr( subscription_dependencies, @@ -126,24 +160,46 @@ def test_subscription_search_batch_uses_one_scheduler_generation(monkeypatch): lambda job_id, **kwargs: calls.append((job_id, kwargs)), ) - subscription_dependencies._start_subscription_search_batch((2, 5), None) + subscription_dependencies._resume_submitted_subscription_search((2, 5)) assert calls == [ ( - "subscribe_search", - {"sids": (2, 5), "state": None, "manual": True}, + "subscribe_search_queue", + {"limit": 2, "manual_sids": (2, 5)}, ), ] @pytest.mark.asyncio -async def test_search_dependency_registers_one_owned_background_batch(): - """请求适配器只登记一个具名后台批次。""" +async def test_search_dependency_persists_before_waking_background_worker(): + """请求适配器先等待持久入队完成，再登记后台恢复任务。""" registry = TaskRegistry() - registry.create_sync = Mock() + calls = [] + + def create_sync(function, *args, owner, **kwargs): + """同步执行测试函数，并返回已经完成的 Future。""" + calls.append((function, args, owner, kwargs)) + future = asyncio.get_running_loop().create_future() + result = function(*args, **kwargs) if owner.endswith("enqueue") else None + future.set_result(result) + return future + + registry.create_sync = Mock(side_effect=create_sync) repository = _Repository(subscribe_ids=[2, 5]) + search_repository = SimpleNamespace( + enqueue=Mock( + return_value=SimpleNamespace( + active_batch_ids=("batch-new", "batch-existing"), + created_count=1, + coalesced_count=1, + ) + ) + ) runtime = SimpleNamespace( - subscription=SimpleNamespace(repository=lambda _db: repository), + subscription=SimpleNamespace( + repository=lambda _db: repository, + search_repository=search_repository, + ), ) command = subscription_dependencies.get_search_subscriptions_command( task_registry=registry, @@ -151,12 +207,23 @@ async def test_search_dependency_registers_one_owned_background_batch(): runtime=runtime, ) - assert await command.execute( + result = 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", + ) + + assert result is not None + assert result.batch_ids == ("batch-new", "batch-existing") + assert result.queued_count == 1 + assert result.ongoing_count == 1 + assert calls[0] == ( + search_repository.enqueue, + (), + "api.subscribe.search.enqueue", + {"subscription_ids": (2, 5), "source": "manual", "priority": 100}, + ) + assert calls[1] == ( + subscription_dependencies._resume_submitted_subscription_search, + ((2, 5),), + "api.subscribe.search.run", + {}, ) diff --git a/tests/test_subscribe_search_state.py b/tests/test_subscribe_search_state.py index 3e80f531c..dbdb27fee 100644 --- a/tests/test_subscribe_search_state.py +++ b/tests/test_subscribe_search_state.py @@ -175,7 +175,7 @@ def test_subscribe_search_aborts_when_lock_times_out(monkeypatch) -> None: subscribe_oper.assert_not_called() progress.assert_called_once_with( value=100, - text="订阅搜索锁等待超时，已跳过本轮", + text="订阅搜索正在处理中，本次不再重复开始", ) @@ -265,7 +265,7 @@ def test_inline_search_conflict_does_not_report_false_completion(monkeypatch) -> assert progress.call_args.kwargs == { "value": 100, - "text": "订阅搜索结束，部分订阅本轮未执行或未完成", + "text": "搜索结束，部分订阅这次没有完成", "data": {"total": 1, "finished": 0}, } assert chain._subscription_execution_admission.release(match_lease) is True @@ -279,4 +279,4 @@ def test_subscribe_match_aborts_when_lock_times_out(monkeypatch) -> None: chain = object.__new__(SubscribeChain) chain.match({"example.org": []}, progress_callback=progress) - progress.assert_any_call(value=100, text="订阅匹配锁等待超时，已跳过本轮") + progress.assert_any_call(value=100, text="订阅资源检查正在进行，本次不再重复开始") diff --git a/tests/test_subscription_execution_status.py b/tests/test_subscription_execution_status.py index 54a0a474e..06aee2788 100644 --- a/tests/test_subscription_execution_status.py +++ b/tests/test_subscription_execution_status.py @@ -14,13 +14,15 @@ def _task( phase: str = "searching", updated_at: str = "2026-09-01T01:00:00+00:00", batch_id: str = "batch-1", + source: str = "manual", + available_at: str | None = None, ) -> SearchTaskSnapshot: """构造最小搜索任务快照。""" return SearchTaskSnapshot( task_id=f"task-{subscription_id}", batch_id=batch_id, subscription_id=subscription_id, - source="manual", + source=source, priority=100, position=subscription_id, state=state, @@ -30,8 +32,15 @@ def _task( lease_token="lease" if state == "running" else None, created_at="2026-09-01T00:00:00+00:00", updated_at=updated_at, + available_at=available_at, current_site_id=9 if phase == "waiting_site_budget" else None, - last_error=" provider\n timeout " if state == "failed" else None, + last_error=( + " provider\n timeout " + if state == "failed" + else "站点暂时忙，系统会自动继续搜索" + if state == "queued" and phase == "waiting_site_budget" + else None + ), ) @@ -85,18 +94,44 @@ def test_execution_status_exposes_site_wait_and_cancel_capability(): def test_execution_status_exposes_queued_site_wait_without_error(): - """重新入队的站点预算冲突应显示等待状态而不是失败。""" + """重新入队的站点繁忙应显示等待状态、说明和继续时间。""" repository = _Repository() - repository.tasks[2] = _task(2, state="queued", phase="waiting_site_budget") + retry_at = "2026-09-01T01:00:10+00:00" + repository.tasks[2] = _task( + 2, + state="queued", + phase="waiting_site_budget", + available_at=retry_at, + ) statuses = asyncio.run(SubscriptionExecutionStatusService(repository).for_subscriptions((2,))) assert statuses[2].state == "waiting_site_budget" assert statuses[2].phase == "waiting_site_budget" - assert statuses[2].error is None + assert statuses[2].error == "站点暂时忙，系统会自动继续搜索" + assert statuses[2].next_run_at == retry_at assert statuses[2].can_cancel is True +def test_execution_status_exposes_scheduled_new_search_without_failure(): + """新订阅编辑等待期应显示为已安排，而不是跳过或失败。""" + repository = _Repository() + retry_at = "2026-09-01T01:01:00+00:00" + repository.tasks[4] = _task( + 4, + state="queued", + phase="scheduled", + source="new", + available_at=retry_at, + ) + + statuses = asyncio.run(SubscriptionExecutionStatusService(repository).for_subscriptions((4,))) + + assert statuses[4].state == "scheduled" + assert statuses[4].next_run_at == retry_at + assert statuses[4].error is None + + def test_failed_search_exposes_safe_error(): """搜索失败文本必须压平且不暴露内部错误细节。""" repository = _Repository() diff --git a/tests/test_subscription_match_admission.py b/tests/test_subscription_match_admission.py index 2b935c50c..094ae7173 100644 --- a/tests/test_subscription_match_admission.py +++ b/tests/test_subscription_match_admission.py @@ -169,7 +169,7 @@ def test_match_skips_subscription_paused_after_admission() -> None: "skipped": 1, "failed": 0, } - assert progress.call_args.kwargs["text"] == "订阅资源匹配完成，部分订阅跳过" + assert progress.call_args.kwargs["text"] == "订阅资源检查完成，部分订阅这次未检查" _assert_channel_and_subscription_released(chain, (listed.id,)) @@ -200,7 +200,7 @@ def test_match_progress_reports_completed_skipped_and_failed_counts() -> None: "skipped": 1, "failed": 1, } - assert progress.call_args.kwargs["text"] == "订阅资源匹配完成，部分订阅失败" + assert progress.call_args.kwargs["text"] == "订阅资源检查结束，部分订阅没有完成" assert chain._subscription_execution_admission.release(skipped_lease) is True _assert_channel_and_subscription_released(chain, (completed.id, failed.id)) @@ -220,7 +220,7 @@ def test_match_stop_does_not_report_unvisited_subscriptions_as_completed(monkeyp assert progress.call_args.kwargs == { "value": 100, - "text": "订阅资源匹配已停止，部分订阅未执行", + "text": "订阅资源检查已停止，部分订阅这次未检查", "data": { "total": 2, "finished": 0, @@ -257,17 +257,12 @@ def test_match_logs_one_bounded_start_and_finish_summary(monkeypatch) -> None: chain.match({"site-a.example": [object()], "site-b.example": [object(), object()]}) assert len(info_logs) == 2 - assert info_logs[0].startswith("订阅治理轮次开始: operation=match ") - assert "subscriptions=3" in info_logs[0] - assert "sites=2" in info_logs[0] - assert "candidates=3" in info_logs[0] - assert info_logs[1].startswith("订阅治理轮次结束: operation=match ") - assert "state=skipped" in info_logs[1] - assert "task_completed=1" in info_logs[1] - assert "task_skipped=2" in info_logs[1] - assert "admission_conflicts=1" in info_logs[1] - assert "cancelled=1" in info_logs[1] - assert "ttl_timeouts=0" in info_logs[1] + assert info_logs[0] == "开始检查订阅资源，共 3 个订阅、3 个资源，来自 2 个站点。" + assert info_logs[1].startswith("订阅资源检查结束：部分订阅这次未检查。") + assert "本次检查 3/3 个订阅，完成 1 个，这次未检查 2 个，失败 0 个" in info_logs[1] + assert "其中 1 个订阅正在处理中，本次没有重复检查" in info_logs[1] + assert "另有 1 个订阅已停止" in info_logs[1] + assert "run_id" not in info_logs[1] assert "订阅成功" not in "\n".join(info_logs) assert chain._subscription_execution_admission.release(search_lease) is True @@ -293,9 +288,8 @@ def test_match_summary_distinguishes_ttl_timeout(monkeypatch) -> None: chain.match({"site.example": [object()]}) assert len(info_logs) == 2 - assert "task_skipped=1" in info_logs[1] - assert "cancelled=0" in info_logs[1] - assert "ttl_timeouts=1" in info_logs[1] + assert "这次未检查 1 个" in info_logs[1] + assert "另有 1 个订阅检查时间过长，已停止" in info_logs[1] def test_match_logs_subscription_admission_release_failure(monkeypatch) -> None: @@ -323,5 +317,5 @@ def test_match_logs_subscription_admission_release_failure(monkeypatch) -> None: chain.match({"site.example": [object()]}) - assert any("订阅准入释放失败: operation=match subscription_id=17" in item for item in error_logs) - assert "release_failures=1" in info_logs[-1] + assert any("订阅 17 的搜索状态没有正常恢复，系统稍后会继续检查" in item for item in error_logs) + assert "另有 1 个订阅的搜索状态没有正常恢复" in info_logs[-1] diff --git a/tests/test_subscription_search_governance.py b/tests/test_subscription_search_governance.py index e96847556..55124df18 100644 --- a/tests/test_subscription_search_governance.py +++ b/tests/test_subscription_search_governance.py @@ -82,6 +82,7 @@ def _chain(tmp_path, subscribes: list[SubscriptionSnapshot]): chain._match_lock = _ForbiddenLock() chain._search_queue_lock = threading.Lock() chain._subscription_execution_admission = SubscriptionExecutionAdmission() + chain.messagehelper = Mock() return chain @@ -179,11 +180,9 @@ def test_successful_sites_remain_available_to_next_due_subscription(tmp_path, mo assert calls == [(40, 11), (40, 12), (41, 11), (41, 12)] assert chain.get_search_batch(batch_id).state == "completed" - assert "sites=2" in info_logs[-1] - assert "site_requests=4" in info_logs[-1] - assert "site_failures=0" in info_logs[-1] - assert "site_cooldown_skips=0" in info_logs[-1] - assert "candidates=4" in info_logs[-1] + assert "访问 2 个站点" in info_logs[-1] + assert "发出 4 次请求" in info_logs[-1] + assert "找到 4 个资源" in info_logs[-1] def test_fallback_queue_executes_without_match_global_lock(tmp_path, monkeypatch): @@ -280,12 +279,12 @@ def test_site_budget_conflict_requeues_task_without_batch_failure(tmp_path, monk assert task.state == "queued" assert task.phase == "waiting_site_budget" assert task.available_at == retry_at - assert task.last_error is None + assert task.last_error == "站点暂时忙，系统会自动继续搜索" assert chain.subscription_search_repository.claim_next(owner="worker-after-retry") is None -def test_search_logs_one_bounded_start_and_finish_summary(tmp_path, monkeypatch): - """Search INFO 只保留轮次摘要，并携带任务终态与耗时字段。""" +def test_search_logs_one_readable_start_and_finish_summary(tmp_path, monkeypatch): + """Search INFO 只保留用户能直接读懂的开始和结束摘要。""" subscribes = [_subscribe(50), _subscribe(51)] chain = _chain(tmp_path, subscribes) _make_tasks_ready(monkeypatch) @@ -305,19 +304,14 @@ def test_search_logs_one_bounded_start_and_finish_summary(tmp_path, monkeypatch) assert batch_id assert len(info_logs) == 2 - assert info_logs[0].startswith("订阅治理轮次开始: operation=search ") - assert f"batch_id={batch_id}" in info_logs[0] - assert "subscriptions=2" in info_logs[0] - assert info_logs[1].startswith("订阅治理轮次结束: operation=search ") - assert "state=failed" in info_logs[1] - assert "processed=2" in info_logs[1] - assert "task_completed=1" in info_logs[1] - assert "task_failed=1" in info_logs[1] - assert "admission_conflicts=0" in info_logs[1] - assert "ttl_timeouts=0" in info_logs[1] - assert "site_requests=0" in info_logs[1] - assert "duration_ms=" in info_logs[1] - assert "订阅成功" not in "\n".join(info_logs) + assert info_logs[0] == "开始订阅定时检查，共 2 个订阅。" + assert info_logs[1].startswith("订阅定时检查结束：部分订阅没有完成。") + assert "本次处理 2/2 个" in info_logs[1] + assert "完成 1 个" in info_logs[1] + assert "失败 1 个" in info_logs[1] + assert "发出 0 次请求" in info_logs[1] + assert "用时 " in info_logs[1] + assert "operation=" not in "\n".join(info_logs) def test_search_logs_subscription_admission_release_failure(tmp_path, monkeypatch): @@ -343,8 +337,8 @@ def test_search_logs_subscription_admission_release_failure(tmp_path, monkeypatc with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): chain.search(state="R") - assert any("订阅准入释放失败: operation=search subscription_id=52" in item for item in error_logs) - assert "release_failures=1" in info_logs[-1] + assert any("订阅《治理电影 52》的搜索状态没有正常恢复" in item for item in error_logs) + assert "另有 1 个搜索状态没有正常恢复" in info_logs[-1] def test_queued_search_logs_failed_finish_summary_when_callback_raises(tmp_path, monkeypatch): @@ -361,9 +355,7 @@ def test_queued_search_logs_failed_finish_summary_when_callback_raises(tmp_path, ) assert len(info_logs) == 2 - assert info_logs[-1].startswith("订阅治理轮次结束: operation=search ") - assert "state=failed" in info_logs[-1] - assert "round_failed=1" in info_logs[-1] + assert info_logs[-1].startswith("订阅定时检查结束：部分订阅没有完成。") def test_resume_search_logs_failed_finish_summary_when_drain_raises(tmp_path, monkeypatch): @@ -380,11 +372,8 @@ def test_resume_search_logs_failed_finish_summary_when_drain_raises(tmp_path, mo with pytest.raises(RuntimeError, match="claim failed"): chain.resume_search_queue() - assert len(info_logs) == 2 - assert info_logs[-1].startswith("订阅治理轮次结束: operation=search ") - assert "source=resume" in info_logs[-1] - assert "state=failed" in info_logs[-1] - assert "round_failed=1" in info_logs[-1] + assert len(info_logs) == 1 + assert info_logs[-1].startswith("等待中的订阅搜索结束：部分订阅没有完成。") def test_swallowed_indexer_failure_marks_task_and_batch_failed_but_continues(tmp_path, monkeypatch): @@ -457,10 +446,9 @@ def test_swallowed_indexer_failure_marks_task_and_batch_failed_but_continues(tmp assert batch.failed_count == 1 assert "Flaky" in batch.last_error assert "HTTP 429" in batch.last_error - assert "sites=2" in info_logs[-1] - assert "site_requests=2" in info_logs[-1] - assert "site_failures=1" in info_logs[-1] - assert "cooldown_seconds=900.0" in info_logs[-1] + assert "失败 1 个" in info_logs[-1] + assert "访问 2 个站点" in info_logs[-1] + assert "发出 2 次请求" in info_logs[-1] def test_same_subscription_conflict_is_skipped_without_waiting(tmp_path, monkeypatch): @@ -487,12 +475,116 @@ def test_same_subscription_conflict_is_skipped_without_waiting(tmp_path, monkeyp assert batch.state == "skipped" assert batch.finished_count == 0 assert batch.skipped_count == 1 - assert batch.last_error == "同一订阅正在由其他通道处理，本轮搜索已跳过" - assert "task_skipped=1" in info_logs[-1] - assert "admission_conflicts=1" in info_logs[-1] + assert batch.last_error == "这个订阅正在处理，本次自动检查无需重复执行" + assert "这次未搜索 1 个" in info_logs[-1] assert chain._subscription_execution_admission.release(match_lease) is True +def test_manual_search_waits_for_same_subscription_then_continues(tmp_path, monkeypatch): + """手动搜索遇到同一订阅正在处理时应稍后自动继续，而不是结束为跳过。""" + subscribe = _subscribe(61) + chain = _chain(tmp_path, [subscribe]) + _make_tasks_ready(monkeypatch) + process = Mock(return_value=subscribe) + monkeypatch.setattr(chain, "_process_search_subscription", process) + match_lease = chain._subscription_execution_admission.try_acquire( + subscription_id=subscribe.id, + operation="match", + ttl_seconds=60, + ) + assert match_lease is not None + + with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): + batch_id = chain.search(sid=subscribe.id, state=None, manual=True) + + batch = chain.get_search_batch(batch_id) + assert batch.state == "queued" + process.assert_not_called() + with chain.subscription_search_repository._session_factory() as session: + task = session.query(SubscriptionSearchTask).filter_by( + subscription_id=subscribe.id, + ).one() + assert task.phase == "waiting_subscription" + assert task.last_error == "这个订阅正在处理，结束后会自动继续搜索" + task.available_at = "1970-01-01T00:00:00+00:00" + session.commit() + + assert chain._subscription_execution_admission.release(match_lease) is True + with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): + chain.resume_search_queue(limit=1) + + process.assert_called_once() + assert chain.get_search_batch(batch_id).state == "completed" + + +def test_recent_new_subscription_is_scheduled_instead_of_skipped(tmp_path, monkeypatch): + """新订阅的一分钟编辑窗口应保留任务并在到时后自动继续。""" + subscribe = replace( + _subscribe(62), + state="N", + date=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + ) + chain = _chain(tmp_path, [subscribe]) + _make_tasks_ready(monkeypatch) + process = Mock(side_effect=AssertionError("编辑等待期内不应开始搜索")) + monkeypatch.setattr(chain, "_process_search_subscription", process) + + with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): + batch_id = chain.search(state="N") + + batch = chain.get_search_batch(batch_id) + assert batch.state == "queued" + assert batch.skipped_count == 0 + with chain.subscription_search_repository._session_factory() as session: + task = session.query(SubscriptionSearchTask).filter_by( + subscription_id=subscribe.id, + ).one() + assert task.phase == "scheduled" + assert task.last_error == "订阅刚刚创建，保存好设置后会自动开始搜索" + assert task.available_at > datetime.now(timezone.utc).isoformat(timespec="seconds") + process.assert_not_called() + + +def test_new_subscription_post_commit_creates_scheduled_search(tmp_path): + """订阅保存后应立即写入自动搜索计划，无需等待五分钟扫描。""" + subscribe = replace( + _subscribe(64), + state="N", + date=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + ) + chain = _chain(tmp_path, [subscribe]) + + batch_id = chain._SubscribeChain__queue_new_subscription_search(subscribe.id) + + assert batch_id + assert chain.get_search_batch(batch_id).state == "queued" + with chain.subscription_search_repository._session_factory() as session: + task = session.query(SubscriptionSearchTask).filter_by( + subscription_id=subscribe.id, + ).one() + assert task.source == "new" + assert task.phase == "scheduled" + assert task.available_at > datetime.now(timezone.utc).isoformat(timespec="seconds") + + +def test_manual_search_bypasses_new_subscription_edit_wait(tmp_path, monkeypatch): + """用户主动点击搜索时不受新订阅编辑等待时间限制。""" + subscribe = replace( + _subscribe(63), + date=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + ) + chain = _chain(tmp_path, [subscribe]) + _make_tasks_ready(monkeypatch) + process = Mock(return_value=subscribe) + monkeypatch.setattr(chain, "_process_search_subscription", process) + + with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): + batch_id = chain.search(sid=subscribe.id, state=None, manual=True) + + process.assert_called_once() + assert chain.get_search_batch(batch_id).state == "completed" + + def test_paused_subscription_is_skipped_after_admission_refresh(tmp_path, monkeypatch): """准入后重新读取到暂停订阅时不得开始搜索或伪造完成。""" subscribe = _subscribe(9) @@ -511,7 +603,7 @@ def test_paused_subscription_is_skipped_after_admission_refresh(tmp_path, monkey assert batch.state == "skipped" assert batch.finished_count == 0 assert batch.skipped_count == 1 - assert batch.last_error == "订阅已暂停，本轮搜索已跳过" + assert batch.last_error == "订阅已暂停，这次没有搜索" def test_cleanup_failures_cannot_leak_subscription_admission(tmp_path, monkeypatch): @@ -699,9 +791,8 @@ def test_ttl_expiry_after_normal_return_marks_task_and_batch_failed(tmp_path, mo assert batch.finished_count == 0 assert batch.failed_count == 1 assert batch.cancelled_count == 0 - assert batch.last_error == "订阅执行已超过协作截止时间" - assert "task_failed=1" in info_logs[-1] - assert "ttl_timeouts=1" in info_logs[-1] + assert batch.last_error == "这次搜索用时过长，已停止，可稍后重试" + assert "失败 1 个" in info_logs[-1] def test_ttl_expiry_after_download_started_completes_with_actual_result(tmp_path, monkeypatch): @@ -758,4 +849,4 @@ def test_site_budget_ttl_expiry_marks_task_and_batch_failed(tmp_path, monkeypatc assert batch.finished_count == 0 assert batch.failed_count == 1 assert batch.cancelled_count == 0 - assert batch.last_error == "订阅执行已超过协作截止时间" + assert batch.last_error == "这次搜索用时过长，已停止，可稍后重试" diff --git a/tests/test_subscription_search_queue.py b/tests/test_subscription_search_queue.py index 62fc76cb5..778c5166b 100644 --- a/tests/test_subscription_search_queue.py +++ b/tests/test_subscription_search_queue.py @@ -40,7 +40,9 @@ def test_search_queue_coalesces_active_subscription_and_raises_priority(tmp_path assert manual.created_count == 0 assert manual.coalesced_count == 1 assert manual.batch.state == "completed" + assert manual.active_batch_ids == (scheduled.batch.batch_id,) assert first.subscription_id == 1 + assert first.source == "manual" assert first.priority == 100 assert second.subscription_id == 2 assert first.task_id != second.task_id @@ -82,6 +84,39 @@ def test_search_queue_claims_each_subscription_only_after_its_available_at(tmp_p assert accelerated.priority == 100 +def test_manual_search_promotes_scheduled_new_subscription(tmp_path): + """用户主动搜索应立即唤醒仍在编辑等待期的新订阅任务。""" + repository, engine = _repository(tmp_path) + later_at = (datetime.now(timezone.utc) + timedelta(minutes=1)).isoformat(timespec="seconds") + automatic = repository.enqueue( + subscription_ids=(22,), + source="new", + priority=50, + available_at_by_subscription={22: later_at}, + ) + + with Session(engine) as session: + scheduled = session.execute( + select(SubscriptionSearchTask).where(SubscriptionSearchTask.subscription_id == 22) + ).scalar_one() + assert scheduled.phase == "scheduled" + + manual = repository.enqueue( + subscription_ids=(22,), + source="manual", + priority=120, + available_at_by_subscription={22: "1970-01-01T00:00:00+00:00"}, + ) + claimed = repository.claim_next(owner="worker-manual") + + assert manual.created_count == 0 + assert manual.active_batch_ids == (automatic.batch.batch_id,) + assert claimed is not None + assert claimed.source == "manual" + assert claimed.priority == 120 + assert claimed.phase == "matching" + + def test_search_queue_recovers_expired_lease_with_same_task_identity(tmp_path): """进程遗留的过期 running 任务应以新 token 恢复且 attempt 单调递增。""" repository, engine = _repository(tmp_path) @@ -269,8 +304,8 @@ def test_search_queue_aggregates_skipped_tasks_without_marking_success(tmp_path) assert batch.last_error == "同一订阅正在由其他通道处理，本轮搜索已跳过" -def test_search_queue_ages_old_fallback_ahead_of_new_manual_work(tmp_path): - """手工任务可优先，但等待超过公平窗口的兜底任务不得持续饥饿。""" +def test_search_queue_keeps_manual_work_ahead_of_aged_fallback(tmp_path): + """用户主动搜索始终先于定时检查，避免点击后长时间没有反馈。""" repository, engine = _repository(tmp_path) repository.enqueue(subscription_ids=(8,), source="fallback", priority=10) aged_at = (datetime.now(timezone.utc) - timedelta(minutes=16)).isoformat(timespec="seconds") @@ -285,5 +320,5 @@ def test_search_queue_ages_old_fallback_ahead_of_new_manual_work(tmp_path): claimed = repository.claim_next(owner="worker-a") - assert claimed.subscription_id == 8 - assert claimed.source == "fallback" + assert claimed.subscription_id == 9 + assert claimed.source == "manual" diff --git a/tests/test_subscription_site_budget.py b/tests/test_subscription_site_budget.py index c53d77392..49d79e731 100644 --- a/tests/test_subscription_site_budget.py +++ b/tests/test_subscription_site_budget.py @@ -35,6 +35,7 @@ def _repository(tmp_path): def test_site_budget_allows_one_inflight_per_site_and_independent_sites(tmp_path): """同站点第二个调用必须等待，不同站点可立即并行。""" repository, _engine = _repository(tmp_path) + before_retry = datetime.now(timezone.utc) first = repository.claim_site(site_id=1, owner="worker-a", lease_seconds=900) same_site = repository.claim_site(site_id=1, owner="worker-b", lease_seconds=900) @@ -43,6 +44,8 @@ def test_site_budget_allows_one_inflight_per_site_and_independent_sites(tmp_path assert first.acquired is True assert same_site.acquired is False assert same_site.lease_token is None + assert same_site.wait_reason == "busy" + assert datetime.fromisoformat(same_site.retry_at) <= before_retry + timedelta(seconds=11) assert other_site.acquired is True