mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-09-08 17:08:35 +08:00
feat(subscription): add search_interval and last_search fields for subscription management
This commit is contained in:
jxxghp
@@ -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.",
|
||||
|
||||
@@ -2862,6 +2862,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 +3069,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": [
|
||||
{
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -28,7 +28,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,
|
||||
@@ -160,11 +160,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 +173,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase):
|
||||
state=state,
|
||||
manual=manual,
|
||||
progress_callback=progress_callback,
|
||||
scheduled_interval=scheduled_interval,
|
||||
)
|
||||
|
||||
def _execute_search(
|
||||
@@ -181,6 +183,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 +205,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase):
|
||||
state=state,
|
||||
manual=manual,
|
||||
progress_callback=progress_callback,
|
||||
scheduled_interval=scheduled_interval,
|
||||
)
|
||||
self._execute_inline_search(
|
||||
sid=sid,
|
||||
@@ -209,6 +213,7 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase):
|
||||
state=state,
|
||||
manual=manual,
|
||||
progress_callback=progress_callback,
|
||||
scheduled_interval=scheduled_interval,
|
||||
)
|
||||
return None
|
||||
|
||||
@@ -219,6 +224,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 +234,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 +361,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,
|
||||
@@ -573,6 +590,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 +599,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 +704,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),
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
# 创建时间
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -103,6 +103,7 @@
|
||||
}
|
||||
},
|
||||
"messages": {
|
||||
"暂无到期订阅，无需搜索": "暫無到期訂閱，無需搜尋",
|
||||
"调用工具失败": "調用工具失敗",
|
||||
"媒体来源和媒体 ID 必须同时提供": "媒體來源和媒體 ID 必須同時提供",
|
||||
"media_source 和 media_id 必须同时提供": "media_source 和 media_id 必須同時提供",
|
||||
|
||||
@@ -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",
|
||||
@@ -168,7 +171,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 +301,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"},
|
||||
)
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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")
|
||||
@@ -422,6 +422,17 @@ 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 刷新不受周期过滤影响；
|
||||
手动主动搜索会更新最近搜索时间。电影、电视剧和音乐均支持独立周期。
|
||||
|
||||
### 插件补充接口
|
||||
|
||||
**GET** `/api/v1/plugin/history/{plugin_id}`
|
||||
|
||||
@@ -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.
|
||||
|
||||
File diff suppressed because one or more lines are too long
@@ -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")]
|
||||
|
||||
@@ -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] == [
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user