diff --git a/app/agent/llm/helper.py b/app/agent/llm/helper.py index 3fd0d1c18..8d6d0c868 100644 --- a/app/agent/llm/helper.py +++ b/app/agent/llm/helper.py @@ -7,6 +7,7 @@ import time from functools import wraps from typing import TYPE_CHECKING, Any, List, Optional from urllib.parse import urlsplit +from uuid import uuid4 from langchain_core.messages import AIMessage, AIMessageChunk @@ -881,11 +882,17 @@ class LLMHelper: default_headers: dict[str, str] | None, model_kwargs: dict[str, Any], ) -> tuple[dict[str, str] | None, dict[str, Any]]: - """为 OpenAI 与 xAI 官方端点构造稳定提示词缓存路由参数。""" + """为官方端点构造稳定会话及提示词缓存路由参数。""" cache_key = str(prompt_cache_key or "").strip() headers = dict(default_headers or {}) kwargs = dict(model_kwargs) provider_name = str(provider or "").strip().lower() + if cls._matches_endpoint_host(base_url, "opencode.ai"): + # 主对话沿用脱敏缓存键；独立测试、摘要调用的标识在模型实例内复用。 + headers["x-opencode-session"] = cache_key or f"moviepilot-{uuid4().hex}" + if not any(key.lower() == "user-agent" for key in headers): + headers["User-Agent"] = "MoviePilot" + return headers, kwargs if not cache_key: return headers or None, kwargs diff --git a/app/agent/policy/mcp.py b/app/agent/policy/mcp.py index 2d9ea150f..3fa8836ac 100644 --- a/app/agent/policy/mcp.py +++ b/app/agent/policy/mcp.py @@ -396,6 +396,8 @@ FIELD_DESCRIPTIONS = { "sample_rate": "Recorded audio sample rate in hertz.", "save_path": "Configured downloader-side save path for the download or subscription.", "scrape": "Generate metadata and images after manual transfer.", + "search_interval": "Scheduled search interval in whole hours (1-8760); null uses the system interval.", + "last_search": "Read-only UTC timestamp of the most recent subscription search attempt.", "search_imdbid": "Use IMDb identity during subscription search when set to 1.", "season": "Season number used by the media, search, subscription, or transfer operation.", "seasons": "Season-number expression recorded in history.", diff --git a/app/agent/policy/resources/api_mcp_schema.json b/app/agent/policy/resources/api_mcp_schema.json index 120c28c0a..d50d6cfc3 100644 --- a/app/agent/policy/resources/api_mcp_schema.json +++ b/app/agent/policy/resources/api_mcp_schema.json @@ -1706,12 +1706,7 @@ "tvdb", "musicbrainz", "theaudiodb", - "doubanmusic", - "bilibili", - "mangguodiscover", - "migu", - "tencentvideodiscover", - "iqiyidiscover" + "doubanmusic" ], "pattern": "^[a-z][a-z0-9._-]{0,63}$", "title": "MediaSource", @@ -2862,6 +2857,18 @@ "description": "Number of episodes still missing from the subscription.", "title": "Lack Episode" }, + "last_search": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "Read-only UTC timestamp of the most recent subscription search attempt.", + "title": "Last Search" + }, "last_update": { "anyOf": [ { @@ -3057,6 +3064,20 @@ "description": "Use IMDb identity during subscription search when set to 1.", "title": "Search Imdbid" }, + "search_interval": { + "anyOf": [ + { + "maximum": 8760.0, + "minimum": 1.0, + "type": "integer" + }, + { + "type": "null" + } + ], + "description": "Scheduled search interval in whole hours (1-8760); null uses the system interval.", + "title": "Search Interval" + }, "season": { "anyOf": [ { diff --git a/app/api/endpoints/media.py b/app/api/endpoints/media.py index 1e5d7f8e1..641bab50f 100644 --- a/app/api/endpoints/media.py +++ b/app/api/endpoints/media.py @@ -64,11 +64,6 @@ _BUILTIN_MEDIA_SOURCES = ( media_source=MediaSource.DoubanMusic, media_types=[MediaType.MUSIC], ), - _SchemaMediaSourceInfo(name="哔哩哔哩", media_source=MediaSource.Bilibili), - _SchemaMediaSourceInfo(name="芒果TV", media_source=MediaSource.MangoTV), - _SchemaMediaSourceInfo(name="咪咕视频", media_source=MediaSource.MiguVideo), - _SchemaMediaSourceInfo(name="腾讯视频", media_source=MediaSource.TencentVideo), - _SchemaMediaSourceInfo(name="爱奇艺", media_source=MediaSource.Iqiyi), ) diff --git a/app/application/subscription/contract.py b/app/application/subscription/contract.py index 871c67d94..f33392ad8 100644 --- a/app/application/subscription/contract.py +++ b/app/application/subscription/contract.py @@ -21,6 +21,8 @@ _SUBSCRIPTION_FIELDS = frozenset( "year", "type", "keyword", + "search_interval", + "last_search", "media_source", "media_id", "music_type", @@ -83,6 +85,7 @@ _SUBSCRIPTION_HISTORY_FIELDS = (_SUBSCRIPTION_FIELDS - { "note", "state", "last_update", + "last_search", "downloader", "manual_total_episode", }) | _CLASSIFICATION_HISTORY_FIELDS @@ -153,6 +156,7 @@ class SubscriptionSnapshot: year: Optional[str] = None type: Optional[str] = None keyword: Optional[str] = None + search_interval: Optional[int] = None media_source: Optional[MediaSource] = None media_id: Optional[str] = None music_type: Optional[str] = None @@ -179,6 +183,7 @@ class SubscriptionSnapshot: note: Optional[builtins.list[int]] = None state: str = "N" last_update: Optional[str] = None + last_search: Optional[str] = None date: Optional[str] = None username: Optional[str] = None sites: Optional[builtins.list[int]] = None @@ -219,6 +224,7 @@ class SubscriptionHistorySnapshot: year: Optional[str] = None type: Optional[str] = None keyword: Optional[str] = None + search_interval: Optional[int] = None media_source: Optional[MediaSource] = None media_id: Optional[str] = None music_type: Optional[str] = None diff --git a/app/application/subscription/execution.py b/app/application/subscription/execution.py index 780718200..a401999f7 100644 --- a/app/application/subscription/execution.py +++ b/app/application/subscription/execution.py @@ -119,7 +119,8 @@ def raise_subscription_site_budget_deferral( return retry_at = min(deferrals, key=lambda item: item.retry_at).retry_at site_ids = tuple(dict.fromkeys(item.site_id for item in deferrals)) - raise SubscriptionSearchDeferred(retry_at=retry_at, site_ids=site_ids) + wait_reason = "cooldown" if all(item.wait_reason == "cooldown" for item in deferrals) else "busy" + raise SubscriptionSearchDeferred(retry_at=retry_at, site_ids=site_ids, wait_reason=wait_reason) def handle_subscription_search_deferred( @@ -135,7 +136,8 @@ def handle_subscription_search_deferred( lease_token=lease_token, available_at=deferred.retry_at, phase="waiting_site_budget", - message="站点暂时忙，系统会自动继续搜索", + message=str(deferred), + pending_site_ids=deferred.site_ids, ) if requeued: record("requeued", "site_budget_deferred") @@ -184,6 +186,7 @@ class SearchTaskSnapshot: finished_at: Optional[str] = None last_error: Optional[str] = None current_site_id: Optional[int] = None + pending_site_ids: Optional[tuple[int, ...]] = None @dataclass(frozen=True, slots=True) @@ -265,8 +268,9 @@ class SubscriptionSearchRepository(Protocol): available_at: str, phase: str = "waiting_site_budget", message: Optional[str] = None, + pending_site_ids: Optional[tuple[int, ...]] = None, ) -> bool: - """把临时不可执行任务退回队列，并保留用户可理解的等待原因。""" + """延后任务并保存等待原因与待搜站点；省略站点时保留已有游标。""" ... def is_cancel_requested(self, task_id: str) -> bool: diff --git a/app/application/subscription/query.py b/app/application/subscription/query.py index 9b5f51727..9442ab028 100644 --- a/app/application/subscription/query.py +++ b/app/application/subscription/query.py @@ -3,6 +3,7 @@ from __future__ import annotations from collections.abc import Mapping +from datetime import datetime, timedelta from typing import Optional from app.application.subscription.contract import ( @@ -253,3 +254,25 @@ class SubscriptionQueryService: return any( subscribe.type == MediaType.MUSIC.value for subscribe in self._repository.list(searchable_states) or [] ) + + +def subscription_search_due( + subscribe: SubscriptionSnapshot, + default_interval: int, + now: datetime, +) -> bool: + """按最近一次主动搜索计算定时搜索到期；旧记录以本地创建时间起算。""" + if subscribe.state not in {"R", "P"}: + return False + previous = subscribe.last_search or subscribe.date + if not previous: + return True + try: + started = datetime.fromisoformat(previous) + except ValueError: + return True + # 旧创建时间没有时区，保持其本地时间语义；新搜索时间始终携带 UTC 时区。 + if started.tzinfo is None: + started = started.astimezone() + interval = subscribe.search_interval or default_interval + return now >= started + timedelta(hours=interval) diff --git a/app/application/subscription/sitebudget.py b/app/application/subscription/sitebudget.py index b1ac30dad..21f1a2c8c 100644 --- a/app/application/subscription/sitebudget.py +++ b/app/application/subscription/sitebudget.py @@ -21,14 +21,17 @@ class SubscriptionSiteBudgetDeferral: site_id: int retry_at: str + wait_reason: Optional[str] = None class SubscriptionSearchDeferred(RuntimeError): """表示订阅搜索未失败，而是应在站点预算可用后重新入队。""" - def __init__(self, *, retry_at: str, site_ids: tuple[int, ...]) -> None: + def __init__( + self, *, retry_at: str, site_ids: tuple[int, ...], wait_reason: Optional[str] = None, + ) -> None: """保存队列恢复所需的时间和站点，避免把临时等待写成错误。""" - super().__init__("站点暂时忙，系统会自动继续搜索") + super().__init__("站点冷却中" if wait_reason == "cooldown" else "等待站点") self.retry_at = retry_at self.site_ids = site_ids @@ -176,8 +179,10 @@ class SubscriptionSiteBudget: clock: Callable[[], datetime] = _utc_now, phase_changed: Optional[Callable[[str, Optional[int]], None]] = None, metrics: Optional[SubscriptionSiteBudgetMetrics] = None, + pending_site_ids: Optional[tuple[int, ...]] = None, ) -> None: - """保存持久化端口及可注入的时钟和阶段回调。""" + """保存站点预算及恢复范围；首次搜索的 pending_site_ids 为 None。""" + self.pending_site_ids = pending_site_ids self._repository = repository self._owner = owner self._cancelled = cancelled diff --git a/app/chain/search/provider.py b/app/chain/search/provider.py index 8506a7b60..0d542f615 100644 --- a/app/chain/search/provider.py +++ b/app/chain/search/provider.py @@ -262,6 +262,7 @@ class _SearchProviderSyncOwner(_SearchOwnerBase): SubscriptionSiteBudgetDeferral( site_id=error.site_id, retry_at=error.retry_at, + wait_reason=error.wait_reason, ) ) logger.debug(str(error)) @@ -429,11 +430,16 @@ class _SearchProviderSyncOwner(_SearchOwnerBase): area: Optional[str] = "title", mtype: Optional[MediaType] = None, ) -> Optional[List[TorrentInfo]]: - """通过共享线程 owner 按站点顺序翻页并汇总同步 provider 结果。""" + """共享线程按站点翻页；恢复搜索仅查询待完成且仍启用的站点。""" indexer_sites = self._sync_indexers(sites) media_type = self._torrent_type(mediainfo, mtype) search_keyword = self._torrent_keyword(keyword, mediainfo, area) - plugin_results = self.search_plugin_torrents( + budget = getattr(self, "_subscription_site_budget", None) + pending_site_ids = budget.pending_site_ids if isinstance(budget, SubscriptionSiteBudget) else None + # 已完成的站点和插件源不再重复查询；仍按当前启用配置过滤被移除的站点。 + if pending_site_ids is not None: + indexer_sites = [site for site in indexer_sites if site.get("id") in pending_site_ids] + plugin_results = [] if pending_site_ids is not None else self.search_plugin_torrents( keyword=search_keyword, mtype=media_type, page=page, diff --git a/app/chain/subscribe/search.py b/app/chain/subscribe/search.py index 4ec69d975..82fab8ea1 100644 --- a/app/chain/subscribe/search.py +++ b/app/chain/subscribe/search.py @@ -1,5 +1,6 @@ """订阅主动搜索编排""" +import asyncio import random import time from datetime import datetime, timedelta, timezone @@ -28,7 +29,7 @@ from app.application.subscription.observability import ( batch_progress_text, inline_search_result, ) -from app.application.subscription.query import SubscriptionQueryService +from app.application.subscription.query import SubscriptionQueryService, subscription_search_due from app.application.subscription.sitebudget import ( SubscriptionSearchCancelled, SubscriptionSearchDeferred, @@ -46,8 +47,10 @@ from app.domain.context import ( MusicInfo, ) from app.domain.meta.metabase import MetaBase +from app.runtime.execution import await_task_to_terminal from app.runtime.log import logger from app.runtime.stop import runtime_stop_state +from app.runtime.tasks import get_task_registry from app.schemas.types import ( MediaType, SystemConfigKey, @@ -92,19 +95,14 @@ def _search_task_available_at( *, now: Optional[datetime] = None, ) -> dict[int, str]: - """把兜底搜索的随机节奏持久化为逐订阅到期时间。""" + """整批仅抖动一次；请求节奏由站点限流控制，不随订阅数量累计空等。""" ordered_ids = tuple(dict.fromkeys(subscription_ids)) if not ordered_ids: return {} cursor = now or datetime.now(timezone.utc) if source == "fallback": cursor += timedelta(seconds=random.randint(0, 60)) - available_at: dict[int, str] = {} - for position, subscription_id in enumerate(ordered_ids): - if source == "fallback" and position: - cursor += timedelta(seconds=random.randint(60, 300)) - available_at[subscription_id] = cursor.isoformat(timespec="seconds") - return available_at + return dict.fromkeys(ordered_ids, cursor.isoformat(timespec="seconds")) class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): @@ -160,11 +158,12 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): manual: Optional[bool] = False, progress_callback: Optional[Callable[..., None]] = None, sids: Optional[tuple[int, ...]] = None, + scheduled_interval: Optional[int] = None, ) -> Optional[str]: """ 执行订阅搜索。 - 保持定时任务、API 和插件使用的公开签名，搜索实现委托给内部执行阶段。 + scheduled_interval 仅供定时调度传入系统间隔；手动和指定目标搜索不受周期限制。 """ return self._execute_search( sid=sid, @@ -172,6 +171,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): state=state, manual=manual, progress_callback=progress_callback, + scheduled_interval=scheduled_interval, ) def _execute_search( @@ -181,6 +181,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): manual: Optional[bool] = False, progress_callback: Optional[Callable[..., None]] = None, sids: Optional[tuple[int, ...]] = None, + scheduled_interval: Optional[int] = None, ) -> Optional[str]: """ 订阅搜索 @@ -202,6 +203,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): state=state, manual=manual, progress_callback=progress_callback, + scheduled_interval=scheduled_interval, ) self._execute_inline_search( sid=sid, @@ -209,6 +211,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): state=state, manual=manual, progress_callback=progress_callback, + scheduled_interval=scheduled_interval, ) return None @@ -219,6 +222,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): state: Optional[str], manual: Optional[bool], progress_callback: Optional[Callable[..., None]], + scheduled_interval: Optional[int] = None, ) -> None: """在独立 Search 通道内按订阅准入执行兼容搜索。""" lock_acquired = self._acquire_run_lock("search", progress_callback) @@ -228,7 +232,10 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): processed = [] summary: Optional[SearchExecutionSummary] = None try: - subscribes = self._load_search_subscriptions(sid=sid, sids=sids, state=state) + subscribes = self._load_search_subscriptions( + sid=sid, sids=sids, state=state, + scheduled_interval=None if manual else scheduled_interval, + ) total = len(subscribes) source, _priority = _search_source_and_priority( sid=sid, @@ -352,9 +359,17 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase): state: Optional[str], manual: Optional[bool], progress_callback: Optional[Callable[..., None]], - ) -> str: + scheduled_interval: Optional[int] = None, + ) -> Optional[str]: """将搜索转为持久任务并在无 Match 长锁的短租约中串行消费。""" - subscribes = self._load_search_subscriptions(sid=sid, sids=sids, state=state) + subscribes = self._load_search_subscriptions( + sid=sid, sids=sids, state=state, + scheduled_interval=None if manual else scheduled_interval, + ) + if scheduled_interval is not None and not subscribes: + if progress_callback: + progress_callback(value=100, text="暂无到期订阅，无需搜索") + return None source, priority = _search_source_and_priority( sid=sid, sids=sids, @@ -501,6 +516,27 @@ class _SubscribeSearchQueueOwner(_SubscribeSearchQueueCoordinator): ) return runner.execute() + async def async_resume_search_queue( + self, + progress_callback: Optional[Callable[..., None]] = None, + limit: int = 50, + manual_sids: Optional[tuple[int, ...]] = None, + ) -> None: + """托管恢复协程及手工反馈，等待同步消费者真实结束后才释放运行所有权。""" + task = get_task_registry().create_sync( + self.resume_search_queue, + owner="subscription.search_queue", + progress_callback=progress_callback, + limit=limit, + manual_sids=manual_sids, + ) + try: + await asyncio.shield(task) + except asyncio.CancelledError: + # 同步线程不可强制取消；真实收尾前不能让下一次轮询取得消费者所有权。 + await await_task_to_terminal(task) + raise + def resume_search_queue( self, progress_callback: Optional[Callable[..., None]] = None, @@ -573,6 +609,7 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): sid: Optional[int], sids: Optional[tuple[int, ...]], state: Optional[str], + scheduled_interval: Optional[int] = None, ) -> list[SubscriptionSnapshot]: """按单条、指定批次或状态读取本轮搜索订阅。""" repository = self.subscription_repository @@ -581,7 +618,11 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): return [subscribe] if subscribe else [] if sids is not None: return [item for current_id in sids if (item := repository.get(current_id)) is not None] - return cast(list[SubscriptionSnapshot], repository.list(self.get_states_for_search(state or "N"))) + subscribes = cast(list[SubscriptionSnapshot], repository.list(self.get_states_for_search(state or "N"))) + if scheduled_interval is None: + return subscribes + now = datetime.now(timezone.utc) + return [item for item in subscribes if subscription_search_due(item, scheduled_interval, now)] @staticmethod def _recent_subscription_retry_at( @@ -682,6 +723,11 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): ) -> Optional[SubscriptionSnapshot]: """处理单个订阅，并返回下载后重新读取的状态快照。""" _ensure_execution_active(execution_context) + subscribe = self._SubscribeChain__apply_subscribe_update( + subscribe, + {"last_search": datetime.now(timezone.utc).isoformat(timespec="seconds")}, + scene="search", + ) logger.debug(f"开始搜索订阅，标题：{subscribe.name} ...") target = prepare_search_target( self, subscribe, MediaChain(), partial(_ensure_execution_active, execution_context), diff --git a/app/chain/subscribe/searchtask.py b/app/chain/subscribe/searchtask.py index 3a440208a..5175dc4b3 100644 --- a/app/chain/subscribe/searchtask.py +++ b/app/chain/subscribe/searchtask.py @@ -209,6 +209,7 @@ class SubscriptionSearchTaskRunner: stop_state=self.stop_state, phase_changed=phase_changed, metrics=self.summary.site_metrics, + pending_site_ids=self.task.pending_site_ids, ) ) current = self.process_subscription( diff --git a/app/db/adapters/subscription.py b/app/db/adapters/subscription.py index 6db8d42e1..131512306 100644 --- a/app/db/adapters/subscription.py +++ b/app/db/adapters/subscription.py @@ -77,6 +77,7 @@ def _project_subscription(record: Subscribe) -> SubscriptionSnapshot: year=record.year, type=record.type, keyword=record.keyword, + search_interval=record.search_interval, media_source=_media_source(record.media_source), media_id=record.media_id, music_type=record.music_type, @@ -103,6 +104,7 @@ def _project_subscription(record: Subscribe) -> SubscriptionSnapshot: note=cast(Optional[builtins.list[int]], record.note), state=record.state, last_update=record.last_update, + last_search=record.last_search, date=record.date, username=record.username, sites=cast(Optional[builtins.list[int]], record.sites), @@ -134,6 +136,7 @@ def _project_history(record: SubscribeHistory) -> SubscriptionHistorySnapshot: year=record.year, type=record.type, keyword=record.keyword, + search_interval=record.search_interval, media_source=_media_source(record.media_source), media_id=record.media_id, music_type=record.music_type, diff --git a/app/db/adapters/subscriptionsearch.py b/app/db/adapters/subscriptionsearch.py index d30b63019..42d846fcf 100644 --- a/app/db/adapters/subscriptionsearch.py +++ b/app/db/adapters/subscriptionsearch.py @@ -67,6 +67,7 @@ def _task(record: SubscriptionSearchTask) -> SearchTaskSnapshot: finished_at=record.finished_at, last_error=record.last_error, current_site_id=record.current_site_id, + pending_site_ids=tuple(record.pending_site_ids) if record.pending_site_ids is not None else None, ) @@ -248,8 +249,9 @@ class TransactionalSubscriptionSearchRepository: available_at: str, phase: str = "waiting_site_budget", message: Optional[str] = None, + pending_site_ids: Optional[tuple[int, ...]] = None, ) -> bool: - """按指定时间和可见原因重新排队任务。""" + """重新排队并保存待搜站点；未提供站点时保留原游标。""" return self._write( lambda repository: repository.defer_task( task_id=task_id, @@ -257,6 +259,7 @@ class TransactionalSubscriptionSearchRepository: available_at=available_at, phase=phase, message=message, + pending_site_ids=pending_site_ids, ) ) diff --git a/app/db/models/subscribe.py b/app/db/models/subscribe.py index fcafe32cf..27da6026d 100644 --- a/app/db/models/subscribe.py +++ b/app/db/models/subscribe.py @@ -22,6 +22,8 @@ class Subscribe(Base): year: Mapped[Optional[str]] = mapped_column(String) # 类型 type: Mapped[Optional[str]] = mapped_column(String) + # 自定义定时搜索间隔（小时）；空值跟随系统 + search_interval: Mapped[Optional[int]] = mapped_column(Integer) # 搜索关键字 keyword: Mapped[Optional[str]] = mapped_column(String) media_source: Mapped[Optional[str]] = mapped_column(String, index=True) @@ -72,6 +74,8 @@ class Subscribe(Base): note: Mapped[Optional[Any]] = mapped_column(JSON) # 状态：N-新建 R-订阅中 P-待定 S-暂停 state: Mapped[str] = mapped_column(String, nullable=False, index=True, default="N") + # 最近一次主动搜索开始时间，使用带时区的 UTC 时间 + last_search: Mapped[Optional[str]] = mapped_column(String) # 最后更新时间 last_update: Mapped[Optional[str]] = mapped_column(String) # 创建时间 diff --git a/app/db/models/subscribehistory.py b/app/db/models/subscribehistory.py index f4ee645c9..943292904 100644 --- a/app/db/models/subscribehistory.py +++ b/app/db/models/subscribehistory.py @@ -20,6 +20,8 @@ class SubscribeHistory(Base): year: Mapped[Optional[str]] = mapped_column(String) # 类型 type: Mapped[Optional[str]] = mapped_column(String) + # 自定义定时搜索间隔（小时）；空值跟随系统 + search_interval: Mapped[Optional[int]] = mapped_column(Integer) # 搜索关键字 keyword: Mapped[Optional[str]] = mapped_column(String) media_source: Mapped[Optional[str]] = mapped_column(String, index=True) diff --git a/app/db/models/subscriptionsearch.py b/app/db/models/subscriptionsearch.py index c5355a30b..49d38f15f 100644 --- a/app/db/models/subscriptionsearch.py +++ b/app/db/models/subscriptionsearch.py @@ -2,7 +2,7 @@ from typing import Optional -from sqlalchemy import Index, Integer, String, Text, UniqueConstraint +from sqlalchemy import JSON, Index, Integer, String, Text, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column from app.db.base import Base, get_id_column @@ -48,6 +48,8 @@ class SubscriptionSearchTask(Base): state: Mapped[str] = mapped_column(String(32), nullable=False, default="queued") phase: Mapped[str] = mapped_column(String(32), nullable=False, default="queued") current_site_id: Mapped[Optional[int]] = mapped_column(Integer) + # None 表示首次完整搜索；重试只保留尚未完成的站点，跨重启不重复请求成功站点。 + pending_site_ids: Mapped[Optional[list[int]]] = mapped_column(JSON(none_as_null=True)) attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) cancel_requested: Mapped[int] = mapped_column(Integer, nullable=False, default=0) lease_owner: Mapped[Optional[str]] = mapped_column(String(128)) diff --git a/app/db/oper/subscriptionsearch.py b/app/db/oper/subscriptionsearch.py index 726004a40..2b48426b1 100644 --- a/app/db/oper/subscriptionsearch.py +++ b/app/db/oper/subscriptionsearch.py @@ -108,6 +108,11 @@ class SubscriptionSearchOper(DbOper): (promote_queued_task, None), else_=SubscriptionSearchTask.last_error, ), + # 用户重新指定搜索时按当前站点配置重搜；自动周期合并保留恢复游标。 + pending_site_ids=case( + (and_(SubscriptionSearchTask.state == "queued", source in {"manual", "targeted"}), None), + else_=SubscriptionSearchTask.pending_site_ids, + ), available_at=case( ( or_( @@ -138,7 +143,7 @@ class SubscriptionSearchOper(DbOper): return batch, created, coalesced, tuple(dict.fromkeys(active_batch_ids)) def claim_next(self, *, owner: str, lease_seconds: int) -> Optional[SubscriptionSearchTask]: - """使用 CAS 认领最高优先级任务，过期 running 任务可被恢复。""" + """CAS 认领任务并清除旧等待提示，过期 running 任务保留站点游标恢复。""" if not isinstance(self._db, Session): raise RuntimeError("订阅搜索认领需要调用方提供同步 Session") now = utc_now_text() @@ -209,6 +214,7 @@ class SubscriptionSearchOper(DbOper): .values( state="running", phase="matching", + last_error=None, current_site_id=None, lease_owner=owner, lease_token=lease_token, @@ -403,8 +409,9 @@ class SubscriptionSearchOper(DbOper): available_at: str, phase: str, message: Optional[str], + pending_site_ids: Optional[tuple[int, ...]] = None, ) -> bool: - """释放当前租约并在指定时间后按可见原因恢复同一任务。""" + """释放租约并保存重试站点；普通准入等待保留已有站点游标。""" if not isinstance(self._db, Session): raise RuntimeError("订阅搜索延后需要调用方提供同步 Session") task = self._db.execute( @@ -443,6 +450,7 @@ class SubscriptionSearchOper(DbOper): updated_at=now, finished_at=None, last_error=message, + pending_site_ids=list(pending_site_ids) if pending_site_ids is not None else task.pending_site_ids, ), execution_options={"synchronize_session": False}, ) diff --git a/app/domain/classification/sources.py b/app/domain/classification/sources.py index f48913338..7278ebaf4 100644 --- a/app/domain/classification/sources.py +++ b/app/domain/classification/sources.py @@ -18,15 +18,8 @@ FIXTURE_CLASSIFICATION_SOURCES: Final[tuple[str, ...]] = ( ) """首版分类体系具备真实标准投影 fixture 的内置来源顺序。""" -BUILTIN_CLASSIFICATION_SOURCES: Final[tuple[str, ...]] = ( - *FIXTURE_CLASSIFICATION_SOURCES, - MediaSource.Bilibili.value, - MediaSource.MangoTV.value, - MediaSource.MiguVideo.value, - MediaSource.TencentVideo.value, - MediaSource.Iqiyi.value, -) -"""媒体来源 API 当前暴露的全部内置来源顺序。""" +BUILTIN_CLASSIFICATION_SOURCES: Final[tuple[str, ...]] = FIXTURE_CLASSIFICATION_SOURCES +"""仅保留宿主模块实现的来源；插件来源由启用插件声明，不预占标识。""" STANDARD_CLASSIFICATION_FIELD_IDS: Final[tuple[str, ...]] = ( "identity.media_source", @@ -155,11 +148,6 @@ _SOURCE_FIELD_SUPPORT: Final[dict[str, dict[str, ClassificationSourceSupport]]] "music.genres": _PARTIAL, "music.tags": _PARTIAL, }, - MediaSource.Bilibili.value: {}, - MediaSource.MangoTV.value: {}, - MediaSource.MiguVideo.value: {}, - MediaSource.TencentVideo.value: {}, - MediaSource.Iqiyi.value: {}, } diff --git a/app/locales/en-US.json b/app/locales/en-US.json index 3e59ac9ce..40213b0d7 100644 --- a/app/locales/en-US.json +++ b/app/locales/en-US.json @@ -103,6 +103,7 @@ } }, "messages": { + "暂无到期订阅，无需搜索": "No subscriptions are due for search", "调用工具失败": "Tool call failed", "无效的媒体来源": "Invalid media source", "该媒体来源不支持此音乐接口": "This media source is not supported by this music endpoint", diff --git a/app/locales/zh-TW.json b/app/locales/zh-TW.json index 12d1e11ab..eb8e4b422 100644 --- a/app/locales/zh-TW.json +++ b/app/locales/zh-TW.json @@ -103,6 +103,7 @@ } }, "messages": { + "暂无到期订阅，无需搜索": "暫無到期訂閱，無需搜尋", "调用工具失败": "調用工具失敗", "媒体来源和媒体 ID 必须同时提供": "媒體來源和媒體 ID 必須同時提供", "media_source 和 media_id 必须同时提供": "media_source 和 media_id 必須同時提供", diff --git a/app/scheduler/catalog.py b/app/scheduler/catalog.py index 320a2c0bb..f8e358962 100644 --- a/app/scheduler/catalog.py +++ b/app/scheduler/catalog.py @@ -36,11 +36,14 @@ class _MediaServerSchedule(TypedDict): interval: int -def _subscription_search_job_specs(services: SchedulerServices) -> tuple[JobSpec, ...]: +def _subscription_search_job_specs( + services: SchedulerServices, search_interval: int = 24, +) -> tuple[JobSpec, ...]: """构造订阅搜索、新增搜索与持久队列恢复任务目录。""" return ( JobSpec( - "subscribe_search", "订阅搜索补全", services.search_subscribe, "subscription", kwargs={"state": "R"} + "subscribe_search", "订阅搜索补全", services.search_subscribe, "subscription", + kwargs={"state": "R", "scheduled_interval": search_interval} ), JobSpec( "new_subscribe_search", @@ -147,16 +150,20 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase): replace_existing=True, ) + def _poll_subscription_search_queue(self) -> None: + """仅在消费者空闲时唤醒托管协程，轮询不等待搜索也不重复报告重入。""" + if not self._is_job_active("subscribe_search_queue"): + self.start("subscribe_search_queue") + def _register_subscription_search_queue_job(self, config: SchedulerRuntimeConfig) -> None: - """注册短周期持久搜索队列恢复任务。""" + """注册轻量轮询；搜索协程的真实生命周期由 Scheduler 持有。""" self._scheduler.add_job( - self.start, + self._poll_subscription_search_queue, "interval", id="subscribe_search_queue", name="恢复订阅搜索队列", seconds=10, next_run_time=datetime.now(pytz.timezone(config.timezone)) + timedelta(seconds=5), - kwargs={"job_id": "subscribe_search_queue"}, ) def _initialize_catalog(self, config: SchedulerRuntimeConfig) -> None: @@ -168,7 +175,7 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase): JobSpec("cookiecloud", "同步CookieCloud站点", services.sync_cookies, "site"), JobSpec("mediaserver_sync", "同步媒体服务器", services.sync_mediaserver, "mediaserver"), JobSpec("subscribe_tmdb", "订阅元数据更新", services.check_subscribe, "subscription"), - *_subscription_search_job_specs(services), + *_subscription_search_job_specs(services, config.subscribe_search_interval), JobSpec("subscribe_refresh", "订阅刷新", services.refresh_subscribe, "subscription"), JobSpec("subscribe_follow", "关注的订阅分享", services.follow_subscribe, "subscription"), JobSpec( @@ -298,14 +305,14 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase): kwargs={"job_id": "subscribe_tmdb"}, ) - # 订阅状态每隔24小时搜索一次 + # 每五分钟检查逐条订阅到期时间；实际搜索仍受系统或独立周期约束。 if config.subscribe_search: self._scheduler.add_job( self.start, "interval", id="subscribe_search", name="订阅搜索补全", - hours=config.subscribe_search_interval, + minutes=5, kwargs={"job_id": "subscribe_search"}, ) diff --git a/app/schemas/event.py b/app/schemas/event.py index a40b0fcc6..35a38f731 100644 --- a/app/schemas/event.py +++ b/app/schemas/event.py @@ -900,22 +900,9 @@ class DiscoverMediaSource(BaseModel): if media_source and not mediaid_prefix: normalized["mediaid_prefix"] = str(media_source) elif mediaid_prefix and not media_source: - normalized["media_source"] = cls._media_source_from_prefix( - str(mediaid_prefix) - ) + normalized["media_source"] = MediaSource(str(mediaid_prefix)) return normalized - @staticmethod - def _media_source_from_prefix(mediaid_prefix: str) -> MediaSource: - """将旧插件前缀映射为内置或插件扩展媒体来源。""" - aliases = { - "mangguo": MediaSource.MangoTV, - "tencentvideo": MediaSource.TencentVideo, - } - if mediaid_prefix in aliases: - return aliases[mediaid_prefix] - return MediaSource(mediaid_prefix) - class MediaSourceInfo(BaseModel): """ diff --git a/app/schemas/media.py b/app/schemas/media.py index ca5e2a341..bb4add604 100644 --- a/app/schemas/media.py +++ b/app/schemas/media.py @@ -28,15 +28,6 @@ MEDIA_SOURCE_ALIASES = { "audio_db": MediaSource.TheAudioDB, "doubanmusic": MediaSource.DoubanMusic, "douban_music": MediaSource.DoubanMusic, - "bilibili": MediaSource.Bilibili, - "mangguodiscover": MediaSource.MangoTV, - "mango_tv": MediaSource.MangoTV, - "migu": MediaSource.MiguVideo, - "migu_video": MediaSource.MiguVideo, - "tencentvideodiscover": MediaSource.TencentVideo, - "tencent_video": MediaSource.TencentVideo, - "iqiyi": MediaSource.Iqiyi, - "iqiyidiscover": MediaSource.Iqiyi, } MEDIA_SOURCE_PREFIXES = { @@ -49,11 +40,6 @@ MEDIA_SOURCE_PREFIXES = { MediaSource.MusicBrainz: "musicbrainz", MediaSource.TheAudioDB: "theaudiodb", MediaSource.DoubanMusic: "doubanmusic", - MediaSource.Bilibili: "bilibili", - MediaSource.MangoTV: "mangguodiscover", - MediaSource.MiguVideo: "migu", - MediaSource.TencentVideo: "tencentvideodiscover", - MediaSource.Iqiyi: "iqiyidiscover", } diff --git a/app/schemas/subscribe.py b/app/schemas/subscribe.py index 935c78132..c04b8e4f6 100644 --- a/app/schemas/subscribe.py +++ b/app/schemas/subscribe.py @@ -111,7 +111,7 @@ class Subscribe(OptionalMediaIdentityMixin, BaseModel): "note", "state", "last_update", "username", "current_priority", "episode_priority", "date", "current_audio_format", "current_bitrate", "current_bit_depth", "current_sample_rate", "classification_rule_id", "classification_policy_revision", "classification_source", - "execution_status", + "execution_status", "last_search", }) id: Optional[int] = None @@ -121,6 +121,10 @@ class Subscribe(OptionalMediaIdentityMixin, BaseModel): year: Optional[str] = None # 订阅类型 电影/电视剧 type: Optional[str] = None + # 定时搜索间隔（小时）；空值跟随系统设置 + search_interval: Optional[int] = Field(default=None, ge=1, le=8760) + # 最近一次主动搜索开始时间，仅供读取 + last_search: Optional[str] = None # 搜索关键字 keyword: Optional[str] = None media_source: Optional[MediaSource] = None @@ -262,7 +266,7 @@ class Subscribe(OptionalMediaIdentityMixin, BaseModel): data = dict(data) for key, value in list(data.items()): if isinstance(value, str) and value == "": - if key in {"media_source", "media_id"} or key in cls.CLEARABLE_FILTER_FIELDS: + if key in {"media_source", "media_id", "search_interval"} or key in cls.CLEARABLE_FILTER_FIELDS: data[key] = None else: data.pop(key) diff --git a/app/schemas/types.py b/app/schemas/types.py index 8ab951639..96493351a 100644 --- a/app/schemas/types.py +++ b/app/schemas/types.py @@ -85,10 +85,6 @@ _MEDIA_SOURCE_VALUE_ALIASES = { "tmdb": "themoviedb", "audio_db": "theaudiodb", "douban_music": "doubanmusic", - "mango_tv": "mangguodiscover", - "migu_video": "migu", - "tencent_video": "tencentvideodiscover", - "iqiyi": "iqiyidiscover", } @@ -104,11 +100,6 @@ class MediaSource(str, Enum): MusicBrainz = "musicbrainz" TheAudioDB = "theaudiodb" DoubanMusic = "doubanmusic" - Bilibili = "bilibili" - MangoTV = "mangguodiscover" - MiguVideo = "migu" - TencentVideo = "tencentvideodiscover" - Iqiyi = "iqiyidiscover" def __str__(self) -> str: """返回可直接用于 API 和数据库的规范值。""" diff --git a/app/startup/initializers/scheduler.py b/app/startup/initializers/scheduler.py index 4e2e952c8..59e46b8de 100644 --- a/app/startup/initializers/scheduler.py +++ b/app/startup/initializers/scheduler.py @@ -61,7 +61,7 @@ def configure_scheduler_services() -> None: sync_mediaserver=mediaserver_chain.sync, check_subscribe=subscribe_chain.check_and_reconcile, search_subscribe=subscribe_chain.search, - resume_subscribe_search=subscribe_chain.resume_search_queue, + resume_subscribe_search=subscribe_chain.async_resume_search_queue, refresh_subscribe=subscribe_chain.refresh, follow_subscribe=subscribe_chain.follow, process_transfer=transfer_chain.process, diff --git a/database/versions/a9c3e7f1b5d8_3_0_31.py b/database/versions/a9c3e7f1b5d8_3_0_31.py new file mode 100644 index 000000000..098c28a2c --- /dev/null +++ b/database/versions/a9c3e7f1b5d8_3_0_31.py @@ -0,0 +1,48 @@ +"""3.0.31 保存订阅搜索重试站点并解除存量批次的累计空等。""" + +# Alembic 的 op 是运行期代理。 +# pylint: disable=no-member + +from datetime import datetime, timezone + +import sqlalchemy as sa +from alembic import op + +revision = "a9c3e7f1b5d8" +down_revision = "f8a2c6e9b1d4" +branch_labels = None +depends_on = None + +_TABLE = "subscriptionsearchtask" + + +def upgrade() -> None: + """增加可恢复站点游标，只提前尚未执行的自动任务，不绕过冷却和用户停止。""" + inspector = sa.inspect(op.get_bind()) + if _TABLE not in inspector.get_table_names(): + return + columns = {column["name"] for column in inspector.get_columns(_TABLE)} + if "pending_site_ids" in columns: + return + op.add_column(_TABLE, sa.Column("pending_site_ids", sa.JSON(none_as_null=True), nullable=True)) + tasks = sa.table( + _TABLE, + sa.column("source"), sa.column("state"), sa.column("phase"), + sa.column("attempt_count"), sa.column("cancel_requested"), sa.column("available_at"), + ) + now = datetime.now(timezone.utc).isoformat(timespec="seconds") + op.execute( + tasks.update().where( + tasks.c.source == "fallback", tasks.c.state == "queued", tasks.c.phase == "queued", + tasks.c.attempt_count == 0, tasks.c.cancel_requested == 0, tasks.c.available_at > now, + ).values(available_at=now) + ) + + +def downgrade() -> None: + """移除站点游标；恢复旧版本后未完成任务仍可完整重试。""" + inspector = sa.inspect(op.get_bind()) + if _TABLE in inspector.get_table_names(): + columns = {column["name"] for column in inspector.get_columns(_TABLE)} + if "pending_site_ids" in columns: + op.drop_column(_TABLE, "pending_site_ids") diff --git a/database/versions/f8a2c6e9b1d4_3_0_30.py b/database/versions/f8a2c6e9b1d4_3_0_30.py new file mode 100644 index 000000000..44defbc8b --- /dev/null +++ b/database/versions/f8a2c6e9b1d4_3_0_30.py @@ -0,0 +1,27 @@ +"""3.0.30 增加逐订阅搜索周期和持久搜索时间。""" + +import sqlalchemy as sa +from alembic import op + +revision = "f8a2c6e9b1d4" +down_revision = "e7f3a9c1d5b2" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + """兼容首次建表和已有数据库，旧订阅默认跟随系统周期。""" + inspector = sa.inspect(op.get_bind()) + for table in ("subscribe", "subscribehistory"): + columns = {column["name"] for column in inspector.get_columns(table)} + if "search_interval" not in columns: + op.add_column(table, sa.Column("search_interval", sa.Integer(), nullable=True)) + if table == "subscribe" and "last_search" not in columns: + op.add_column(table, sa.Column("last_search", sa.String(), nullable=True)) + + +def downgrade() -> None: + """移除新增周期字段，保留原有订阅和历史数据。""" + op.drop_column("subscribe", "last_search") + for table in ("subscribe", "subscribehistory"): + op.drop_column(table, "search_interval") diff --git a/docker/Dockerfile b/docker/Dockerfile index 37a82f6f1..3c619621e 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -304,6 +304,8 @@ COPY --from=uv /uv /usr/local/bin/uv # 浏览器运行依赖 RUN playwright install-deps chromium \ + && apt-get update \ + && apt-get install -y --no-install-recommends libde265-0 \ && apt-get autoremove -y \ && apt-get clean \ && rm -rf \ diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index b6101f92a..be0b15e43 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -755,7 +755,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 983 | -| 内部导入边 | 8,338 | +| 内部导入边 | 8,340 | | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 53（债务已清零，53 条精确 containment） | diff --git a/docs/architecture/optimization-checklist.md b/docs/architecture/optimization-checklist.md index 8834fb491..b42aa854f 100644 --- a/docs/architecture/optimization-checklist.md +++ b/docs/architecture/optimization-checklist.md @@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 983 / 8,338 | `dependency-baseline.json` 当前快照；分类离线词表与下载资源分类新增模块及其受控依赖 | +| 宿主 Python 模块 / 内部依赖边 | 983 / 8,340 | `dependency-baseline.json` 当前快照；分类离线词表、下载资源分类与订阅搜索运行时任务新增模块及其受控依赖 | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | diff --git a/docs/mcp-api.md b/docs/mcp-api.md index b2850e1ae..2cc52ffe9 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -265,7 +265,9 @@ FastAPI 的 HTTP 异常和参数校验异常统一使用 `message`，不再返 #### 媒体识别 / 整理 -媒体识别、搜索和手动整理统一使用 `media_source` + `media_id` 表示媒体主身份。内置来源通过 `MediaSource` 提供 `themoviedb`、`douban`、`bangumi`、`anilist`、`imdb`、`tvdb`、`musicbrainz`、`theaudiodb`、`doubanmusic`、`bilibili`、`mangguodiscover`、`migu` 和 `tencentvideodiscover` 等常量；该列表不是插件来源白名单，插件可以注册符合 OpenAPI 格式约束的稳定扩展标识。`media_id` 是该来源的原生 ID，不添加 `tmdb:` 等前缀。需要精确身份时两个字段必须同时提供，不能只传其中一个。 +媒体识别、搜索和手动整理统一使用 `media_source` + `media_id` 表示媒体主身份。内置来源通过 `MediaSource` 提供 `themoviedb`、`douban`、`bangumi`、`anilist`、`imdb`、`tvdb`、`musicbrainz`、`theaudiodb`、`doubanmusic` 九个具有宿主模块实现的常量；该列表不是插件来源白名单，插件可以注册符合 OpenAPI 格式约束的稳定扩展标识。`media_id` 是该来源的原生 ID，不添加 `tmdb:` 等前缀。需要精确身份时两个字段必须同时提供，不能只传其中一个。 + +媒体来源列表 `/api/v1/media/source` 仅预置上述九个来源，其余来源由启用插件注册后提供。哔哩哔哩、芒果 TV、咪咕视频、腾讯视频、爱奇艺不再占用内置来源标识，宿主也不再转换这些插件来源的旧别名；调用方应使用插件声明的准确来源 ID。 影视自动识别在未指定来源时只使用 TMDB，未命中时不会继续查询其它影视源。音乐路径识别严格按 AcoustID 音频指纹、文件标签、文件名三级依次执行；指纹或标签直接提供 MusicBrainz Recording ID 时，会直接查询 MusicBrainz 详情，标签和文件名标题识别也只使用 MusicBrainz。其它元数据源仅在手动操作通过请求级 `media_source`，或通过完整的 `media_source` + `media_id` 精确指定时使用，不修改系统默认值，也不会跨来源兜底。`MediaInfo` 响应仍可能包含 `tmdb_id`、`douban_id`、`bangumi_id`、`anilist_id` 等跨源映射辅助字段，但这些字段不是通用请求入口。明确归属 `/tmdb`、`/douban`、`/bangumi`、`/anilist` 的接口，以及固定使用 TMDB 的剧集组和排期接口，仍可按其单数据源契约接收原生 ID。 @@ -429,6 +431,24 @@ TMDB 缓存查询响应的 `data` 包含 `count`、`recognized`、`unrecognized` 旧版未包含这些证据的派生缓存在升级后重新建立，不影响下载历史或订阅数据。 名称确认规则更新时同样重建旧派生缓存，避免艺术家前后缀误截断的旧结果继续命中。 +### 单条订阅搜索周期 + +`POST /api/v1/subscribe/` 和 `PUT /api/v1/subscribe/` 支持 `search_interval`： +取值为 1–8760 的整数小时数，`null` 表示跟随系统的 `SUBSCRIBE_SEARCH_INTERVAL`。 +更新时省略该字段保留原值，显式传入 `null` 恢复系统周期。默认订阅规则同样可保存此字段。 + +仅在启用 `SUBSCRIBE_SEARCH` 时执行定时搜索，每五分钟检查到期订阅，再通过原有搜索队列和站点限流执行； +实际开始时间可能因站点忙而延后。`last_search` 为系统维护的 UTC 搜索尝试开始时间，重启后仍有效，公共写接口忽略它。 +旧订阅尚无搜索时间时，以添加时间计算到期。新订阅首次搜索、手动搜索和 RSS 刷新不受周期过滤影响； +手动主动搜索会更新最近搜索时间。电影、电视剧和音乐均支持独立周期。 + +自动批次只在启动时随机错峰 0–60 秒，不再按订阅数量累加分钟级等待；同站点访问间隔、唯一在途租约和错误冷却继续生效。 +站点暂不可用时，任务保存未完成站点并按 `next_run_at` 恢复，不重复查询已完成站点或插件源；队列和站点游标可跨重启恢复。 +`waiting_site_budget` 表示可恢复等待，`error` 中的“等待站点”或“站点冷却中”是原因提示，不表示搜索失败；重新执行时清除旧提示。 +卡片应按 `state` / `phase` 展示简短标签，原因放入详情提示。恢复调度每 10 秒检查空闲消费者，长搜索由调度器持续托管。 +实际吞吐仍受站点访问限制和网络响应时间影响，配置周期不是全部站点必须完成的截止时间。 + + ### 插件补充接口 **GET** `/api/v1/plugin/history/{plugin_id}` diff --git a/skills/database-operation/SKILL.md b/skills/database-operation/SKILL.md index b84a3e904..4d3a6ff28 100644 --- a/skills/database-operation/SKILL.md +++ b/skills/database-operation/SKILL.md @@ -228,13 +228,13 @@ python scripts/mp-db.py write "UPDATE subscribe SET state = 'S' WHERE id = 123" - Purpose: Stores active movie, TV, or music subscriptions, filters, progress, and download targets. - Useful queries: Inspecting state, missing episodes/tracks, quality rules, site scope, and match progress. - Write boundary: Create, update, search, or delete through the subscription API to preserve state-machine consistency. -- Columns: `id`, `name`, `year`, `type`, `keyword`, `media_source`, `media_id`, `music_type`, `total_tracks`, `season`, `poster`, `backdrop`, `vote`, `description`, `filter`, `include`, `exclude`, `quality`, `resolution`, `effect`, `audio_quality`, `audio_format`, `min_bitrate`, `min_bit_depth`, `min_sample_rate`, `total_episode`, `start_episode`, `lack_episode`, `note`, `state`, `last_update`, `date`, `username`, `sites`, `downloader`, `best_version`, `best_version_full`, `current_priority`, `current_audio_format`, `current_bitrate`, `current_bit_depth`, `current_sample_rate`, `episode_priority`, `save_path`, `search_imdbid`, `manual_total_episode`, `custom_words`, `media_category_id`, `media_category`, `filter_groups`, `episode_group` +- Columns: `id`, `name`, `year`, `type`, `search_interval`, `keyword`, `media_source`, `media_id`, `music_type`, `total_tracks`, `season`, `poster`, `backdrop`, `vote`, `description`, `filter`, `include`, `exclude`, `quality`, `resolution`, `effect`, `audio_quality`, `audio_format`, `min_bitrate`, `min_bit_depth`, `min_sample_rate`, `total_episode`, `start_episode`, `lack_episode`, `note`, `state`, `last_search`, `last_update`, `date`, `username`, `sites`, `downloader`, `best_version`, `best_version_full`, `current_priority`, `current_audio_format`, `current_bitrate`, `current_bit_depth`, `current_sample_rate`, `episode_priority`, `save_path`, `search_imdbid`, `manual_total_episode`, `custom_words`, `media_category_id`, `media_category`, `filter_groups`, `episode_group` ### `subscribehistory` - Purpose: Stores snapshots of completed or archived subscriptions and their final filter state. - Useful queries: Auditing historical subscriptions, media identity, completion criteria, and filter configuration. - Write boundary: Generated by subscription completion and archival; restore or delete through its business API. -- Columns: `id`, `name`, `year`, `type`, `keyword`, `media_source`, `media_id`, `music_type`, `total_tracks`, `season`, `poster`, `backdrop`, `vote`, `description`, `filter`, `include`, `exclude`, `quality`, `resolution`, `effect`, `audio_quality`, `audio_format`, `min_bitrate`, `min_bit_depth`, `min_sample_rate`, `total_episode`, `start_episode`, `date`, `username`, `sites`, `best_version`, `best_version_full`, `current_priority`, `current_audio_format`, `current_bitrate`, `current_bit_depth`, `current_sample_rate`, `episode_priority`, `save_path`, `search_imdbid`, `custom_words`, `media_category_id`, `media_category`, `classification_rule_id`, `classification_policy_revision`, `classification_source`, `filter_groups`, `episode_group` +- Columns: `id`, `name`, `year`, `type`, `search_interval`, `keyword`, `media_source`, `media_id`, `music_type`, `total_tracks`, `season`, `poster`, `backdrop`, `vote`, `description`, `filter`, `include`, `exclude`, `quality`, `resolution`, `effect`, `audio_quality`, `audio_format`, `min_bitrate`, `min_bit_depth`, `min_sample_rate`, `total_episode`, `start_episode`, `date`, `username`, `sites`, `best_version`, `best_version_full`, `current_priority`, `current_audio_format`, `current_bitrate`, `current_bit_depth`, `current_sample_rate`, `episode_priority`, `save_path`, `search_imdbid`, `custom_words`, `media_category_id`, `media_category`, `classification_rule_id`, `classification_policy_revision`, `classification_source`, `filter_groups`, `episode_group` ### `subscriptionsearchbatch` - Purpose: Stores durable subscription search batches, source, aggregate state, counts, and cancellation requests. diff --git a/skills/moviepilot-api/SKILL.md b/skills/moviepilot-api/SKILL.md index 76879d0b3..eea38d4b0 100644 --- a/skills/moviepilot-api/SKILL.md +++ b/skills/moviepilot-api/SKILL.md @@ -552,6 +552,7 @@ Purpose: List seasons for one exact media identity or a title-and-year fallback. ### `media.sources` `GET /api/v1/media/source`; policy effect: `safe_read`. Purpose: List metadata sources currently registered for MoviePilot media operations. +Only sources implemented by host modules are built in. Other sources appear after an enabled plugin registers them; use the exact returned identifier without converting plugin source aliases. - `response`: `data` remains a list; omitting both `page` and `count` keeps the complete legacy result. `collection.result_count` reports the returned items and `collection.total_count` reports the exact pre-pagination total. For counts or summaries, send `page=1,count=1`, read `collection.total_count`, and do not fall back to a database query because the item preview was truncated. - `path_params`: none - `query`: `count` (integer|null): Optional page size for a legacy full-list endpoint. Supplying page or count activates pagination; an omitted count then uses 50.; `page` (integer|null): Optional one-based page for a legacy full-list endpoint. Omit both page and count to keep the original unpaginated full result. @@ -1129,7 +1130,7 @@ Purpose: Read configured directory or storage settings. Purpose: Create one movie, TV, or music subscription. - `path_params`: none - `query`: none -- `body`: `audio_format` (string|null): Requested or recorded audio container or codec, such as FLAC or MP3.; `audio_quality` (string|null): Subscription audio-quality rule, such as hires, lossless, or lossy.; `backdrop` (string|null): Backdrop image URL stored with the media or subscription.; `best_version` (integer|null): Enable normal best-version upgrading when set to 1.; `best_version_full` (integer|null): Enable full best-version upgrading when set to 1.; `classification_policy_revision` (integer|null): Policy revision that produced the persisted classification snapshot.; `classification_rule_id` (string|null): Stable rule ID that selected the persisted classification category.; `classification_source` (string|null): Selection source recorded with the persisted classification snapshot.; `completed_episode` (integer|null): Highest episode number already completed for the subscription.; `current_audio_format` (string|null): Audio format of the best version currently held.; `current_bit_depth` (integer|null): Bit depth of the best version currently held.; `current_bitrate` (integer|null): Bitrate of the best version currently held.; `current_priority` (integer|null): Calculated priority of the best version currently held.; `current_sample_rate` (integer|null): Sample rate of the best version currently held.; `custom_words` (string|null): Custom recognition or rename words applied to this media workflow.; `date` (string|null): Record creation or completion timestamp used by the history item.; `description` (string|null): Human-readable media, torrent, or subscription description.; `downloader` (string|null): Configured downloader instance name.; `effect` (string|null): Video or release-effect filter expression used by the subscription.; `episode_group` (string|null): TMDB episode-group identifier used for alternate episode ordering.; `episode_priority` (object|null): Per-episode best-version priority state.; `exclude` (string|null): Regular expression or filter expression that rejects matching releases.; `execution_status` (SubscriptionExecutionStatus|null): Current subscription execution status returned with the subscription snapshot.; `filter` (string|null): Named filter rule or rule expression applied to this site or subscription.; `filter_groups` (array|null): Ordered filter-rule group names applied to the subscription.; `id` (integer|null): Persistent database identifier of the supplied record.; `include` (string|null): Regular expression or filter expression that a release must match.; `keyword` (string|null): Case-insensitive substring used to discover settings or filter storage entries.; `lack_episode` (integer|null; default `0`): Number of episodes still missing from the subscription.; `last_update` (string|null): Timestamp of the subscription's most recent update.; `media_category` (string|null): MoviePilot library category assigned to the media.; `media_category_id` (string|null): Stable classification category ID; preserve it separately from the current category path snapshot.; `media_id` (string|null): Source-native media ID. Always pair it with the exact media_source returned by search.; `media_source` (MediaSource|null): Metadata source identifier. Preserve the exact value returned with media_id.; `min_bit_depth` (integer|null): Minimum acceptable audio bit depth in bits.; `min_bitrate` (integer|null): Minimum acceptable audio bitrate in bits per second.; `min_sample_rate` (integer|null): Minimum acceptable audio sample rate in hertz.; `music_type` (string|null): Music identity level: recording, album, or artist where supported.; `name` (string|null): Human-readable name of the site, storage item, subscription, or rule group.; `note` (array|null): Structured auxiliary metadata stored with the record.; `poster` (string|null): Poster image URL stored with the media or subscription.; `quality` (string|null): Video or release quality filter expression.; `resolution` (string|null): Video resolution filter expression, such as 1080p or 2160p.; `save_path` (string|null): Configured downloader-side save path for the download or subscription.; `search_imdbid` (integer|null; default `0`): Use IMDb identity during subscription search when set to 1.; `season` (integer|null): Season number used by the media, search, subscription, or transfer operation.; `sites` (array|null): Exact site IDs included in the search or subscription scope.; `start_episode` (integer|null; default `0`): First episode number requested by the subscription.; `state` (string|null): Current site, subscription, marketplace, or transfer state filter.; `total_episode` (integer|null; default `0`): Expected total episode count for the subscription.; `total_tracks` (integer|null): Expected or recorded track count for a music item.; `type` (string|null): MoviePilot media or storage item type required by the selected operation.; `username` (string|null): MoviePilot or site username required by the selected operation.; `vote` (number|null; default `0.0`): Media vote average stored with the subscription.; `year` (string|null): Release or premiere year used to disambiguate the media title. +- `body`: `audio_format` (string|null): Requested or recorded audio container or codec, such as FLAC or MP3.; `audio_quality` (string|null): Subscription audio-quality rule, such as hires, lossless, or lossy.; `backdrop` (string|null): Backdrop image URL stored with the media or subscription.; `best_version` (integer|null): Enable normal best-version upgrading when set to 1.; `best_version_full` (integer|null): Enable full best-version upgrading when set to 1.; `classification_policy_revision` (integer|null): Policy revision that produced the persisted classification snapshot.; `classification_rule_id` (string|null): Stable rule ID that selected the persisted classification category.; `classification_source` (string|null): Selection source recorded with the persisted classification snapshot.; `completed_episode` (integer|null): Highest episode number already completed for the subscription.; `current_audio_format` (string|null): Audio format of the best version currently held.; `current_bit_depth` (integer|null): Bit depth of the best version currently held.; `current_bitrate` (integer|null): Bitrate of the best version currently held.; `current_priority` (integer|null): Calculated priority of the best version currently held.; `current_sample_rate` (integer|null): Sample rate of the best version currently held.; `custom_words` (string|null): Custom recognition or rename words applied to this media workflow.; `date` (string|null): Record creation or completion timestamp used by the history item.; `description` (string|null): Human-readable media, torrent, or subscription description.; `downloader` (string|null): Configured downloader instance name.; `effect` (string|null): Video or release-effect filter expression used by the subscription.; `episode_group` (string|null): TMDB episode-group identifier used for alternate episode ordering.; `episode_priority` (object|null): Per-episode best-version priority state.; `exclude` (string|null): Regular expression or filter expression that rejects matching releases.; `execution_status` (SubscriptionExecutionStatus|null): Current subscription execution status returned with the subscription snapshot.; `filter` (string|null): Named filter rule or rule expression applied to this site or subscription.; `filter_groups` (array|null): Ordered filter-rule group names applied to the subscription.; `id` (integer|null): Persistent database identifier of the supplied record.; `include` (string|null): Regular expression or filter expression that a release must match.; `keyword` (string|null): Case-insensitive substring used to discover settings or filter storage entries.; `lack_episode` (integer|null; default `0`): Number of episodes still missing from the subscription.; `last_search` (string|null): Read-only UTC timestamp of the most recent subscription search attempt.; `last_update` (string|null): Timestamp of the subscription's most recent update.; `media_category` (string|null): MoviePilot library category assigned to the media.; `media_category_id` (string|null): Stable classification category ID; preserve it separately from the current category path snapshot.; `media_id` (string|null): Source-native media ID. Always pair it with the exact media_source returned by search.; `media_source` (MediaSource|null): Metadata source identifier. Preserve the exact value returned with media_id.; `min_bit_depth` (integer|null): Minimum acceptable audio bit depth in bits.; `min_bitrate` (integer|null): Minimum acceptable audio bitrate in bits per second.; `min_sample_rate` (integer|null): Minimum acceptable audio sample rate in hertz.; `music_type` (string|null): Music identity level: recording, album, or artist where supported.; `name` (string|null): Human-readable name of the site, storage item, subscription, or rule group.; `note` (array|null): Structured auxiliary metadata stored with the record.; `poster` (string|null): Poster image URL stored with the media or subscription.; `quality` (string|null): Video or release quality filter expression.; `resolution` (string|null): Video resolution filter expression, such as 1080p or 2160p.; `save_path` (string|null): Configured downloader-side save path for the download or subscription.; `search_imdbid` (integer|null; default `0`): Use IMDb identity during subscription search when set to 1.; `search_interval` (integer|null): Scheduled search interval in whole hours (1-8760); null uses the system interval.; `season` (integer|null): Season number used by the media, search, subscription, or transfer operation.; `sites` (array|null): Exact site IDs included in the search or subscription scope.; `start_episode` (integer|null; default `0`): First episode number requested by the subscription.; `state` (string|null): Current site, subscription, marketplace, or transfer state filter.; `total_episode` (integer|null; default `0`): Expected total episode count for the subscription.; `total_tracks` (integer|null): Expected or recorded track count for a music item.; `type` (string|null): MoviePilot media or storage item type required by the selected operation.; `username` (string|null): MoviePilot or site username required by the selected operation.; `vote` (number|null; default `0.0`): Media vote average stored with the subscription.; `year` (string|null): Release or premiere year used to disambiguate the media title. ### `subscription.delete` `DELETE /api/v1/subscribe/{subscribe_id}`; policy effect: `destructive_write`. @@ -1247,6 +1248,10 @@ Purpose: Reset one accessible subscription so it can be processed again. - `query`: none - `body`: none +Subscription search uses a durable queue. Automatic batches share a single 0–60 second startup jitter; site rate limits and cooldowns still apply. +Deferred attempts resume only pending sites, including after restart. Treat `waiting_site_budget` as recoverable waiting and use `next_run_at` for the next attempt; +its `error` text is a waiting reason and is cleared when execution resumes. Do not interpret the configured search interval as a completion deadline. + ### `subscription.search` `POST /api/v1/subscribe/search/{subscribe_id}`; policy effect: `external_side_effect`. Purpose: Run an immediate search for one existing subscription. @@ -1303,7 +1308,7 @@ Purpose: Set one accessible subscription to running, paused, or stopped state. Purpose: Update one existing movie, TV, or music subscription. - `path_params`: none - `query`: none -- `body`: `audio_format` (string|null): Requested or recorded audio container or codec, such as FLAC or MP3.; `audio_quality` (string|null): Subscription audio-quality rule, such as hires, lossless, or lossy.; `backdrop` (string|null): Backdrop image URL stored with the media or subscription.; `best_version` (integer|null): Enable normal best-version upgrading when set to 1.; `best_version_full` (integer|null): Enable full best-version upgrading when set to 1.; `classification_policy_revision` (integer|null): Policy revision that produced the persisted classification snapshot.; `classification_rule_id` (string|null): Stable rule ID that selected the persisted classification category.; `classification_source` (string|null): Selection source recorded with the persisted classification snapshot.; `completed_episode` (integer|null): Highest episode number already completed for the subscription.; `current_audio_format` (string|null): Audio format of the best version currently held.; `current_bit_depth` (integer|null): Bit depth of the best version currently held.; `current_bitrate` (integer|null): Bitrate of the best version currently held.; `current_priority` (integer|null): Calculated priority of the best version currently held.; `current_sample_rate` (integer|null): Sample rate of the best version currently held.; `custom_words` (string|null): Custom recognition or rename words applied to this media workflow.; `date` (string|null): Record creation or completion timestamp used by the history item.; `description` (string|null): Human-readable media, torrent, or subscription description.; `downloader` (string|null): Configured downloader instance name.; `effect` (string|null): Video or release-effect filter expression used by the subscription.; `episode_group` (string|null): TMDB episode-group identifier used for alternate episode ordering.; `episode_priority` (object|null): Per-episode best-version priority state.; `exclude` (string|null): Regular expression or filter expression that rejects matching releases.; `execution_status` (SubscriptionExecutionStatus|null): Current subscription execution status returned with the subscription snapshot.; `filter` (string|null): Named filter rule or rule expression applied to this site or subscription.; `filter_groups` (array|null): Ordered filter-rule group names applied to the subscription.; `id` (integer|null): Persistent database identifier of the supplied record.; `include` (string|null): Regular expression or filter expression that a release must match.; `keyword` (string|null): Case-insensitive substring used to discover settings or filter storage entries.; `lack_episode` (integer|null; default `0`): Number of episodes still missing from the subscription.; `last_update` (string|null): Timestamp of the subscription's most recent update.; `media_category` (string|null): MoviePilot library category assigned to the media.; `media_category_id` (string|null): Stable classification category ID; preserve it separately from the current category path snapshot.; `media_id` (string|null): Source-native media ID. Always pair it with the exact media_source returned by search.; `media_source` (MediaSource|null): Metadata source identifier. Preserve the exact value returned with media_id.; `min_bit_depth` (integer|null): Minimum acceptable audio bit depth in bits.; `min_bitrate` (integer|null): Minimum acceptable audio bitrate in bits per second.; `min_sample_rate` (integer|null): Minimum acceptable audio sample rate in hertz.; `music_type` (string|null): Music identity level: recording, album, or artist where supported.; `name` (string|null): Human-readable name of the site, storage item, subscription, or rule group.; `note` (array|null): Structured auxiliary metadata stored with the record.; `poster` (string|null): Poster image URL stored with the media or subscription.; `quality` (string|null): Video or release quality filter expression.; `resolution` (string|null): Video resolution filter expression, such as 1080p or 2160p.; `save_path` (string|null): Configured downloader-side save path for the download or subscription.; `search_imdbid` (integer|null; default `0`): Use IMDb identity during subscription search when set to 1.; `season` (integer|null): Season number used by the media, search, subscription, or transfer operation.; `sites` (array|null): Exact site IDs included in the search or subscription scope.; `start_episode` (integer|null; default `0`): First episode number requested by the subscription.; `state` (string|null): Current site, subscription, marketplace, or transfer state filter.; `total_episode` (integer|null; default `0`): Expected total episode count for the subscription.; `total_tracks` (integer|null): Expected or recorded track count for a music item.; `type` (string|null): MoviePilot media or storage item type required by the selected operation.; `username` (string|null): MoviePilot or site username required by the selected operation.; `vote` (number|null; default `0.0`): Media vote average stored with the subscription.; `year` (string|null): Release or premiere year used to disambiguate the media title. +- `body`: `audio_format` (string|null): Requested or recorded audio container or codec, such as FLAC or MP3.; `audio_quality` (string|null): Subscription audio-quality rule, such as hires, lossless, or lossy.; `backdrop` (string|null): Backdrop image URL stored with the media or subscription.; `best_version` (integer|null): Enable normal best-version upgrading when set to 1.; `best_version_full` (integer|null): Enable full best-version upgrading when set to 1.; `classification_policy_revision` (integer|null): Policy revision that produced the persisted classification snapshot.; `classification_rule_id` (string|null): Stable rule ID that selected the persisted classification category.; `classification_source` (string|null): Selection source recorded with the persisted classification snapshot.; `completed_episode` (integer|null): Highest episode number already completed for the subscription.; `current_audio_format` (string|null): Audio format of the best version currently held.; `current_bit_depth` (integer|null): Bit depth of the best version currently held.; `current_bitrate` (integer|null): Bitrate of the best version currently held.; `current_priority` (integer|null): Calculated priority of the best version currently held.; `current_sample_rate` (integer|null): Sample rate of the best version currently held.; `custom_words` (string|null): Custom recognition or rename words applied to this media workflow.; `date` (string|null): Record creation or completion timestamp used by the history item.; `description` (string|null): Human-readable media, torrent, or subscription description.; `downloader` (string|null): Configured downloader instance name.; `effect` (string|null): Video or release-effect filter expression used by the subscription.; `episode_group` (string|null): TMDB episode-group identifier used for alternate episode ordering.; `episode_priority` (object|null): Per-episode best-version priority state.; `exclude` (string|null): Regular expression or filter expression that rejects matching releases.; `execution_status` (SubscriptionExecutionStatus|null): Current subscription execution status returned with the subscription snapshot.; `filter` (string|null): Named filter rule or rule expression applied to this site or subscription.; `filter_groups` (array|null): Ordered filter-rule group names applied to the subscription.; `id` (integer|null): Persistent database identifier of the supplied record.; `include` (string|null): Regular expression or filter expression that a release must match.; `keyword` (string|null): Case-insensitive substring used to discover settings or filter storage entries.; `lack_episode` (integer|null; default `0`): Number of episodes still missing from the subscription.; `last_search` (string|null): Read-only UTC timestamp of the most recent subscription search attempt.; `last_update` (string|null): Timestamp of the subscription's most recent update.; `media_category` (string|null): MoviePilot library category assigned to the media.; `media_category_id` (string|null): Stable classification category ID; preserve it separately from the current category path snapshot.; `media_id` (string|null): Source-native media ID. Always pair it with the exact media_source returned by search.; `media_source` (MediaSource|null): Metadata source identifier. Preserve the exact value returned with media_id.; `min_bit_depth` (integer|null): Minimum acceptable audio bit depth in bits.; `min_bitrate` (integer|null): Minimum acceptable audio bitrate in bits per second.; `min_sample_rate` (integer|null): Minimum acceptable audio sample rate in hertz.; `music_type` (string|null): Music identity level: recording, album, or artist where supported.; `name` (string|null): Human-readable name of the site, storage item, subscription, or rule group.; `note` (array|null): Structured auxiliary metadata stored with the record.; `poster` (string|null): Poster image URL stored with the media or subscription.; `quality` (string|null): Video or release quality filter expression.; `resolution` (string|null): Video resolution filter expression, such as 1080p or 2160p.; `save_path` (string|null): Configured downloader-side save path for the download or subscription.; `search_imdbid` (integer|null; default `0`): Use IMDb identity during subscription search when set to 1.; `search_interval` (integer|null): Scheduled search interval in whole hours (1-8760); null uses the system interval.; `season` (integer|null): Season number used by the media, search, subscription, or transfer operation.; `sites` (array|null): Exact site IDs included in the search or subscription scope.; `start_episode` (integer|null; default `0`): First episode number requested by the subscription.; `state` (string|null): Current site, subscription, marketplace, or transfer state filter.; `total_episode` (integer|null; default `0`): Expected total episode count for the subscription.; `total_tracks` (integer|null): Expected or recorded track count for a music item.; `type` (string|null): MoviePilot media or storage item type required by the selected operation.; `username` (string|null): MoviePilot or site username required by the selected operation.; `vote` (number|null; default `0.0`): Media vote average stored with the subscription.; `year` (string|null): Release or premiere year used to disambiguate the media title. ### `subscription.user.list` `GET /api/v1/subscribe/user/{username}`; policy effect: `safe_read`. diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 9c442677a..91dc042a7 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1074,8 +1074,8 @@ "runtime_only": true } }, - "edge_count": 8338, - "edge_sha256": "181d760e7da33f12022ca86f906617a11e1d672d95e7e3fd4c3bc324a9d9dd63", + "edge_count": 8340, + "edge_sha256": "b3092e1b2d83dd6b9356e1107e034c1a5b65f2246a5677f7457687a37aeba4cb", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -4927,8 +4927,10 @@ "app.chain.subscribe.search -> app.domain.meta", "app.chain.subscribe.search -> app.domain.meta.metabase", "app.chain.subscribe.search -> app.runtime", + "app.chain.subscribe.search -> app.runtime.execution", "app.chain.subscribe.search -> app.runtime.log", "app.chain.subscribe.search -> app.runtime.stop", + "app.chain.subscribe.search -> app.runtime.tasks", "app.chain.subscribe.search -> app.schemas", "app.chain.subscribe.search -> app.schemas.types", "app.chain.subscribe.searchtask -> app.application", diff --git a/tests/test_database_backup_scheduler.py b/tests/test_database_backup_scheduler.py index 3e5c32b42..338d6f6ee 100644 --- a/tests/test_database_backup_scheduler.py +++ b/tests/test_database_backup_scheduler.py @@ -148,6 +148,29 @@ def test_scheduled_backup_uses_registered_database_governance(monkeypatch) -> No governance.create_backup.assert_called_once_with() +@pytest.mark.parametrize("enabled", [False, True]) +def test_subscription_search_scans_due_items_only_when_enabled(monkeypatch, enabled) -> None: + """系统开关控制到期扫描，五分钟扫描节奏与实际搜索间隔分别传递。""" + scheduler = _scheduler() + scheduler._services = Mock() + background_scheduler = Mock() + monkeypatch.setattr(scheduler_catalog, "BackgroundScheduler", lambda **_kwargs: background_scheduler) + monkeypatch.setattr(scheduler_catalog, "get_plugin_manager", lambda: Mock()) + monkeypatch.setattr(scheduler_catalog, "get_mediaserver_configs", lambda **_kwargs: []) + monkeypatch.setattr(scheduler, "init_workflow_jobs", lambda: None) + monkeypatch.setattr(scheduler, "init_agent_task_jobs", lambda: None) + monkeypatch.setattr(scheduler, "init_plugin_jobs", lambda: None) + + scheduler._initialize_catalog(_config(subscribe_search=enabled, subscribe_search_interval=48)) + + calls = [call for call in background_scheduler.add_job.call_args_list + if call.kwargs.get("id") == "subscribe_search"] + assert bool(calls) is enabled + if enabled: + assert calls[0].kwargs["minutes"] == 5 + assert scheduler._jobs["subscribe_search"]["kwargs"]["scheduled_interval"] == 48 + + def test_scheduler_database_dependencies_are_explicit_module_imports() -> None: scheduler_root = Path(__file__).parents[1] / "app" / "scheduler" trees = [ast.parse(path.read_text(encoding="utf-8")) for path in scheduler_root.glob("*.py")] diff --git a/tests/test_explicit_response_models.py b/tests/test_explicit_response_models.py index 8dce44482..40725d511 100644 --- a/tests/test_explicit_response_models.py +++ b/tests/test_explicit_response_models.py @@ -158,8 +158,8 @@ def test_collection_json_schemas_define_items_or_tuple_members(): assert crew_schema["items"]["$ref"].endswith("/TmdbEpisodeCrew") -def test_discover_media_source_keeps_legacy_prefix_compatible(): - """发现源应兼容旧插件前缀，并同时输出规范媒体来源。""" +def test_discover_media_source_preserves_plugin_identifiers(): + """发现源按插件标识补齐身份字段，不转换插件特定前缀。""" legacy = DiscoverMediaSource( name="哔哩哔哩", mediaid_prefix="bilibili", @@ -167,7 +167,7 @@ def test_discover_media_source_keeps_legacy_prefix_compatible(): ) current = DiscoverMediaSource( name="腾讯视频", - media_source=MediaSource.TencentVideo, + media_source=MediaSource("tencentvideodiscover"), api_path="plugin/TencentVideoDiscover/discover", ) historical_alias = DiscoverMediaSource( @@ -181,9 +181,9 @@ def test_discover_media_source_keeps_legacy_prefix_compatible(): api_path="plugin/AcmeVideo/discover", ) - assert legacy.media_source is MediaSource.Bilibili + assert legacy.media_source == MediaSource("bilibili") assert legacy.model_dump(mode="json")["mediaid_prefix"] == "bilibili" - assert current.mediaid_prefix == MediaSource.TencentVideo.value - assert historical_alias.media_source is MediaSource.MangoTV + assert current.mediaid_prefix == "tencentvideodiscover" + assert historical_alias.media_source == MediaSource("mangguo") assert plugin_source.media_source == MediaSource("acme.video") assert plugin_source.mediaid_prefix == "acme.video" diff --git a/tests/test_llm_opencode_session.py b/tests/test_llm_opencode_session.py new file mode 100644 index 000000000..f1b42c7e8 --- /dev/null +++ b/tests/test_llm_opencode_session.py @@ -0,0 +1,101 @@ +"""OpenCode 官方端点的会话路由请求头回归测试。""" + +import asyncio +from unittest.mock import AsyncMock + +import httpx +import pytest + +from app.agent.llm import helper +from app.agent.llm.helper import LLMHelper + + +@pytest.mark.parametrize("base_url", [ + "https://opencode.ai/zen/v1", + "https://opencode.ai/zen/go/v1", +]) +@pytest.mark.parametrize("cache_key", ["moviepilot-agent-private-hash", None]) +def test_opencode_model_sends_stable_session_on_every_request(monkeypatch, base_url, cache_key): + """主对话与独立调用经过真实 SDK 后仍携带稳定标识，工具绑定不丢失请求头。""" + requests = [] + + def respond(request): + """在本地捕获 SDK 请求并返回最小聊天响应，禁止真实出站。""" + requests.append(request) + return httpx.Response(200, json={ + "id": "chatcmpl-test", + "object": "chat.completion", + "created": 0, + "model": "test-model", + "choices": [{"index": 0, "message": {"role": "assistant", "content": "OK"}, + "finish_reason": "stop"}], + }) + + transport = httpx.MockTransport(respond) + with httpx.Client(transport=transport) as client: + async def invoke(): + """使用离线运行时创建模型，并验证同步与异步请求使用同一标识。""" + async with httpx.AsyncClient(transport=transport) as async_client: + monkeypatch.setattr( + helper, "_build_httpx_client", + lambda _proxy, **kwargs: async_client if kwargs.get("async_client") else client, + ) + runtime = AsyncMock() + runtime.resolve_runtime.return_value = { + "runtime": "openai_compatible", "model_id": "test-model", + "api_key": "test-key", "base_url": base_url, + } + model = await LLMHelper.get_llm( + provider="opencode", model="test-model", user_agent="", + use_proxy=False, api_protocol="chat_completions", web_search_mode="disabled", + prompt_cache_key=cache_key, provider_runtime=runtime, + ) + bound = model.bind_tools([{ + "type": "function", "function": {"name": "example", "description": "测试工具", + "parameters": {"type": "object", "properties": {}}}, + }]) + assert bound.invoke("hello").content == "OK" + assert (await bound.ainvoke("again")).content == "OK" + + asyncio.run(invoke()) + + assert len(requests) == 2 + session = requests[0].headers["x-opencode-session"] + assert session + assert requests[1].headers["x-opencode-session"] == session + if cache_key: + assert session == cache_key + assert requests[0].headers["user-agent"] == "MoviePilot" + + +@pytest.mark.parametrize("base_url", [ + "https://opencode.ai/zen/go/v1", "https://OPENCODE.AI/zen/v1", +]) +def test_opencode_independent_models_have_distinct_sessions(base_url): + """没有对话标识的独立模型互不共用会话，并保留自定义 UA 和模型参数。""" + original = {"user-agent": "custom-client/1.0", "X-Test": "value"} + options = {"extra_body": {"test": True}} + headers = [LLMHelper._build_openai_prompt_cache_options( + provider="custom", base_url=base_url, use_responses_api=True, + prompt_cache_key=None, default_headers=original, model_kwargs=options, + ) for _ in range(2)] + assert headers[0][0]["x-opencode-session"] != headers[1][0]["x-opencode-session"] + assert headers[0][0]["user-agent"] == "custom-client/1.0" + assert "User-Agent" not in headers[0][0] + assert headers[0][0]["X-Test"] == "value" + assert headers[0][1] == options + assert "x-opencode-session" not in original + + +@pytest.mark.parametrize("base_url", [ + "https://opencode.ai.example/zen/go/v1", "https://proxy.example/opencode.ai", + "https://opencode.ai@proxy.example/v1", "https://[invalid", None, +]) +def test_opencode_headers_do_not_leak_to_other_hosts(base_url): + """供应商名称与路径不能代替官方主机校验，兼容端点保持原有参数。""" + headers, kwargs = LLMHelper._build_openai_prompt_cache_options( + provider="opencode", base_url=base_url, use_responses_api=False, + prompt_cache_key="private-cache-key", default_headers=None, model_kwargs={}, + ) + assert headers is None + assert kwargs == {} diff --git a/tests/test_media_classification_catalog.py b/tests/test_media_classification_catalog.py index 1aac05ca5..9db3a7063 100644 --- a/tests/test_media_classification_catalog.py +++ b/tests/test_media_classification_catalog.py @@ -98,20 +98,16 @@ def test_catalog_separates_retired_fields_from_new_rule_options() -> None: assert retired_fields[0].replacement_field == "media.countries" -def test_discover_only_builtin_sources_are_explicitly_unavailable() -> None: - """仅提供发现入口的内置来源不能被误认为可形成分类事实。""" +def test_plugin_sources_have_no_builtin_capability_declarations() -> None: + """无宿主模块的来源不得预占插件标识或预设不可用的字段能力。""" discover_sources = { - MediaSource.Bilibili.value, - MediaSource.MangoTV.value, - MediaSource.MiguVideo.value, - MediaSource.TencentVideo.value, - MediaSource.Iqiyi.value, + "bilibili", "mangguodiscover", "migu", "tencentvideodiscover", "iqiyidiscover", } - for media_source in discover_sources: - assert set(builtin_source_field_support(media_source).values()) == { - "unavailable" - } + assert discover_sources.isdisjoint(BUILTIN_CLASSIFICATION_SOURCES) + assert discover_sources.isdisjoint(source.value for source in MediaSource) + for field in get_standard_classification_fields(): + assert discover_sources.isdisjoint(field.source_support) def test_catalog_dictionary_values_match_normalized_facts() -> None: diff --git a/tests/test_media_classification_plugin_extensions.py b/tests/test_media_classification_plugin_extensions.py index 4e5672c95..11fb5249f 100644 --- a/tests/test_media_classification_plugin_extensions.py +++ b/tests/test_media_classification_plugin_extensions.py @@ -306,6 +306,61 @@ def test_registry_rejects_builtin_duplicate_and_cross_plugin_sources() -> None: assert registry.sources(OTHER_PLUGIN_ID) == [] +@pytest.mark.parametrize( + "media_source", + ["bilibili", "mangguodiscover", "migu", "tencentvideodiscover", "iqiyidiscover"], +) +def test_plugin_owned_sources_follow_registration_lifecycle( + media_source: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """无内置模块的来源由插件注册，API 和分类字段随插件启停增删。""" + from app.api.endpoints.media import _registered_media_sources + from app.application.classification.catalog import build_classification_field_catalog + from app.domain.classification.sources import BUILTIN_CLASSIFICATION_SOURCES + + registry = _registry() + monkeypatch.setattr( + "app.application.plugin.runtime.get_plugin_manager", + lambda: SimpleNamespace(get_media_sources=registry.sources), + ) + source_id = media_source + field_id = f"extensions.{source_id}.region_group" + baseline_sources = _registered_media_sources() + assert {source.media_source.value for source in baseline_sources} == set( + BUILTIN_CLASSIFICATION_SOURCES + ) + assert media_source not in {source.media_source for source in baseline_sources} + assert all( + source_id not in field.source_support + for field in build_classification_field_catalog() + ) + + registry.replace(PLUGIN_ID, [_source( + source_id=source_id, + fields=[_field("region_group", source_id=source_id)], + )]) + sources = _registered_media_sources() + assert [source for source in sources if source.media_source == media_source] == [ + MediaSourceInfo.model_validate(registry.sources()[0]) + ] + fields = {field.id: field for field in build_classification_field_catalog(registry.fields())} + assert fields[field_id].source_support == {source_id: "extension"} + media = MediaInfo( + media_source=media_source, + media_id="native-1", + type=MediaType.MOVIE, + classification_facts={field_id: "亚洲"}, + ) + assert registry.facts(media) == {source_id: {"region_group": "亚洲"}} + + registry.remove(PLUGIN_ID) + assert _registered_media_sources() == baseline_sources + assert field_id not in { + field.id for field in build_classification_field_catalog(registry.fields()) + } + + @pytest.mark.parametrize( ("field", "error_text"), [ diff --git a/tests/test_media_source_routing.py b/tests/test_media_source_routing.py index eebe519b4..aad606e71 100644 --- a/tests/test_media_source_routing.py +++ b/tests/test_media_source_routing.py @@ -26,13 +26,14 @@ def test_generic_source_id_resolves_as_fixed_enum() -> None: def test_iqiyi_source_routes_through_unified_identity() -> None: """爱奇艺探索来源应支持选择解析、身份解析和媒体键构造。""" assert parse_media_source_selection("iqiyi,iqiyidiscover") == ( - MediaSource.Iqiyi, + MediaSource("iqiyi"), MediaSource("iqiyidiscover"), ) assert resolve_media_identity( - media_source="iqiyi", + media_source="iqiyidiscover", media_id="album-1", - ) == (MediaSource.Iqiyi, "album-1") - assert build_media_key("iqiyi", "album-1") == "iqiyidiscover:album-1" + ) == (MediaSource("iqiyidiscover"), "album-1") + assert build_media_key("iqiyidiscover", "album-1") == "iqiyidiscover:album-1" + assert build_media_key("iqiyi", "album-1") == "iqiyi:album-1" def test_plugin_source_is_preserved_as_dynamic_enum() -> None: diff --git a/tests/test_release_supply_chain.py b/tests/test_release_supply_chain.py index 9cf347f09..1d19a7edc 100644 --- a/tests/test_release_supply_chain.py +++ b/tests/test_release_supply_chain.py @@ -126,6 +126,10 @@ def test_base_image_uses_refreshable_tag_and_apt_does_not_upgrade_in_place() -> assert "ARG MOVIEPILOT_PYTHON_VERSION" in free_threaded_stage assert 'uv python install --no-bin "${MOVIEPILOT_PYTHON_VERSION}t"' in free_threaded_stage assert "apt-get upgrade" not in dockerfile + assert ( + "apt-get install -y --no-install-recommends libde265-0" + in dockerfile + ) assert "\n openssl \\\n" in dockerfile assert "\n util-linux \\\n" in dockerfile diff --git a/tests/test_scheduler_lifecycle.py b/tests/test_scheduler_lifecycle.py index 693ecd678..5f5c1faae 100644 --- a/tests/test_scheduler_lifecycle.py +++ b/tests/test_scheduler_lifecycle.py @@ -849,3 +849,46 @@ def test_cancelled_cross_thread_proxy_waits_for_target_loop_cleanup( assert scheduler._registry.is_active("cancel-before-start") is False assert scheduler._registry.handles() == () assert not any("was never awaited" in str(item.message) for item in captured) + + +@pytest.mark.anyio +async def test_subscription_queue_poll_returns_while_owned_search_is_running(monkeypatch) -> None: + """长搜索不占用轮询调用，重复轮询不重入且取消必须等待同步 worker 结束。""" + from app.chain.subscribe.facade import SubscribeChain + + _patch_progress(monkeypatch) + chain = object.__new__(SubscribeChain) + started = threading.Event() + release = threading.Event() + calls = [] + + def consume(**kwargs): + """模拟慢搜索并捕获手工搜索参数，禁止真实业务调用。""" + calls.append(kwargs) + started.set() + assert release.wait(timeout=5) + + chain.resume_search_queue = consume + job_id = "subscribe_search_queue" + scheduler = _scheduler(job_id, chain.async_resume_search_queue) + scheduler._jobs[job_id]["kwargs"] = {"limit": 2, "manual_sids": (1, 2)} + scheduler._poll_subscription_search_queue() + try: + assert await asyncio.to_thread(started.wait, 2) + assert scheduler._is_job_active(job_id) + for _ in range(3): + scheduler._poll_subscription_search_queue() + assert len(calls) == 1 + assert calls[0]["limit"] == 2 + assert calls[0]["manual_sids"] == (1, 2) + handles = scheduler._registry.handles() + assert handles + stopping = asyncio.create_task(scheduler.stop_async()) + await asyncio.sleep(0) + assert not stopping.done() + assert scheduler._is_job_active(job_id) + finally: + release.set() + await asyncio.wait_for(stopping, timeout=3) + assert not scheduler._is_job_active(job_id) + assert scheduler._registry.handles() == () diff --git a/tests/test_search_media_sources.py b/tests/test_search_media_sources.py index e3de08d70..9f8ad3603 100644 --- a/tests/test_search_media_sources.py +++ b/tests/test_search_media_sources.py @@ -7,7 +7,7 @@ import pytest from app.api.endpoints import media as media_endpoint from app.api.endpoints import search as search_endpoint from app.chain.subscribe import create as subscribe_create -from app.chain.subscribe import SubscribeChain +from app.chain.subscribe.facade import SubscribeChain from app.domain.context import MediaInfo from app.schemas.types import MediaSource, MediaType from app.schemas.media import normalize_media_source @@ -21,11 +21,16 @@ def test_media_source_normalization_accepts_plugin_source() -> None: assert normalize_media_source("plugin source:invalid") is None -def test_iqiyi_media_source_aliases_are_normalized() -> None: - """爱奇艺探索来源的历史前缀和规范前缀应归一到同一媒体来源。""" - assert normalize_media_source("iqiyi") is MediaSource.Iqiyi - assert normalize_media_source("iqiyidiscover") is MediaSource.Iqiyi - assert MediaSource("iqiyi") is MediaSource.Iqiyi +@pytest.mark.parametrize("source_id", [ + "bilibili", "mangguodiscover", "migu", "tencentvideodiscover", "iqiyidiscover", + "mango_tv", "migu_video", "tencent_video", "iqiyi", +]) +def test_plugin_media_sources_preserve_exact_identifiers(source_id: str) -> None: + """插件来源不再属于内置枚举，也不替插件转换来源别名。""" + source = normalize_media_source(source_id) + assert source is MediaSource(source_id) + assert source.value == source_id + assert source not in list(MediaSource) def test_resolve_anilist_search_params_preserves_identity() -> None: diff --git a/tests/test_subscribe_search_state.py b/tests/test_subscribe_search_state.py index dbdb27fee..5e514412a 100644 --- a/tests/test_subscribe_search_state.py +++ b/tests/test_subscribe_search_state.py @@ -125,8 +125,9 @@ def test_new_subscribe_search_marks_state_after_attempt(monkeypatch) -> None: chain.search(state="N", manual=False) media_chain.recognize_media.assert_called_once() - assert len(_SubscribeOper.updates) == 1 - subscribe_id, subscription_patch = _SubscribeOper.updates[0] + assert len(_SubscribeOper.updates) == 2 + assert "last_search" in _SubscribeOper.updates[0][1].to_payload() + subscribe_id, subscription_patch = _SubscribeOper.updates[1] assert subscribe_id == 31 assert subscription_patch == SubscriptionPatch({"state": "R"}) @@ -150,6 +151,7 @@ def test_targeted_batch_searches_all_ids_without_state_scan(monkeypatch) -> None with patch.object(subscribe_search, "MediaChain", return_value=media_chain): chain = object.__new__(SubscribeChain) chain.subscription_repository = subscribe_oper + monkeypatch.setattr(chain, "_SubscribeChain__apply_subscribe_update", lambda sub, *_args, **_kwargs: sub) chain.search(sids=(31, 32), state=None, manual=False) assert [item.args for item in subscribe_oper.get.call_args_list] == [ diff --git a/tests/test_subscription_schedule.py b/tests/test_subscription_schedule.py new file mode 100644 index 000000000..80a960a88 --- /dev/null +++ b/tests/test_subscription_schedule.py @@ -0,0 +1,136 @@ +"""逐订阅搜索周期的到期、写入和数据库迁移合同。""" + +import importlib +from dataclasses import replace +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations +from pydantic import ValidationError + +from app.application.subscription.contract import SubscriptionSnapshot +from app.application.subscription.query import subscription_search_due +from app.chain.subscribe.facade import SubscribeChain +from app.db.adapters.subscription import TransactionalSubscriptionRepository +from app.db.models.subscribe import Subscribe +from app.db.session import SessionFactory, async_session_scope +from app.scheduler.catalog import _subscription_search_job_specs +from app.schemas.subscribe import Subscribe as SubscribeSchema + +NOW = datetime(2026, 9, 8, 12, tzinfo=timezone.utc) + + +@pytest.mark.parametrize("interval,elapsed,due", [ + (None, 23, False), (None, 24, True), (1, 0.99, False), (1, 1, True), + (48, 24, False), (48, 48, True), (1, -1, False), +]) +def test_search_due_uses_custom_or_system_interval(interval, elapsed, due): + """独立周期可短于或长于系统周期，并在精确到期时放行。""" + subscribe = SubscriptionSnapshot( + id=1, name="周期测试", state="R", search_interval=interval, + last_search=(NOW - timedelta(hours=elapsed)).isoformat(), + ) + assert subscription_search_due(subscribe, 24, NOW) is due + assert subscription_search_due(replace(subscribe, state="P"), 24, NOW) is due + assert not subscription_search_due(replace(subscribe, state="S"), 24, NOW) + assert not subscription_search_due(replace(subscribe, state="N"), 24, NOW) + + +def test_legacy_creation_time_and_missing_timestamp(): + """旧本地时间按本地时区解析，缺失或损坏时间允许恢复搜索。""" + recent = (NOW - timedelta(hours=1)).astimezone().strftime("%Y-%m-%d %H:%M:%S") + subscribe = SubscriptionSnapshot(id=1, name="旧订阅", state="R", date=recent) + assert not subscription_search_due(subscribe, 24, NOW) + for value in (None, "invalid"): + assert subscription_search_due(replace(subscribe, date=value), 24, NOW) + + +def test_targeted_and_manual_selection_bypasses_schedule(): + """自动调度仅选到期记录，手动及指定目标保持原有立即搜索语义。""" + recent = datetime.now(timezone.utc).isoformat() + subscribe = SubscriptionSnapshot(id=1, name="测试", state="R", last_search=recent) + chain = object.__new__(SubscribeChain) + chain.subscription_repository = SimpleNamespace( + list=Mock(return_value=[subscribe]), get=Mock(return_value=subscribe), + ) + assert chain._load_search_subscriptions(None, None, "R", scheduled_interval=24) == [] + assert chain._load_search_subscriptions(None, None, "R") == [subscribe] + assert chain._load_search_subscriptions(1, None, "R", scheduled_interval=24) == [subscribe] + assert chain._load_search_subscriptions(None, (1,), "R", scheduled_interval=24) == [subscribe] + + +def test_scheduled_scan_does_not_create_empty_batches(): + """尚未到期的订阅不产生空批次，也不触发队列消费。""" + subscribe = SubscriptionSnapshot( + id=1, name="测试", state="R", last_search=datetime.now(timezone.utc).isoformat(), + ) + chain = object.__new__(SubscribeChain) + chain.subscription_repository = SimpleNamespace(list=Mock(return_value=[subscribe])) + chain.subscription_search_repository = Mock() + progress = Mock() + assert chain.search(state="R", scheduled_interval=24, progress_callback=progress) is None + chain.subscription_search_repository.enqueue.assert_not_called() + chain.subscription_search_repository.claim_next.assert_not_called() + assert progress.call_args.kwargs["value"] == 100 + + +@pytest.mark.parametrize("interval", [0, -1, 1.5, 8761]) +def test_invalid_search_interval_is_rejected(interval): + """接口拒绝零值、负数、小数和超出范围的独立周期。""" + with pytest.raises(ValidationError): + SubscribeSchema(search_interval=interval) + + +def test_public_write_can_clear_interval_but_cannot_forge_search_time(): + """恢复系统周期必须保留 null，客户端不能覆盖内部搜索时钟。""" + for value in (None, ""): + payload = SubscribeSchema(search_interval=value, last_search=NOW.isoformat()) + assert payload.to_public_write_payload(exclude_unset=True) == {"search_interval": None} + assert SubscribeSchema().to_public_write_payload(exclude_unset=True) == {} + + +def test_schedule_persists_across_repository_recreation(db): + """周期和搜索时间在 Session 关闭及仓储重建后仍可用于到期判断。""" + row = db.add(Subscribe( + name="持久周期", state="R", search_interval=48, last_search=NOW.isoformat(), + )) + for _ in range(2): + repository = TransactionalSubscriptionRepository( + sync_session=SessionFactory, async_session=async_session_scope, + ) + snapshot = repository.get(row.id) + assert snapshot.search_interval == 48 + assert snapshot.last_search == NOW.isoformat() + assert not subscription_search_due(snapshot, 24, NOW + timedelta(hours=24)) + + +def test_scheduler_passes_system_interval_only_to_periodic_search(): + """调度目录传递可热重载的系统间隔，新增订阅保持首次立即搜索。""" + specs = {spec.job_id: spec for spec in _subscription_search_job_specs(Mock(), 12)} + assert specs["subscribe_search"].kwargs == {"state": "R", "scheduled_interval": 12} + assert specs["new_subscribe_search"].kwargs == {"state": "N"} + + +def test_schedule_migration_preserves_existing_rows_and_is_reversible(monkeypatch): + """旧数据默认跟随系统，重复升级与回滚不损坏已有订阅。""" + migration = importlib.import_module("database.versions.f8a2c6e9b1d4_3_0_30") + engine = sa.create_engine("sqlite://") + with engine.begin() as connection: + for table in ("subscribe", "subscribehistory"): + connection.execute(sa.text(f"CREATE TABLE {table} (id INTEGER PRIMARY KEY, name TEXT)")) + connection.execute(sa.text(f"INSERT INTO {table} VALUES (1, '旧订阅')")) + monkeypatch.setattr(migration, "op", Operations(MigrationContext.configure(connection))) + migration.upgrade() + migration.upgrade() + for table in ("subscribe", "subscribehistory"): + row = connection.execute(sa.text(f"SELECT name, search_interval FROM {table}")).one() + assert row == ("旧订阅", None) + assert connection.execute(sa.text("SELECT last_search FROM subscribe")).scalar_one() is None + migration.downgrade() + migration.upgrade() + assert connection.execute(sa.text("SELECT name FROM subscribe")).scalar_one() == "旧订阅" + engine.dispose() diff --git a/tests/test_subscription_search_cursor_migration.py b/tests/test_subscription_search_cursor_migration.py new file mode 100644 index 000000000..108b748db --- /dev/null +++ b/tests/test_subscription_search_cursor_migration.py @@ -0,0 +1,43 @@ +"""搜索站点游标与旧排期升级回归。""" + +import importlib +from datetime import datetime, timedelta, timezone + +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations + + +def test_search_cursor_migration_preserves_cooldown_and_running_tasks(monkeypatch): + """只提前未开始的自动排期，冷却、停止、新增保护期和在途任务保持原状。""" + migration = importlib.import_module("database.versions.a9c3e7f1b5d8_3_0_31") + engine = sa.create_engine("sqlite://") + metadata = sa.MetaData() + tasks = sa.Table( + "subscriptionsearchtask", metadata, + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("source", sa.String()), sa.Column("state", sa.String()), + sa.Column("phase", sa.String()), sa.Column("attempt_count", sa.Integer()), + sa.Column("cancel_requested", sa.Integer()), sa.Column("available_at", sa.String()), + ) + future = (datetime.now(timezone.utc) + timedelta(days=2)).isoformat(timespec="seconds") + with engine.begin() as connection: + metadata.create_all(connection) + rows = [dict(id=index, source="fallback", state="queued", phase="queued", attempt_count=0, + cancel_requested=0, available_at=future) for index in range(5)] + rows[1].update(phase="waiting_site_budget", attempt_count=1) + rows[2].update(cancel_requested=1) + rows[3].update(source="new", phase="scheduled") + rows[4].update(state="running", attempt_count=1) + connection.execute(tasks.insert(), rows) + monkeypatch.setattr(migration, "op", Operations(MigrationContext.configure(connection))) + migration.upgrade() + migration.upgrade() + stored = connection.execute(sa.select(tasks.c.id, tasks.c.available_at).order_by(tasks.c.id)).all() + assert stored[0].available_at < future + assert all(row.available_at == future for row in stored[1:]) + assert "pending_site_ids" in {col["name"] for col in sa.inspect(connection).get_columns(tasks.name)} + migration.downgrade() + assert "pending_site_ids" not in {col["name"] for col in sa.inspect(connection).get_columns(tasks.name)} + assert connection.scalar(sa.select(sa.func.count()).select_from(tasks)) == 5 + engine.dispose() diff --git a/tests/test_subscription_search_governance.py b/tests/test_subscription_search_governance.py index 525fb7737..f7bfa3edf 100644 --- a/tests/test_subscription_search_governance.py +++ b/tests/test_subscription_search_governance.py @@ -100,25 +100,15 @@ def _make_tasks_ready(monkeypatch) -> None: ) -def test_fallback_task_schedule_staggers_each_subscription(monkeypatch): - """兜底批次首条抖动后，每条后续订阅都按独立随机间隔到期。""" +def test_fallback_task_schedule_jitters_once_for_large_batch(monkeypatch): + """千条订阅共享一次启动抖动，不能再累计成数日空等。""" now = datetime(2026, 9, 3, 1, 2, 3, tzinfo=timezone.utc) - delays = iter((12, 60, 300)) - monkeypatch.setattr( - "app.chain.subscribe.search.random.randint", - lambda _low, _high: next(delays), - ) - - schedule = _search_task_available_at( - "fallback", - (1, 2, 3), - now=now, - ) - - available = [datetime.fromisoformat(schedule[subscribe_id]) for subscribe_id in (1, 2, 3)] - assert (available[0] - now).total_seconds() == 12 - assert (available[1] - available[0]).total_seconds() == 60 - assert (available[2] - available[1]).total_seconds() == 300 + jitter = Mock(return_value=12) + monkeypatch.setattr("app.chain.subscribe.search.random.randint", jitter) + schedule = _search_task_available_at("fallback", tuple(range(1000)), now=now) + assert len(schedule) == 1000 + assert set(schedule.values()) == {(now + timedelta(seconds=12)).isoformat(timespec="seconds")} + jitter.assert_called_once_with(0, 60) def test_inline_fallback_search_preserves_site_pressure_stagger(tmp_path, monkeypatch): @@ -349,7 +339,8 @@ 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 == "站点暂时忙，系统会自动继续搜索" + assert task.last_error == "等待站点" + assert task.pending_site_ids == [31] assert chain.subscription_search_repository.claim_next(owner="worker-after-retry") is None diff --git a/tests/test_subscription_search_queue.py b/tests/test_subscription_search_queue.py index f2f29f74e..310835306 100644 --- a/tests/test_subscription_search_queue.py +++ b/tests/test_subscription_search_queue.py @@ -362,3 +362,54 @@ def test_search_queue_keeps_manual_work_ahead_of_aged_fallback(tmp_path): assert claimed.subscription_id == 9 assert claimed.source == "manual" + + +def test_retry_sites_survive_reopen_and_admission_wait_without_stale_error(tmp_path): + """站点游标跨重建仓储和准入等待保留，真正恢复时清除上轮等待提示。""" + repository, engine = _repository(tmp_path) + repository.enqueue(subscription_ids=(701,), source="fallback", priority=10) + first = repository.claim_next(owner="before-restart") + ready_at = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat(timespec="seconds") + assert repository.defer_task( + task_id=first.task_id, lease_token=first.lease_token, available_at=ready_at, + message="站点冷却中", pending_site_ids=(8, 9), + ) + reopened = TransactionalSubscriptionSearchRepository(sessionmaker(bind=engine)) + resumed = reopened.claim_next(owner="after-restart") + assert resumed.task_id == first.task_id + assert resumed.pending_site_ids == (8, 9) + assert resumed.state == "running" + assert resumed.last_error is None + assert reopened.defer_task( + task_id=resumed.task_id, lease_token=resumed.lease_token, available_at=ready_at, + phase="waiting_subscription", message="等待任务", + ) + continued = reopened.claim_next(owner="after-match") + assert continued.pending_site_ids == (8, 9) + assert continued.last_error is None + assert reopened.defer_task( + task_id=continued.task_id, lease_token=first.lease_token, available_at=ready_at, + pending_site_ids=(99,), + ) is False + assert reopened.finish_task(task_id=continued.task_id, lease_token=continued.lease_token, state="completed") + engine.dispose() + + +def test_manual_search_restarts_full_scope_but_automatic_merge_keeps_cursor(tmp_path): + """自动调度合并保留进度，用户主动重搜可包含新配置的站点。""" + repository, engine = _repository(tmp_path) + repository.enqueue(subscription_ids=(702,), source="fallback", priority=10) + first = repository.claim_next(owner="worker") + ready_at = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat(timespec="seconds") + assert repository.defer_task( + task_id=first.task_id, lease_token=first.lease_token, available_at=ready_at, + pending_site_ids=(9,), + ) + repository.enqueue(subscription_ids=(702,), source="fallback", priority=10) + with Session(engine) as session: + assert session.scalar(select(SubscriptionSearchTask.pending_site_ids)) == [9] + repository.enqueue(subscription_ids=(702,), source="manual", priority=100) + resumed = repository.claim_next(owner="manual") + assert resumed.pending_site_ids is None + assert resumed.source == "manual" + engine.dispose() diff --git a/tests/test_subscription_site_budget.py b/tests/test_subscription_site_budget.py index 4a09db9a1..4953a757f 100644 --- a/tests/test_subscription_site_budget.py +++ b/tests/test_subscription_site_budget.py @@ -549,3 +549,48 @@ def test_search_provider_logs_site_budget_release_failure(monkeypatch): assert result == ["torrent"] assert metrics.snapshot().release_failure_count == 1 assert errors == ["订阅站点预算释放失败: site_id=14 site=Stale"] + + +def test_retry_provider_only_searches_pending_enabled_sites(monkeypatch): + """恢复时不重搜成功站点和插件源，也不重新启用已经移除的站点。""" + from unittest.mock import Mock + + from app.chain.search.provider import SearchProviderOwner + + chain = object.__new__(SearchChain) + chain.configure_subscription_site_budget(SubscriptionSiteBudget( + repository=_WaitingRepository(), owner="retry", cancelled=lambda: False, + stop_state=ProcessStopState(), pending_site_ids=(2, 3), + )) + chain._sync_indexers = lambda _sites: [{"id": 1}, {"id": 2}] + chain._torrent_type = lambda *_args: None + chain._torrent_keyword = lambda *_args: "movie" + chain._build_search_pages = lambda _page: [0] + chain.search_plugin_torrents = Mock(return_value=[]) + captured = [] + + def collect(**kwargs): + """只观察 provider 选择，禁止真实站点和插件调用。""" + captured.extend(site["id"] for site in kwargs["indexer_sites"]) + return {} + + monkeypatch.setattr(SearchProviderOwner, "_collect_sync_site_results", lambda _self, **kwargs: collect(**kwargs)) + monkeypatch.setattr("app.chain.search.provider.ProgressHelper", Mock()) + SearchProviderOwner._search_all_sites(chain, keyword="movie", sites=[1, 2]) + assert captured == [2] + chain.search_plugin_torrents.assert_not_called() + + +def test_deferral_reports_cooldown_separately_from_busy(): + """冷却提示不伪装成站点占用，并保留所有未完成站点。""" + from app.application.subscription.execution import raise_subscription_site_budget_deferral + from app.application.subscription.sitebudget import SubscriptionSearchDeferred, SubscriptionSiteBudgetDeferral + + deferred_sites = ( + SubscriptionSiteBudgetDeferral(site_id=1, retry_at="2026-09-08T10:00:00+00:00", wait_reason="cooldown"), + SubscriptionSiteBudgetDeferral(site_id=2, retry_at="2026-09-08T11:00:00+00:00", wait_reason="cooldown"), + ) + with pytest.raises(SubscriptionSearchDeferred, match="站点冷却中") as caught: + raise_subscription_site_budget_deferral(deferred_sites, None) + assert caught.value.site_ids == (1, 2) + assert caught.value.retry_at == deferred_sites[0].retry_at