fix(subscribe): stop failing sites from stalling subscription downloads

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
InfinityPacer
2026-11-03 02:53:36 +08:00
committed by InfinityPacer
parent 9eaaf5f715
commit 5de950fb7c
24 changed files with 669 additions and 258 deletions

View File

@@ -85,6 +85,8 @@ def restore_collection(values: dict[str, Any]) -> SearchCollection:
params = dict(data)
# 旧检查点未记录作品匹配前史;已有进度时不能断言此前从未命中,只应用页数上限。
params.setdefault("matched_work", bool(params.get("next_page")))
# 旧检查点的长等待计数已废弃,站点要等太久时改为本轮直接跳过。
params.pop("budget_deferrals", None)
for name in ("seen", "closed"):
params[name] = set(params.get(name, []))
sources[key] = SearchSourceCursor(**params)

View File

@@ -181,10 +181,11 @@ def raise_subscription_site_budget_deferral(
"""
if not deferrals or (execution_context and execution_context.download_started and not execution_context.incremental_search):
return
retry_at = min(deferrals, key=lambda item: item.retry_at).retry_at
earliest = min(deferrals, key=lambda item: item.retry_at)
site_ids = tuple(dict.fromkeys(item.site_id for item in deferrals))
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)
# 以最先恢复的站点说明等待原因,和恢复时间保持一致。
raise SubscriptionSearchDeferred(retry_at=earliest.retry_at, site_ids=site_ids,
wait_reason=earliest.wait_reason or "busy")
def handle_subscription_search_deferred(

View File

@@ -41,12 +41,15 @@ class SubscriptionSearchDeferred(RuntimeError):
def __init__(
self, *, retry_at: str, site_ids: tuple[int, ...], wait_reason: Optional[str] = None,
message: Optional[str] = None,
) -> None:
"""
保存队列恢复所需的时间和站点,避免把临时等待写成错误。
:param message: 给用户看的等待说明,例如在等哪个站、为什么、几点继续;短暂停顿不传
"""
messages = {"slice": "分页进度已保存,等待下一时间片", "cooldown": "站点冷却中"}
super().__init__(messages.get(wait_reason or "", "等待站点"))
messages = {"rate_limit": "站点设置的访问次数已用完,到时自动继续", "busy": "站点正在被其他订阅搜索,稍等自动继续"}
super().__init__(message or messages.get(wait_reason or "", "正在搜索"))
self.retry_at = retry_at
self.site_ids = site_ids
self.wait_reason = wait_reason
@@ -63,15 +66,28 @@ class SubscriptionSiteBudgetUnavailable(RuntimeError):
site_id: int,
retry_at: str,
wait_reason: Optional[str] = None,
consecutive_failures: int = 0,
last_error: Optional[str] = None,
) -> None:
"""
保存站点和下一次可尝试时间,供订阅队列恢复。
:param wait_reason: busy 其他订阅正在用该站,cooldown 两次请求之间的间隔,
rate_limit 站点设置的访问次数已用完,failing 站点最近搜索出错、暂停访问中
:param consecutive_failures: 该站点连续失败次数,用于判断是否需要提醒用户检查站点
:param last_error: 该站点最近一次失败的原因,给用户看
"""
message = "站点正在处理其他搜索,稍后会自动继续" if wait_reason == "busy" else "站点暂时不可用,稍后会自动重试"
super().__init__(message)
messages = {
"busy": "站点正在被其他订阅搜索",
"rate_limit": "站点设置的访问次数已用完",
"failing": f"站点最近搜索出错:{last_error}" if last_error else "站点最近搜索出错",
}
super().__init__(messages.get(wait_reason or "", "两次搜索之间需要间隔一会儿"))
self.site_id = site_id
self.retry_at = retry_at
self.wait_reason = wait_reason
self.consecutive_failures = consecutive_failures
self.last_error = last_error
@dataclass(frozen=True, slots=True)
@@ -84,7 +100,8 @@ class SiteBudgetClaim:
:param retry_at: 该站点最早可再次尝试的时间文本
:param consecutive_failures: 该站点连续失败次数,用于计算后续冷却
:param lease_token: 当前认领者的租约令牌;未取得租约时为 None
:param wait_reason: 等待原因,例如在途占用、冷却或时间片耗尽
:param wait_reason: 等待原因,取值见 SubscriptionSiteBudgetUnavailable
:param last_error: 该站点最近一次失败的原因;从未失败或已恢复时为 None
"""
site_id: int
@@ -93,6 +110,7 @@ class SiteBudgetClaim:
consecutive_failures: int
lease_token: Optional[str] = None
wait_reason: Optional[str] = None
last_error: Optional[str] = None
@dataclass(frozen=True, slots=True)
@@ -301,6 +319,8 @@ class SubscriptionSiteBudget:
site_id=site_id,
retry_at=claim.retry_at,
wait_reason=claim.wait_reason,
consecutive_failures=claim.consecutive_failures,
last_error=claim.last_error,
)
def _report_phase(self, phase: str, site_id: Optional[int]) -> None:

View File

@@ -25,6 +25,7 @@ class SubscriptionExecutionStatus:
:param next_run_at: 下一次允许执行的时间文本;没有计划时为 None
:param error: 可展示的错误摘要;无错误时为 None
:param can_cancel: 当前状态是否支持用户取消
:param submitted_count: 本次任务已添加的下载数量,用于区分「已添加下载」与「暂无资源」;旧活动为 0
"""
state: str
@@ -37,6 +38,7 @@ class SubscriptionExecutionStatus:
next_run_at: Optional[str] = None
error: Optional[str] = None
can_cancel: bool = False
submitted_count: int = 0
@dataclass(frozen=True, slots=True)
@@ -249,6 +251,7 @@ class SubscriptionExecutionStatusService:
updated_at=task.updated_at,
error=cls._safe_error(task.last_error),
can_cancel=state in cls._ACTIVE_STATES,
submitted_count=task.download_counts.get("submitted", 0),
)
@classmethod

View File

@@ -991,12 +991,14 @@ class TorrentHelper:
torrent_meta: MetaBase,
) -> Tuple[bool, str]:
"""
判断媒体 ID 冲突时是否存在足以覆盖候选识别的同作品证据。
判断无年份别名资源能否按订阅目标处理,只有明确属于另一部作品时才拒绝。
TMDB 的电影与电视剧数字 ID 不可跨类型比较;类型冲突时仍需资源精确年份、
其他共同标识或一致的作品元数据,不能仅凭相同 TMDB ID 覆盖候选。
同名撞车偶尔下错的代价远低于资源一直下不下来:原名或语言不同、候选条目没有首播日期
(多为未播出或残缺的重复条目)都不足以否定目标,按目标处理。唯一拒绝的情况是候选是另一部
已播出作品,且首播年份与目标及其各季年份都对不上,例如 2024 年韩剧的英文别名撞上 2008 年同名美剧。
TMDB 的电影与电视剧数字 ID 不可跨类型比较。
:return: 是否有同作品证据,以及用于诊断的证据或拒绝原因
:return: 是否按目标处理,以及给用户看的依据或拒绝原因
"""
target_identity = resolve_media_identity(media=target_mediainfo)
candidate_identity = resolve_media_identity(media=candidate_mediainfo)
@@ -1004,60 +1006,35 @@ class TorrentHelper:
if all(target_identity) and target_identity == candidate_identity and (
same_type or target_identity[0] != MediaSource.TMDB
):
return True, f"媒体身份 {target_identity[0]}:{target_identity[1]}"
return True, "识别结果就是订阅的作品"
candidate_year = str(candidate_mediainfo.year or "")[:4]
if not candidate_year:
return True, f"按名字查到的《{candidate_mediainfo.title}》没有首播日期,不能说明是另一部作品"
torrent_year = str(getattr(torrent_meta, "year", None) or "")[:4]
target_years = {
str(year)[:4]
for year in (
[getattr(target_mediainfo, "year", None)]
+ list((getattr(target_mediainfo, "season_years", None) or {}).values())
)
for year in [target_mediainfo.year, *(target_mediainfo.season_years or {}).values()]
if year
}
if torrent_year and torrent_year in target_years:
return True, f"资源年份 {torrent_year}"
torrent_year = str(torrent_meta.year or "")[:4]
if candidate_year in target_years or (torrent_year and torrent_year in target_years):
return True, "年份与订阅的作品一致"
for field in ("tmdb_id", "imdb_id", "tvdb_id", "douban_id", "bangumi_id"):
if field == "tmdb_id" and not same_type:
continue
target_id = getattr(target_mediainfo, field, None)
candidate_id = getattr(candidate_mediainfo, field, None)
target_id = getattr(target_mediainfo, field)
candidate_id = getattr(candidate_mediainfo, field)
if target_id and candidate_id and str(target_id) == str(candidate_id):
return True, f"共同 {field}"
return True, "与订阅的作品有相同的媒体编号"
target_year = str(getattr(target_mediainfo, "year", None) or "")[:4]
candidate_year = str(getattr(candidate_mediainfo, "year", None) or "")[:4]
if target_year and candidate_year and target_year != candidate_year:
return False, f"首播年份冲突 {candidate_year} != {target_year}"
target_original_title = text_tools.normalize_upper(
getattr(target_mediainfo, "original_title", None) or ""
if not target_years:
return True, "订阅的作品没有年份可以比较"
return False, (
f"按名字查到的是另一部作品《{candidate_mediainfo.title}》({candidate_year}),"
f"和订阅的《{target_mediainfo.title}》({target_mediainfo.year})首播年份不同"
)
candidate_original_title = text_tools.normalize_upper(
getattr(candidate_mediainfo, "original_title", None) or ""
)
if (
target_original_title
and candidate_original_title
and target_original_title != candidate_original_title
):
return False, "原始标题冲突"
target_language = str(
getattr(target_mediainfo, "original_language", None) or ""
).casefold()
candidate_language = str(
getattr(candidate_mediainfo, "original_language", None) or ""
).casefold()
if target_language and candidate_language and target_language != candidate_language:
return False, f"原始语言冲突 {candidate_language} != {target_language}"
if target_year and candidate_year:
return True, f"共同首播年份 {target_year}"
if target_original_title and candidate_original_title:
return True, "共同原始标题"
return False, "资源无年份,候选与目标也没有可核验的共同元数据"
@staticmethod
def filter_torrent(torrent_info: TorrentInfo,

View File

@@ -2,6 +2,7 @@
站点与插件资源搜索的并发分发、预算控制与结果归集。
"""
import threading
import time
from concurrent.futures import FIRST_COMPLETED, Future, wait
from contextlib import aclosing
@@ -42,6 +43,37 @@ from app.schemas.types import MediaType, ProgressKey, SystemConfigKey
SiteIndexer = Dict[str, Any]
SyncPending = dict[Future[Any], tuple[SiteIndexer, int, int]]
# 站点连续搜索失败达到该次数后,订阅搜索跳过它时在日志里提醒检查站点。
SITE_FAILING_NOTICE_FAILURES = 3
# 同一站点提醒的间隔,单位为秒;跳过发生在每个订阅的每一轮,不能每次都写警告。
SITE_FAILING_NOTICE_SECONDS = 24 * 3600
_site_failing_noticed_at: Dict[int, float] = {}
_site_failing_notice_lock = threading.Lock()
def log_site_skipped(site: SiteIndexer, error: SubscriptionSiteBudgetUnavailable, *, subject: str = "") -> None:
"""
订阅搜索跳过出错暂停中的站点时写日志;连续失败较多时每天给一次带处理建议的警告。
只写日志不发通知,提醒时间只保存在进程内,重启后会再警告一次。
:param subject: 日志开头的订阅名称,没有时省略
"""
name = site.get("name") or f"站点 {error.site_id}"
prefix = f"{subject} " if subject else ""
logger.info(f"{prefix}这一轮先跳过 {name}:{error}")
if error.consecutive_failures < SITE_FAILING_NOTICE_FAILURES:
return
now = time.time()
with _site_failing_notice_lock:
if now - _site_failing_noticed_at.get(error.site_id, 0.0) < SITE_FAILING_NOTICE_SECONDS:
return
_site_failing_noticed_at[error.site_id] = now
reason = f"最近一次错误:{error.last_error}。" if error.last_error else ""
logger.warning(f"站点 {name} 最近连续 {error.consecutive_failures} 次搜索都失败了,订阅搜索会先跳过它。{reason}"
"请检查站点能否正常访问(Cookie、登录信息或网络),不再使用可以在站点管理里停用")
def _site_keyword(site: SiteIndexer, keyword: str, mediainfo: Optional[MediaInfo]) -> str:
"""
英文站点在本轮直接使用可搜索的媒体名称,避免中文站点命中后提前停止导致漏搜。
@@ -390,6 +422,12 @@ class _SearchProviderSyncOwner(_SearchOwnerBase):
try:
claim = budget.acquire(site_id)
except SubscriptionSiteBudgetUnavailable as error:
if error.wait_reason == "failing":
# 出错暂停中的站点这一轮直接不搜,用其他站点的结果继续,不让整个订阅等它。
if activity:
activity.site(error.site_id, name=str(site.get("name") or ""), status="failed")
log_site_skipped(site, error)
return []
if activity:
activity.site(error.site_id, name=str(site.get("name") or ""), status="waiting")
self.record_subscription_site_budget_deferred(

View File

@@ -173,22 +173,26 @@ def _same_work_matched(
cache: DisambiguationCache,
) -> bool:
"""
确认无年份别名候选与目标媒体属于同一作品。
无年份资源只靠别名对上订阅时,按名字再识别一次;只有明确属于另一部作品才跳过。
识别不到不算反证:宁可偶尔下错同名作品,也不能让资源一直下不下来。
"""
key = _disambiguation_key(torrent_meta)
if key not in cache:
cache[key] = MediaChain().recognize_by_meta(torrent_meta, obtain_images=False)
candidate = cache[key]
if not candidate:
logger.info(f"{torrent.site_name} - {torrent.title} 仅通过无年份别名命中且候选媒体身份无法确认,已跳过")
return False
logger.debug(f"{torrent.site_name} - {torrent.title} 资源名没有年份,按名字也识别不到作品,按订阅的作品处理")
return True
matched, evidence = TorrentHelper.match_same_work_evidence(
target_mediainfo=mediainfo,
candidate_mediainfo=candidate,
torrent_meta=torrent_meta,
)
if not matched:
logger.info(f"{torrent.site_name} - {torrent.title} 无年份同名候选未通过消歧:{evidence}")
if matched:
logger.debug(f"{torrent.site_name} - {torrent.title} 资源名没有年份,{evidence},按订阅的作品处理")
else:
logger.info(f"跳过 {torrent.title}({torrent.site_name}):资源名没有年份,{evidence}")
return matched

View File

@@ -35,6 +35,7 @@ from app.chain.search.provider import (
_site_rate_limit_backoff,
_site_request_interval,
_wait_for_site_request,
log_site_skipped,
)
from app.chain.search.result import DisambiguationCache, SearchResultOwner, _resource_targets, stored_meta
from app.domain.context import Context, MediaInfo, TorrentInfo
@@ -52,11 +53,26 @@ RESTART_AFTER_SECONDS = 36 * 3600
_CHECKPOINT_FORMAT = 3
# 重取临时下载票据所在原页连续失败达到该次数后放弃该候选,避免一个候选卡住整个订阅。
RECOVERY_ATTEMPTS = 3
# 单个来源本轮因站点预算长冷却被推迟达到该次数后结算该来源:检查点随本轮结束删除,下一轮重新尝试,
# 既打破"全部来源结束才提交"的屏障死锁,也不永久拉黑偶尔不可用的站点。
BUDGET_DEFERRAL_LIMIT = 3
# 只累计不短于该秒数的冷却推迟;请求间隔与并发租约的秒级等待是健康站点噪声,不计入。
BUDGET_DEFERRAL_SECONDS = 600.0
# 站点要等超过该秒数才能再请求时,本轮不再等它,先用其他站点的结果提交,下一轮重新带上;
# 出错暂停中的站点不论要等多久都直接跳过。请求间隔、其他订阅占用这类短暂停顿照常等待。
LONG_WAIT_SECONDS = 600.0
# 等待超过该秒数才写入常规日志;更短的停顿是正常搜索节奏,只写调试日志。
VISIBLE_WAIT_SECONDS = 120.0
# 站点出错暂停或要等太久时,来源在本轮记为这两类结果并跳过;下一轮重新搜索。
SKIPPED_THIS_ROUND = frozenset({"unavailable", "deferred"})
# 等待原因给用户看的说明,接在站点名后面。
_WAIT_REASON_TEXT = {
"rate_limit": "设置的访问次数已用完",
"busy": "正在被其他订阅搜索",
"cooldown": "两次搜索之间需要间隔一会儿",
}
def _local_time(value: str) -> str:
"""
把检查点里的 UTC 时间转成用户所在时区的「时:分」,用于日志和状态说明。
"""
return datetime.fromisoformat(value).astimezone().strftime("%H:%M")
def search_targets(plan: MediaSearchPlan) -> set[str]:
@@ -87,12 +103,14 @@ class ScanPage:
:param observation: 本次站点请求的成功、失败和分页观察事实
:param retry_at: 因预算或流控延迟时允许恢复的 UTC 时间字符串,None 表示未延迟
:param wait_reason: 需要暂缓本页请求的原因,None 表示未延迟
:param unavailable: 站点预算拒绝本次请求的原因与站点失败情况,只在预算拒绝时存在
"""
torrents: list[TorrentInfo]
observation: SiteSearchObservation
retry_at: Optional[str] = None
wait_reason: Optional[str] = None
unavailable: Optional[SubscriptionSiteBudgetUnavailable] = None
class SearchSources:
@@ -168,8 +186,10 @@ class SearchSources:
minimum_interval=max(AUTOMATIC_REQUEST_INTERVAL, _site_request_interval(site)))
except SubscriptionSiteBudgetUnavailable as error:
if self.activity:
self.activity.site(error.site_id, name=str(site.get("name") or ""), status="waiting")
return ScanPage([], SiteSearchObservation(error=str(error)), error.retry_at, error.wait_reason)
# 出错暂停的站本轮直接跳过,活动里记为失败,不显示成「等待继续」。
self.activity.site(error.site_id, name=str(site.get("name") or ""),
status="failed" if error.wait_reason == "failing" else "waiting")
return ScanPage([], SiteSearchObservation(error=str(error)), error.retry_at, error.wait_reason, error)
with capture_site_search_observation() as observation:
deferral_seconds = 0.0
try:
@@ -182,7 +202,7 @@ class SearchSources:
# 本地流控未发出请求,只按实际受限规则退避,不能把短间隔扩大为小时计数窗口。
deferral_seconds = _site_rate_limit_backoff(site, observation.error) or float(AUTOMATIC_REQUEST_INTERVAL)
retry_at = (datetime.now(timezone.utc) + timedelta(seconds=deferral_seconds)).isoformat(timespec="seconds")
return ScanPage(items, observation, retry_at, "cooldown")
return ScanPage(items, observation, retry_at, "rate_limit")
if claim and budget:
budget.record_request(claim.site_id, len(items))
if observation.outcome == "success":
@@ -213,11 +233,14 @@ class SearchScan:
target_scope: Optional[tuple[int, int]] = None,
page_limit: Optional[int] = None,
prefer_complete: bool = False,
complete_scan: Optional[bool] = None,
) -> None:
"""
创建或恢复一轮搜索,候选与匹配前史由检查点一起持有。
:param prefer_complete: 提交分集前先准备覆盖全部目标的候选;由订阅当前整包优先策略决定
:param complete_scan: 是否翻完所有页再择优(用户选了完整搜索或洗版);否则各站第一页到齐就先提交,
凑不齐的目标再继续翻页。未指定时跟随 full
"""
if not isinstance(plan.mediainfo, MediaInfo):
raise ValueError("分页影视搜索不适用于音乐")
@@ -225,11 +248,12 @@ class SearchScan:
self.owner, self.plan, self.execution = cast(_SearchOwnerBase, owner), plan, execution
self.repository, self.snapshot = repository, snapshot
targets = search_targets(plan)
# 整季缺失即使展开为逐集结算目标,也必须保留整季收集屏障。
# 整季缺失即使展开为逐集结算目标,各站第一页也要到齐才提交,避免快的站散集抢先、慢的站整季包没机会参与。
self.full = (full or any(target.endswith(":season") for target in targets)
or (self.media.type == MediaType.TV and any(
not missing.episodes for seasons in (plan.no_exists or {}).values()
for missing in seasons.values())))
self.complete_scan = full if complete_scan is None else complete_scan
self.target_scope = target_scope
self.page_limit = page_limit
self.prefer_complete = prefer_complete
@@ -249,6 +273,12 @@ class SearchScan:
self.recovery_failures: dict[str, int] = {}
# 本轮已提交下载的目标,仅用于进度展示。
self.submitted: set[str] = set()
# 本进程已写过「本轮跳过」日志的站点,同一站点多个关键词只提示一次。
self.skipped_sites: set[str] = set()
# 各来源第一页到齐后是否已先按现有资源提交过一次;随检查点保存,续跑不会把深页资源当成第一页提前提交。
self.first_pass_done = False
# 本轮因要等很久而先跳过的来源及其可再搜时间和原因;整轮什么都没交时改为等它们,不判搜索失败。
self.long_waits: dict[str, tuple[str, str]] = {}
self.sources = SearchSources(owner, plan)
self._media_identity = encode_search_state({
"source": self.media.media_source, "id": self.media.media_id, "type": self.media.type,
@@ -326,6 +356,9 @@ class SearchScan:
self.retry_reasons = data.get("retry_reasons", {})
self.recovery_failures = data.get("recovery_failures", {})
self.submitted = set(data.get("submitted", []))
# 旧检查点没有这两项:按未先提交处理,升级后卡在整季屏障的任务能先交第一页已有的资源。
self.first_pass_done = bool(data.get("first_pass_done", False))
self.long_waits = {key: (value[0], value[1]) for key, value in data.get("long_waits", {}).items()}
same_media = data["media_identity"] == self._media_identity
if not same_media:
self.restart = True
@@ -378,25 +411,46 @@ class SearchScan:
"keywords": self.sources.keywords, "keyword_index": self.sources.keyword_index,
"retry": self.retry, "retry_reasons": self.retry_reasons,
"recovery_failures": {key: count for key, count in self.recovery_failures.items() if key in candidates},
"submitted": sorted(self.submitted), "saved_at": time.time()}
"submitted": sorted(self.submitted), "first_pass_done": self.first_pass_done,
"long_waits": self.long_waits, "saved_at": time.time()}
def progress_text(self) -> str:
"""
写入日志的进度摘要:提交、仍缺集和本轮结束分开表达,不把分页结束叫作补齐。
写入日志的进度摘要:已添加下载、还缺哪些和本轮是否结束分开说,不把翻页结束说成补齐。
"""
episodes = [int(target.split(":")[1]) for target in self.collection.remaining
if ":" in target and not target.endswith(":season")]
remaining = format_ranges(episodes) or (("影片" if self.media.type == MediaType.MOVIE else "整季")
if self.collection.remaining else "无")
if self.collection.remaining else "")
submitted = format_ranges([int(target.split(":")[1]) for target in self.submitted
if ":" in target and not target.endswith(":season")]) or "无"
page = max((source.next_page for source in self.collection.sources.values()), default=0)
if ":" in target and not target.endswith(":season")])
if not submitted and self.submitted:
submitted = "影片" if self.media.type == MediaType.MOVIE else "整季"
added = f"已添加下载 {submitted}" if submitted else "还没有添加下载"
missing = f",还缺 {remaining}" if remaining else ",不缺了"
unreleased = "" if self.full else format_ranges([int(target.split(":")[1]) for target in self.collection.unreleased()])
waiting = f"(其中 {unreleased} 站点尚未收录,交给订阅模式追更)" if unreleased else ""
if not unreleased:
waiting = ""
elif unreleased == remaining:
waiting = "(站点上还没有,之后的定时检查会继续找)"
else:
waiting = f"(其中 {unreleased} 站点上还没有,之后的定时检查会继续找)"
if self.ended:
failed = any(source.failed for source in self.collection.sources.values())
return f"本轮搜索结束,已提交:{submitted},仍缺:{remaining}{waiting}" + (";部分来源失败" if failed else "")
return f"已搜索到第 {page} 页,已提交:{submitted},仍缺:{remaining}{waiting}"
skipped = sorted({self._site_name(key) for key, source in self.collection.sources.items()
if source.failed and not key.startswith("plugin:")})
note = f"。{('、'.join(skipped))} 这次没有搜成" if skipped else ""
return f"搜索结束,{added}{missing}{waiting}{note}"
page = max((source.next_page for source in self.collection.sources.values()), default=0)
return f"已搜到第 {page} 页,{added}{missing}{waiting}"
def _site_name(self, key: str) -> str:
"""
搜索来源对应的站点名,日志里给用户看,不暴露「站点编号:关键词」这种内部键。
"""
site_id = self.queries[key]["site"]
if site_id == "plugin":
return "插件"
return str(self.sources.sites.get(site_id, {}).get("name") or f"站点 {site_id}")
def _lost_ownership(self) -> None:
"""
@@ -480,10 +534,9 @@ class SearchScan:
stop_unmatched=self._stop_unmatched_query(key))
if accepted:
if source.unmatched_pages == UNMATCHED_PAGE_LIMIT:
logger.warning(f"{self.media.title_year} 搜索来源 {key} 连续 {UNMATCHED_PAGE_LIMIT} 页明确不匹配目标作品,"
"本轮停止该来源,继续处理其他来源候选")
logger.info(f"{self.media.title_year} 在 {self._site_name(key)} 用「{self.queries[key]['keyword']}」"
f"连续 {UNMATCHED_PAGE_LIMIT} 页都不是这部作品,这个站不再往后翻")
source.failure_outcome = None
source.budget_deferrals = 0
self.retry.pop(key, None)
self.retry_reasons.pop(key, None)
self.live.update(contexts)
@@ -498,29 +551,37 @@ class SearchScan:
def _record_budget_wait(self, key: str, result: ScanPage) -> None:
"""
翻页与候选票据恢复共享来源长冷却计数,达到上限后仅停止本轮来源。
站点暂时不能请求时决定等还是本轮跳过:出错暂停中的站和要等很久的站本轮不等,
先用其他站点的结果提交,下一轮重新带上;请求间隔这类短暂停顿照常等待。
:param key: 当前页或待恢复候选所属的搜索来源
:param result: 携带重试时间和等待原因的页面结果
"""
source = self.collection.sources[key]
wait_reason = result.wait_reason or "busy"
self.retry[key] = cast(str, result.retry_at)
self.retry_reasons[key] = wait_reason
retry_at = cast(str, result.retry_at)
source.error = result.observation.error
if wait_reason == "cooldown":
# 预算时间保存到秒,当前时间使用相同精度,避免恰好 600 秒的冷却被小数部分排除。
deferral_seconds = (datetime.fromisoformat(cast(str, result.retry_at))
- datetime.now(timezone.utc).replace(microsecond=0)).total_seconds()
if deferral_seconds >= BUDGET_DEFERRAL_SECONDS:
source.budget_deferrals += 1
if source.budget_deferrals >= BUDGET_DEFERRAL_LIMIT:
source.failure_outcome = "deferred"
source.failed = True
self.retry.pop(key, None)
self.retry_reasons.pop(key, None)
logger.warning(f"{self.media.title_year} 搜索来源 {key} 本轮 {source.budget_deferrals} 次等待站点预算未果,"
"跳过该来源以提交已收集候选,下一轮将重新搜索")
wait_seconds = (datetime.fromisoformat(retry_at) - datetime.now(timezone.utc)).total_seconds()
if wait_reason != "failing" and wait_seconds <= LONG_WAIT_SECONDS:
self.retry[key] = retry_at
self.retry_reasons[key] = wait_reason
return
source.failure_outcome = "unavailable" if wait_reason == "failing" else "deferred"
source.failed = True
self.retry.pop(key, None)
self.retry_reasons.pop(key, None)
if wait_reason != "failing":
self.long_waits[key] = (retry_at, wait_reason)
site = self._site_name(key)
if site in self.skipped_sites:
return
self.skipped_sites.add(site)
if wait_reason == "failing" and result.unavailable:
log_site_skipped(self.sources.sites[self.queries[key]["site"]], result.unavailable,
subject=self.media.title_year)
else:
logger.info(f"{self.media.title_year} {site}:{result.observation.error},"
f"要到 {_local_time(retry_at)} 才能再搜,先用其他站点的结果")
def _candidate_context(self, key: str) -> Optional[Context]:
"""
@@ -577,7 +638,8 @@ class SearchScan:
record = self.candidates[key]
source, page = record["source"], record["page"]
cursor = self.collection.sources[source]
if cursor.budget_deferrals >= BUDGET_DEFERRAL_LIMIT:
if cursor.failure_outcome in SKIPPED_THIS_ROUND:
# 所在站点本轮已跳过,临时下载票据取不回来,让其他站点的候选继续提交。
del self.candidates[key]
self.checkpoint()
return None
@@ -587,7 +649,7 @@ class SearchScan:
result = self.sources.fetch(source, page, budget=self._budget())
if result.retry_at:
self._record_budget_wait(source, result)
if cursor.budget_deferrals < BUDGET_DEFERRAL_LIMIT:
if cursor.failure_outcome not in SKIPPED_THIS_ROUND:
self._defer(waiting_source=source)
# 临时票据尚未恢复时不能下载;放弃该候选,让仍可用的次优候选继续提交。
del self.candidates[key]
@@ -601,11 +663,10 @@ class SearchScan:
# 多次重取仍失败时放弃该候选,让排在后面的候选和兜底继续推进,下一轮搜索再重新发现。
self.recovery_failures.pop(key, None)
del self.candidates[key]
logger.warning(f"{self.media.title_year} 候选资源所在页面连续 {failures} 次重取失败,本轮放弃该候选:"
f"{result.observation.error or '请求未成功'}")
logger.warning(f"{self.media.title_year} 资源所在的 {self._site_name(source)} 第 {page + 1} 页连续 {failures} 次"
f"没打开({result.observation.error or '请求未成功'}),这一轮先不下载它")
self.checkpoint()
return None
cursor.budget_deferrals = 0
self.recovery_failures.pop(key, None)
self.retry.pop(source, None)
self.retry_reasons.pop(source, None)
@@ -707,6 +768,25 @@ class SearchScan:
self.collection.deepen(ready - settled - cached)
self.checkpoint()
def _first_pass_targets(self) -> set[str]:
"""
各站第一页到齐后可以先提交的目标,和只搜第一页时一样快;没选中的由 choose 转入深页兜底。
只放行已有候选覆盖的目标,没有候选的目标仍按原规则继续翻页,不提前转成完整兜底。
逐集缺失时各集独立,有候选就先提交。整季缺失时只有第一页已经凑齐全部缺集(整季包或全套散集)
才先提交,否则继续收集到底:先下一部分散集、后面再来整季包会重复下载。首次下载优先整包时,
只有整包能先交,全套散集不算,避免深页的整季包失去被选中的机会。
"""
untried = [key for key in self.candidates if key not in self.attempted]
covered = {target for key in untried for target in self._coverage(key)}
remaining = set(self.collection.remaining)
if self.prefer_complete:
# 首次下载优先整包:只有单个资源就覆盖全部缺集才先交,散集凑齐也继续找整包。
return remaining if any(self._coverage(key) >= remaining for key in untried) else set()
if self.full:
return remaining if covered >= remaining else set()
return covered
def _has_cached_ready(self) -> bool:
"""
末页后仍有分批未提交的就绪候选时,先处理缓存,不能提前结束整轮。
@@ -722,6 +802,8 @@ class SearchScan:
self.collection = SearchCollection(search_targets(self.plan), settled=self.collection.settled)
self.identity, self.candidates, self.live, self.retry, self.retry_reasons = {}, {}, {}, {}, {}
self.recovery_failures = {}
self.first_pass_done = False
self.long_waits = {}
self.sources.queries = {}
self.sources.keyword_index = 0
self.attempted.clear()
@@ -751,10 +833,17 @@ class SearchScan:
retry_at = min(self.retry.values(), default=default_retry_at)
site_ids = tuple(int(self.queries[key]["site"]) for key in active_sources
if self.queries[key]["site"] != "plugin")
reason = "cooldown" if self.retry and all(
self.retry_reasons.get(key) == "cooldown" for key in self.retry) else "busy" if self.retry else "slice"
logger.info(f"{self.media.title_year} 分页搜索暂停({reason}),{retry_at} 后继续:{self.progress_text()}")
raise SubscriptionSearchDeferred(retry_at=retry_at, site_ids=tuple(dict.fromkeys(site_ids)), wait_reason=reason)
waiting_key = min(self.retry, key=lambda key: self.retry[key], default=None)
reason = self.retry_reasons.get(waiting_key, "busy") if waiting_key else "slice"
message = None
if waiting_key and (datetime.fromisoformat(retry_at) - now).total_seconds() > VISIBLE_WAIT_SECONDS:
# 只有明显的等待才告诉用户在等哪个站、为什么、几点继续;请求间隔这类停顿属于正常搜索节奏。
message = f"{self._site_name(waiting_key)} {_WAIT_REASON_TEXT.get(reason, '暂时不能搜索')},{_local_time(retry_at)} 继续搜索"
logger.info(f"{self.media.title_year} {message}。{self.progress_text()}")
else:
logger.debug(f"{self.media.title_year} 搜索稍停,{_local_time(retry_at)} 继续。{self.progress_text()}")
raise SubscriptionSearchDeferred(retry_at=retry_at, site_ids=tuple(dict.fromkeys(site_ids)),
wait_reason=reason, message=message)
def _stopping(self) -> bool:
"""
@@ -776,12 +865,33 @@ class SearchScan:
continue
advanced = self.step(key) or advanced
ready = self.collection.ready(full=self.full)
if not self.first_pass_done and not self.complete_scan and self.collection.first_pages_ready():
self.first_pass_done = True
ready |= self._first_pass_targets()
if ready:
self.choose(ready, submit)
if self.execution and time.monotonic() - started >= SEARCH_SLICE_SECONDS:
self._defer()
return advanced
def _wait_for_long_waits(self) -> bool:
"""
本轮没有添加任何下载、又有站只是要等很久(不是出错)时,改为等最早能再搜的站,
避免把一次到点就能完成的搜索判成失败、拖到下一次定时搜索。
:return: 已把等待中的来源放回重试队列时为 True,调用方继续循环并按重试时间交还队列
"""
if self.submitted or not self.long_waits or not self.collection.remaining:
return False
for key, (retry_at, reason) in self.long_waits.items():
source = self.collection.sources[key]
source.failed = False
source.failure_outcome = None
self.retry[key] = retry_at
self.retry_reasons[key] = reason
self.long_waits = {}
return True
def _finish_round(self) -> None:
"""
本轮结束:删除检查点,避免已完成任务的候选与识别缓存长期留存;没有任何成功来源时沿用任务失败语义。
@@ -814,6 +924,8 @@ class SearchScan:
continue
# 智能模式只剩站点未收录目标时不再尝试别名;完整模式仍保留别名的深页收集机会。
if not (self.full or self.collection.pending) or not self._add_sources():
if self._wait_for_long_waits():
continue
self._finish_round()
return
self.checkpoint()

View File

@@ -155,23 +155,16 @@ def _is_title_match_allowed(
torrent_info: TorrentInfo,
) -> bool:
"""
判断资源是否可以通过标题兜底匹配订阅目标。
判断识别不到作品的资源能否按标题或别名对上订阅目标。
只有资源自身识别不到作品时才会走到这里,没有反证;无年份别名也按订阅目标处理,
宁可偶尔下错同名作品,也不能让资源一直下不下来。
"""
if not TorrentHelper.match_torrent(
return TorrentHelper.match_torrent(
mediainfo=mediainfo,
torrent_meta=torrent_meta,
torrent=torrent_info,
):
return False
if not TorrentHelper.requires_identity_disambiguation(
mediainfo=mediainfo,
torrent_meta=torrent_meta,
):
return True
logger.debug(
f"{torrent_info.site_name} - {torrent_info.title} 仅通过无年份别名命中且候选媒体身份无法确认,已跳过"
)
return False
def _get_media_id_match_source(mediainfo: Optional[MediaInfo]) -> str:

View File

@@ -234,7 +234,7 @@ def _process_paged_subscription(owner: _SubscribeOwnerBase, subscribe: Subscript
page_limit=1 if strategy == "single_page" else None,
execution=execution_context, repository=repository, snapshot=snapshot,
target_scope=(subscribe.start_episode or 1, subscribe.total_episode or 0),
prefer_complete=prefer_initial_pack)
prefer_complete=prefer_initial_pack, complete_scan=strategy == "full")
if execution_context:
execution_context.incremental_search = True

View File

@@ -433,7 +433,13 @@ class TransactionalSubscriptionSearchRepository:
and record.lease_expires_at > now.isoformat(timespec="seconds")
)
if not acquired and cooldown_active:
wait_reason = "cooldown"
# 出错后的暂停与站点访问次数限制要和普通请求间隔分开,调用方据此决定本轮跳过还是等待。
if record.last_outcome == "deferred":
wait_reason = "rate_limit"
elif record.last_outcome not in {None, "success", "skipped"}:
wait_reason = "failing"
else:
wait_reason = "cooldown"
elif not acquired and lease_busy:
wait_reason = "busy"
short_retry = (now + timedelta(seconds=_BUSY_SITE_RETRY_SECONDS)).isoformat(
@@ -447,6 +453,7 @@ class TransactionalSubscriptionSearchRepository:
consecutive_failures=record.consecutive_failures,
lease_token=record.lease_token if acquired else None,
wait_reason=wait_reason,
last_error=record.last_error,
)
return self._write(operation)

View File

@@ -2,7 +2,8 @@
订阅执行状态的请求级异步 SQLAlchemy 适配器。
"""
from typing import Optional
from dataclasses import replace
from typing import Any, Optional
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
@@ -12,14 +13,21 @@ from app.db.models.subscribeactivity import SubscribeActivity
from app.db.models.subscriptionsearch import SubscriptionSearchBatch, SubscriptionSearchTask
def _download_counts(counts: dict[str, Any]) -> dict[str, int]:
"""
从活动计数中取出成功提交的分类计数。
"""
return {key: value for key, value in counts.items()
if key in {"submitted", "initial_submitted", "upgrade_submitted", "full_upgrade_submitted"}
and isinstance(value, int)}
def _task(record: SubscriptionSearchTask, activity: Optional[SubscribeActivity] = None) -> SearchTaskSnapshot:
"""
复制可脱离 AsyncSession 使用的搜索任务快照。
"""
return SearchTaskSnapshot(
download_counts={key: value for key, value in (activity.counts if activity else {}).items()
if key in {"submitted", "initial_submitted", "upgrade_submitted", "full_upgrade_submitted"}
and isinstance(value, int)},
download_counts=_download_counts(activity.counts if activity else {}),
downloads=tuple({**resource, "subscription_id": record.subscription_id, "task_id": record.task_id}
for resource in (activity.details.get("resources", []) if activity else [])
if isinstance(resource, dict) and resource.get("decision") == "submitted"),
@@ -96,7 +104,16 @@ class SessionSubscriptionExecutionStatusRepository:
snapshots: dict[int, SearchTaskSnapshot] = {}
for record in result.scalars().all():
snapshots.setdefault(record.subscription_id, _task(record))
return snapshots
if not snapshots:
return snapshots
# 列表只需要提交计数,单独读取最新任务的 counts,避免带出活动明细大字段。
counts = await self._session.execute(
select(SubscribeActivity.event_id, SubscribeActivity.counts)
.where(SubscribeActivity.event_id.in_([task.task_id for task in snapshots.values()]))
)
by_task = {event_id: value for event_id, value in counts.all()}
return {subscription_id: replace(task, download_counts=_download_counts(by_task.get(task.task_id) or {}))
for subscription_id, task in snapshots.items()}
async def list_batches(self, *, limit: int) -> list[SearchBatchSnapshot]:
"""

View File

@@ -68,7 +68,6 @@ class SearchSourceCursor:
:param page_limit: 当前来源页数上限;未指定时使用全局上限
:param matched_work: 作品匹配不局限于当前缺集;见过同作品的其他季集也不能按异常查询提前停止
:param unmatched_pages: 从未匹配目标作品时连续出现的明确不匹配页数
:param budget_deferrals: 本轮因站点预算长冷却被推迟的次数;达到上限后本轮结算该来源,跨轮重新累计
"""
next_page: int = 0
@@ -86,7 +85,6 @@ class SearchSourceCursor:
page_limit: Optional[int] = None
matched_work: bool = False
unmatched_pages: int = 0
budget_deferrals: int = 0
@property
def limit_reached(self) -> bool:
@@ -224,6 +222,15 @@ class SearchCollection:
result.add(target)
return result
def first_pages_ready(self) -> bool:
"""
各来源都已拿到第一页,或本轮已结束、失败、被跳过;此时可以先用已有资源提交。
"""
return bool(self.sources) and all(
source.next_page > 0 or source.exhausted or source.failed or source.limit_reached
for source in self.sources.values()
)
def ready(self, *, full: bool = False) -> set[str]:
"""
完整模式等待所有来源结束;智能模式可按关闭证据或站点未收录判断提前就绪。

View File

@@ -131,6 +131,7 @@ class SubscriptionExecutionStatus(BaseModel):
next_run_at: Optional[str] = Field(default=None, description="下次计划执行时间")
error: Optional[str] = Field(default=None, description="错误摘要")
can_cancel: bool = Field(default=False, description="当前任务是否允许取消")
submitted_count: int = Field(default=0, description="本次任务已添加的下载数量,用于区分已添加下载与暂无资源")
model_config = ConfigDict(from_attributes=True)

View File

@@ -678,10 +678,11 @@ def _run_site_pressure_case(
)
cooldown_calls = boundary.calls - calls_before_cooldown
cooldown_deferrals = loser.consume_subscription_site_budget_deferrals()
# 出错暂停中的站点本轮直接跳过:不发请求,也不让订阅进入等待。
error_cooldown_blocked = bool(
not cooldown_result
and cooldown_calls == 0
and cooldown_deferrals
and not cooldown_deferrals
)
error_cooldown_persisted = bool(
error_record

View File

@@ -82,7 +82,7 @@ def checkpoint(scan: SearchScan) -> SearchSessionSnapshot:
def test_third_unmatched_page_releases_other_site_candidate(search):
"""
异常站点恰好请求 3 页,随后提交另一站点早已匹配的电影,而不是等待 20 页。
另一站点第一页已匹配电影时,各站第一页到齐就提交,不等异常站点翻页。
"""
owner, data, calls = search
data[1, 0] = [resource(1, 0, matched=True)]
@@ -92,15 +92,15 @@ def test_third_unmatched_page_releases_other_site_candidate(search):
def submit(items, ready):
"""
记录提交发生时的请求边界,确认尚未请求异常来源第 4 页。
记录提交发生时的请求边界,确认异常来源只请求了第一页。
"""
batches.append((list(calls), [item.torrent_info.site for item in items]))
return ready
scan.run(submit)
assert batches == [([(1, 0), (2, 0), (1, 1), (2, 1), (2, 2)], [1])]
assert batches == [([(1, 0), (2, 0)], [1])]
source = scan.collection.sources["2:tt4633694"]
assert source.unmatched_pages == 3 and source.limit_reached
assert source.unmatched_pages == 1
assert not source.exhausted and not source.failed
assert scan.ended and not scan.collection.remaining

View File

@@ -162,3 +162,61 @@ def test_parallel_filter_propagates_request_context(monkeypatch):
assert len(contexts) == 2
assert observed == ["filter-request", "filter-request"]
def _hospital_playlist() -> MediaInfo:
"""
构造订阅目标《机智的医生生活》:英文名只作为别名出现,资源名通常不带年份。
"""
return MediaInfo(
media_source=MediaSource.TMDB,
media_id="96102",
tmdb_id=96102,
title="机智的医生生活",
original_title="슬기로운 의사생활",
names=["Hospital Playlist"],
type=MediaType.TV,
year="2020",
original_language="ko",
season_years={1: "2020", 2: "2021"},
)
def _parse_with_candidate(monkeypatch, candidate, title="Hospital Playlist S02 1080p NF WEB-DL DDP2.0 x264-DoA"):
"""
让按名字的再次识别返回指定候选,返回解析后的搜索结果。
"""
media_chain = Mock()
media_chain.recognize_by_meta.return_value = candidate
monkeypatch.setattr(search_module, "MediaChain", Mock(return_value=media_chain))
monkeypatch.setattr(search_module.TorrentHelper, "sort_torrents", staticmethod(lambda contexts: contexts))
chain = object.__new__(SearchChain)
torrent = TorrentInfo(title=title, site_name="测试站点", category=MediaType.TV.value)
return chain._parse_result(torrents=[torrent], mediainfo=_hospital_playlist(), rule_groups=[])
def test_no_year_alias_accepts_candidate_without_first_air_date(monkeypatch):
"""
按名字认出的是没有首播日期的残缺重复条目时,不算另一部作品,资源按订阅目标处理。
"""
candidate = MediaInfo(media_source=MediaSource.TMDB, media_id="290143", tmdb_id=290143,
title="Hospital Playlist", original_title="Hospital Playlist",
type=MediaType.TV, original_language="tr")
assert len(_parse_with_candidate(monkeypatch, candidate)) == 1
def test_no_year_alias_accepts_same_year_work_with_different_original_title(monkeypatch):
"""
同年的同名衍生节目只是原名不同,不足以否定订阅目标,宁可偶尔下错也要能下。
"""
candidate = MediaInfo(media_source=MediaSource.TMDB, media_id="124625", tmdb_id=124625,
title="슬기로운 하드털이", original_title="슬기로운 하드털이",
type=MediaType.TV, year="2020", original_language="ko")
assert len(_parse_with_candidate(monkeypatch, candidate)) == 1
def test_no_year_alias_accepts_when_name_is_not_recognized(monkeypatch):
"""
按名字识别不到任何作品时没有反证,资源按订阅目标处理。
"""
assert len(_parse_with_candidate(monkeypatch, None)) == 1

View File

@@ -21,8 +21,7 @@ from app.chain.search.facade import SearchChain
from app.chain.search.result import TorrentHelper
from app.chain.search.scan import (
AUTOMATIC_REQUEST_INTERVAL,
BUDGET_DEFERRAL_LIMIT,
BUDGET_DEFERRAL_SECONDS,
LONG_WAIT_SECONDS,
ScanPage,
SearchScan,
SearchSources,
@@ -84,7 +83,7 @@ def pages(owner, monkeypatch, data):
def test_thirty_seven_episodes_fill_older_pages_and_download_early(owner, monkeypatch):
"""
多页搜索补齐缺集,同时在较早目标具备候选时提前提交下载。
多页搜索补齐缺集;第一页到齐就先提交已有的集,较早的集继续往后翻。
"""
calls = pages(owner, monkeypatch, {0: [torrent(ep, v) for ep in range(16, 38) for v in range(1, 4)],
1: [torrent(ep) for ep in range(1, 16)]})
@@ -95,7 +94,7 @@ def test_thirty_seven_episodes_fill_older_pages_and_download_early(owner, monkey
return ready
scan.run(submit)
assert calls == [0, 1, 2]
assert batches[0] == (1, {f"1:{ep}" for ep in range(16, 38)})
assert batches[0] == (0, {f"1:{ep}" for ep in range(16, 38)})
assert not scan.collection.remaining
@@ -117,7 +116,7 @@ def test_single_page_policy_does_not_expand_after_missing_or_failed_download(own
source = scan.collection.sources["1:Show"]
assert source.limit_reached
assert not source.exhausted
assert scan.ended and "本轮搜索结束" in scan.progress_text()
assert scan.ended and scan.progress_text().startswith("搜索结束")
def test_single_page_policy_keeps_each_alias_first_page(owner, monkeypatch):
@@ -195,10 +194,10 @@ def test_single_page_limit_survives_checkpoint_before_download(owner, monkeypatc
assert restored.collection.sources["1:Show"].limit_reached
@pytest.mark.parametrize("full,expected", [(False, [0, 1]), (True, [0, 1, 2, 3])])
@pytest.mark.parametrize("full,expected", [(False, [0]), (True, [0, 1, 2, 3])])
def test_movie_smart_boundary_and_complete_upgrade_search(owner, monkeypatch, full, expected):
"""
电影智能搜索按覆盖边界停止,完整搜索继续收集后续升级候选。
电影智能搜索第一页找到就提交,完整搜索继续收集后续升级候选。
"""
from app.application.subscription.contract import SubscriptionSnapshot
from app.chain.download import DownloadChain
@@ -359,7 +358,8 @@ def test_whole_season_restarts_old_smart_checkpoint_and_keeps_submitted_targets(
calls = pages(owner, monkeypatch, {0: [torrent(1), torrent(2)]})
batches = []
scan.run(lambda _items, ready: batches.append(ready) or ready)
assert calls == [0, 1]
# 第一页已凑齐剩下的集,先提交,不再往后翻。
assert calls == [0]
assert batches == [{"1:2"}]
assert scan.submitted == {"1:1", "1:2"}
@@ -391,7 +391,8 @@ def test_unchanged_resource_recognized_once_even_when_repeated(owner, monkeypatc
identified.append(params["torrent"].description)
return original(**params)
monkeypatch.setattr(result_module, "_torrent_meta", identify)
SearchScan(owner=owner, plan=plan([1, 4])).run(lambda _items, ready: ready)
# 下载层一直拒绝,搜索才会翻过重复资源所在的后续页。
SearchScan(owner=owner, plan=plan([1, 4])).run(lambda _items, _ready: set())
assert calls == [0, 1, 2]
assert identified == ["1-1", "4-1"]
@@ -400,7 +401,8 @@ def test_twelve_hundred_episodes_with_multiple_versions(owner, monkeypatch):
"""
20 页范围内仍能收集 1200 集的多版本候选,不在最后一页遗漏提交。
"""
data = {page: [torrent(episode, version) for episode in range(page * 60 + 1, (page + 1) * 60 + 1)
# 站点按上传时间倒序返回,越新的集越靠前。
data = {page: [torrent(episode, version) for episode in range(1200 - page * 60, 1200 - (page + 1) * 60, -1)
for version in (1, 2)] for page in range(20)}
calls = pages(owner, monkeypatch, data)
scan = SearchScan(owner=owner, plan=plan(list(range(1, 1201))))
@@ -438,7 +440,7 @@ def test_full_mode_collects_late_candidates_after_a_gap(owner, monkeypatch):
def test_pack_waits_for_all_covered_goals_and_is_submitted_once(owner, monkeypatch):
"""
合集等待其覆盖目标全部具备选择条件后只提交一次。
第一页的合集已覆盖全部缺集时立即提交,且只提交一次。
"""
pack = TorrentInfo(site=1, description="pack", title="Show S01E01-E04 2026 1080p", enclosure="https://site.example/download?id=pack")
calls = pages(owner, monkeypatch, {0: [pack], 1: [torrent(4)], 2: [torrent(9)]})
@@ -446,27 +448,27 @@ def test_pack_waits_for_all_covered_goals_and_is_submitted_once(owner, monkeypat
scan = SearchScan(owner=owner, plan=plan([1, 4]))
scan.run(lambda items, ready: batches.append((calls[-1], {item.torrent_info.description for item in items}, ready)) or ready)
assert len(batches) == 1
assert batches[0][0] == 2
assert batches[0][0] == 0
assert "pack" in batches[0][1]
assert batches[0][2] == {"1:1", "1:4"}
def test_single_episode_can_download_while_a_pack_waits(owner, monkeypatch):
"""
合集仍在等待时,独立集数可以先提交下载。
第一页同时有合集和单集时一起交给下载层择优,合集只提交一次。
"""
pack = TorrentInfo(site=1, description="pack", title="Show S01E01-E04 2026 1080p", enclosure="https://site.example/download?id=pack")
calls = pages(owner, monkeypatch, {0: [pack, torrent(1)], 1: [torrent(4)], 2: [torrent(9)]})
batches = []
scan = SearchScan(owner=owner, plan=plan([1, 4]))
scan.run(lambda items, ready: batches.append((calls[-1], {item.torrent_info.description for item in items}, ready)) or ready)
assert batches[0] == (1, {"1-1"}, {"1:1"})
assert batches[0] == (0, {"pack", "1-1"}, {"1:1", "1:4"})
assert sum("pack" in items for _, items, _ in batches) == 1
def test_cross_site_barrier_keeps_late_opportunistic_candidate(owner, monkeypatch):
"""
跨站点等待期间保留其他站点稍后发现的可用候选。
各站第一页到齐后先提交已找到的集,不等其他站翻完。
"""
monkeypatch.setattr(owner, "_sync_indexers", lambda _ids: [{"id": 1}, {"id": 2}])
data = {1: {0: [torrent(1)], 1: [torrent(4)], 2: [torrent(1, 7), torrent(4, 2)]},
@@ -485,9 +487,7 @@ def test_cross_site_barrier_keeps_late_opportunistic_candidate(owner, monkeypatc
batches = []
SearchScan(owner=owner, plan=plan([1, 4])).run(
lambda items, ready: batches.append((calls[-1], {item.torrent_info.description for item in items}, ready)) or ready)
assert batches[0][0] == (2, 2)
assert "1-7" in batches[0][1]
assert batches[0][2] == {"1:1"}
assert batches[0] == ((2, 0), {"1-1"}, {"1:1"})
def test_rejected_candidate_is_rejudged_only_in_a_new_round(owner, monkeypatch):
@@ -595,7 +595,7 @@ def test_rejected_ticket_candidates_finish_across_short_budget_waits(
assert submitted == ["1-1", "1-2"]
assert calls == ([0, 1, 2] if source_active else [0, 1])
assert scan.collection.remaining == {"1:1", "1:2"}
assert scan.collection.sources[source_key].budget_deferrals == 0
assert not scan.collection.sources[source_key].failed
finally:
engine.dispose()
@@ -612,7 +612,7 @@ def test_site_window_wait_keeps_cursor(owner, monkeypatch):
scan = SearchScan(owner=owner, plan=plan([1]))
key = next(key for key in scan.queries if not key.startswith("plugin:"))
assert not scan.step(key)
assert scan.retry_reasons[key] == "cooldown"
assert scan.retry_reasons[key] == "rate_limit"
assert scan.collection.sources[key].next_page == 0
assert not scan.collection.sources[key].failed
assert not scan.collection.sources[key].exhausted
@@ -646,12 +646,18 @@ def test_site_rate_limit_window_defers_retry_until_window_refresh(owner, monkeyp
assert not scan.step(key)
delay = (datetime.fromisoformat(scan.retry[key]) - before).total_seconds()
assert seconds - 1 < delay <= seconds
assert scan.retry_reasons[key] == "cooldown"
assert not scan.collection.sources[key].failed
assert not scan.collection.sources[key].exhausted
assert scan.collection.sources[key].budget_deferrals == (1 if seconds >= 600 else 0)
source = scan.collection.sources[key]
assert not source.exhausted
if seconds <= LONG_WAIT_SECONDS:
delay = (datetime.fromisoformat(scan.retry[key]) - before).total_seconds()
assert seconds - 1 < delay <= seconds
assert scan.retry_reasons[key] == "rate_limit"
assert not source.failed
else:
# 次数窗口要等一小时,本轮不等这个站,下一轮再搜。
assert key not in scan.retry
assert source.failed
assert source.failure_outcome == "deferred"
def test_scan_persists_local_rate_limit_window_on_site_budget(owner, monkeypatch):
@@ -776,7 +782,7 @@ def test_all_sources_failing_is_not_reported_as_exhausted(owner, monkeypatch):
source = next(value for key, value in scan.collection.sources.items() if not key.startswith("plugin:"))
assert not source.exhausted
assert source.next_page == 0
assert "部分来源失败" in scan.progress_text()
assert "这次没有搜成" in scan.progress_text()
def test_language_or_capability_skip_keeps_existing_non_failure_semantics(owner, monkeypatch):
@@ -917,7 +923,7 @@ def test_one_unidentifiable_resource_does_not_block_the_page(owner, monkeypatch)
scan = SearchScan(owner=owner, plan=plan([1]))
scan.run(lambda items, ready: submitted.extend(item.torrent_info.description for item in items) or ready)
assert submitted == ["1-1"]
assert calls == [0, 1]
assert calls == [0]
assert not scan.collection.remaining
@@ -965,7 +971,7 @@ def test_progress_log_summary_reports_page_and_remaining_episodes(owner, monkeyp
scan = SearchScan(owner=owner, plan=plan(list(range(1, 38))))
for key in list(scan.collection.sources):
scan.step(key)
assert scan.progress_text() == "已搜索到第 1 页,已提交:无,仍缺:E01-E37"
assert scan.progress_text() == "已搜到第 1 页,还没有添加下载,还缺 E01-E37"
def test_site_repeating_the_same_page_ends_the_source_instead_of_looping(owner, monkeypatch):
"""
@@ -980,10 +986,11 @@ def test_site_repeating_the_same_page_ends_the_source_instead_of_looping(owner,
return [torrent(1)] if params["page"] == 0 else repeated
monkeypatch.setattr(owner, "search_site_torrents", request)
scan = SearchScan(owner=owner, plan=plan([1, 2]))
scan.run(lambda _items, ready: ready)
# 下载层一直拒绝,搜索才会继续往后翻到重复页。
scan.run(lambda _items, _ready: set())
assert calls == [0, 1, 2]
assert scan.ended
assert scan.collection.remaining == {"1:2"}
assert scan.collection.remaining == {"1:1", "1:2"}
def _snapshot_after_first_pages(owner, episodes, **changes):
@@ -1064,8 +1071,8 @@ def test_checkpoint_older_than_thirty_six_hours_restarts_from_first_page(owner,
def test_failed_episode_falls_back_and_uses_a_later_page_candidate_after_sources_finish(owner, monkeypatch):
# 第 0 页有 E1,第 1 页没有:E1 结束收集并提交,但下载失败;
# 为 E5 继续翻页时第 2 页又出现新的 E1,兜底在来源翻完后用它重新择优,失败的不再重试。
# 第 0 页有 E1 和 E5:第一页到齐后一起提交,E1 下载失败;
# 为 E1 继续翻页时第 2 页又出现新的 E1,兜底在来源翻完后用它重新择优,失败的不再重试。
"""
单集下载失败后,在来源结束时使用后续页候选回退。
"""
@@ -1078,7 +1085,7 @@ def test_failed_episode_falls_back_and_uses_a_later_page_candidate_after_sources
item.torrent_info.description == "1-2" for item in items) else set())
scan = SearchScan(owner=owner, plan=plan([1, 5]))
scan.run(submit)
assert ["1-1"] in attempts
assert attempts[0] == ["1-1", "5-1"]
assert "1-2" in attempts[-1] and "1-1" not in attempts[-1]
assert calls == [0, 1, 2, 3]
assert not scan.collection.remaining
@@ -1186,11 +1193,11 @@ def test_ongoing_series_releases_round_after_available_episodes_are_submitted(ow
submitted = []
scan = SearchScan(owner=owner, plan=plan([1003, 1006, 1007]))
scan.run(lambda _items, ready: submitted.append(set(ready)) or ready)
assert calls == [0, 1]
assert calls == [0]
assert submitted == [{"1:1003"}]
assert scan.ended
assert scan.collection.remaining == {"1:1006", "1:1007"}
assert "E1006-E1007 站点尚未收录" in scan.progress_text()
assert "还缺 E1006-E1007(站点上还没有,之后的定时检查会继续找)" in scan.progress_text()
def test_only_unreleased_episodes_check_first_page_and_do_not_try_aliases(owner, monkeypatch):
@@ -1331,13 +1338,12 @@ def _restored_ticket_scan(owner, monkeypatch, first_page, refetched, *, prefer_c
snapshot=snapshot), fetched
@pytest.mark.parametrize("outcome", ["cooldown", "error"])
@pytest.mark.parametrize("recovery_succeeds", [False, True])
@pytest.mark.parametrize("outcome, recovery_succeeds", [("cooldown", True), ("error", False), ("error", True)])
def test_complete_pack_recovery_waits_across_checkpoints_before_partial_fallback(
owner, monkeypatch, scan_clock, outcome, recovery_succeeds,
):
"""
整包恢复等待期间不提交分集;检查点保留重试次数,恢复成功或达到上限后才能推进。
整包恢复的短暂等待期间不提交分集;检查点保留重试次数,恢复成功或失败达到上限后才能推进。
"""
pack = TorrentInfo(site=1, description="pack", title="Show S01 Complete 2026 1080p")
scan, _ = _restored_ticket_scan(owner, monkeypatch, [torrent(1), pack], [], prefer_complete=True)
@@ -1355,7 +1361,7 @@ def test_complete_pack_recovery_waits_across_checkpoints_before_partial_fallback
enclosure="https://site.example/download?id=pack")
return ScanPage([torrent(1), restored], SiteSearchObservation(True, "success", raw_count=2))
if outcome == "cooldown":
retry_at = (scan_clock["now"] + timedelta(hours=1)).isoformat(timespec="seconds")
retry_at = (scan_clock["now"] + timedelta(minutes=5)).isoformat(timespec="seconds")
return ScanPage([], SiteSearchObservation(), retry_at, "cooldown")
return ScanPage([], SiteSearchObservation(True, "error", error="timeout"))
@@ -1384,6 +1390,22 @@ def test_complete_pack_recovery_waits_across_checkpoints_before_partial_fallback
assert submitted == ([({"1-1", "pack"}, {"1:1", "1:2"})] if recovery_succeeds else [({"1-1"}, {"1:1"})])
def test_complete_pack_recovery_long_cooldown_falls_back_to_parts_at_once(owner, monkeypatch, scan_clock):
"""
整包所在站点要等一小时才能重取票据时,本轮不等,直接用已有分集提交。
"""
pack = TorrentInfo(site=1, description="pack", title="Show S01 Complete 2026 1080p")
scan, _ = _restored_ticket_scan(owner, monkeypatch, [torrent(1), pack], [], prefer_complete=True)
part_key = next(key for key, saved in scan.candidates.items() if saved["torrent"]["description"] == "1-1")
scan.candidates[part_key]["torrent"]["enclosure"] = "https://site.example/download?id=1"
retry_at = (scan_clock["now"] + timedelta(hours=1)).isoformat(timespec="seconds")
monkeypatch.setattr(SearchSources, "fetch", lambda *_args, **_kwargs: ScanPage(
[], SiteSearchObservation(), retry_at, "rate_limit"))
submitted = []
scan.run(lambda items, goals: submitted.append(({item.torrent_info.description for item in items}, goals)) or goals)
assert submitted == [({"1-1"}, {"1:1"})]
@pytest.mark.parametrize("still_listed", [False, True])
def test_several_candidates_vanishing_from_one_refetched_page_do_not_abort_the_round(owner, monkeypatch, still_listed):
# A、B 同在原页且都需重取票据;重取后两者都不在原页(原页为空或被其他资源挤占),不能抛 KeyError。
@@ -1484,10 +1506,9 @@ def scan_clock(monkeypatch):
return clock
def test_budget_deferred_dead_site_settles_and_releases_cached_candidates(owner, monkeypatch, scan_clock):
def test_long_wait_site_is_skipped_and_cached_candidates_submit(owner, monkeypatch, scan_clock):
"""
永久预算冷却的站点在本轮结算后放行"全部来源才提交"的屏障,缓存候选得以提交;
结算只在本轮有效,下一轮该站点重新参与搜索。
要等很久的站点本轮直接跳过,不拖住其他站点已找到的资源;跳过只在本轮有效。
"""
monkeypatch.setattr(owner, "_sync_indexers",
lambda _sites: [{"id": 1, "name": "test", "parser": "mTorrent"},
@@ -1506,39 +1527,47 @@ def test_budget_deferred_dead_site_settles_and_releases_cached_candidates(owner,
monkeypatch.setattr(owner, "search_site_torrents", request)
monkeypatch.setattr(owner, "_subscription_site_budget",
_StubSiteBudget(deferred_site_id=2, retry_seconds=3600.0), raising=False)
scan = SearchScan(owner=owner, plan=plan([1]))
submitted = []
def submit(contexts, ready):
"""
记录屏障放行后真正提交的候选。
记录不等故障站点就提交的候选。
"""
submitted.extend(contexts)
return ready
# 按保存的重试时间推进时钟,并恢复检查点,模拟持久队列再次唤醒。
for _ in range(5):
scan.run(submit)
if scan.ended:
break
if scan.retry:
scan_clock["now"] = datetime.fromisoformat(min(scan.retry.values())) + timedelta(seconds=1)
snapshot = SearchSessionSnapshot("round", 0, encode_search_state(scan.state()))
scan = SearchScan(owner=owner, plan=plan([1]), snapshot=snapshot)
scan = SearchScan(owner=owner, plan=plan([1]))
scan.run(submit)
deferred_keys = [key for key in scan.collection.sources if key.startswith("2:")]
assert deferred_keys
assert all(scan.collection.sources[key].failed for key in deferred_keys)
assert all(scan.collection.sources[key].budget_deferrals >= BUDGET_DEFERRAL_LIMIT for key in deferred_keys)
assert all(scan.collection.sources[key].failure_outcome == "deferred" for key in deferred_keys)
assert not any(key in scan.retry for key in deferred_keys)
assert scan.collection.settled == {"1:1"}
assert submitted
assert scan.ended is True
assert not SearchScan(owner=owner, plan=plan([1])).collection.sources[deferred_keys[0]].failed
def test_failing_site_is_skipped_even_when_cooldown_is_short(owner, monkeypatch):
"""
出错暂停中的站点不论暂停多久都不等,记为本轮不可用,不写入重试时间。
"""
retry_at = (datetime.now(timezone.utc) + timedelta(seconds=180)).isoformat(timespec="seconds")
monkeypatch.setattr(SearchSources, "fetch", lambda *_args, **_kwargs: ScanPage(
[], SiteSearchObservation(error="站点最近搜索出错"), retry_at, "failing"))
scan = SearchScan(owner=owner, plan=plan([1]))
assert not scan.step("1:Show")
source = scan.collection.sources["1:Show"]
assert source.failed is True
assert source.failure_outcome == "unavailable"
assert "1:Show" not in scan.retry
def test_short_budget_wait_does_not_settle_source(owner, monkeypatch):
"""
秒级预算等待(请求间隔、并发租约)属于健康站点噪声,不得计数或结算来源。
秒级预算等待(请求间隔、并发租约)属于健康站点的正常节奏,照常等待,不结算来源。
"""
monkeypatch.setattr(owner, "_subscription_site_budget",
_StubSiteBudget(deferred_site_id=1, retry_seconds=11.0), raising=False)
@@ -1560,28 +1589,28 @@ def test_short_budget_wait_does_not_settle_source(owner, monkeypatch):
key = next(key for key in scan.collection.sources if key.startswith("1:"))
source = scan.collection.sources[key]
assert source.failed is False
assert source.budget_deferrals == 0
assert submitted == []
assert scan.ended is False
def test_exact_long_cooldown_counts_after_timestamp_truncation(owner, monkeypatch, scan_clock):
@pytest.mark.parametrize("seconds, skipped", [(LONG_WAIT_SECONDS, False), (LONG_WAIT_SECONDS + 1, True)])
def test_wait_longer_than_limit_skips_source_for_round(owner, monkeypatch, scan_clock, seconds, skipped):
"""
预算时间保存到秒,恰好达到长冷却门槛时不能因当前时间的小数部分漏计。
恰好等到上限仍然等待,超过上限才本轮跳过该来源。
"""
fixed_now = scan_clock["now"]
retry_at = (fixed_now + timedelta(seconds=BUDGET_DEFERRAL_SECONDS)).isoformat(timespec="seconds")
retry_at = (scan_clock["now"] + timedelta(seconds=seconds)).isoformat(timespec="seconds")
monkeypatch.setattr(SearchSources, "fetch", lambda *_args, **_kwargs: ScanPage(
[], SiteSearchObservation(), retry_at, "cooldown"))
[], SiteSearchObservation(), retry_at, "rate_limit"))
scan = SearchScan(owner=owner, plan=plan([1]))
scan.step("1:Show")
assert scan.collection.sources["1:Show"].budget_deferrals == 1
assert scan.collection.sources["1:Show"].failed is skipped
assert ("1:Show" in scan.retry) is not skipped
@pytest.mark.parametrize("ticket_count", [1, 2])
def test_ticket_recovery_long_cooldown_is_bounded_across_checkpoints(owner, monkeypatch, scan_clock, ticket_count):
def test_ticket_recovery_long_cooldown_submits_backup_without_waiting(owner, monkeypatch, scan_clock, ticket_count):
"""
缓存候选重取票据也必须有限等待;恢复检查点后保留计数并继续提交可用候选。
缓存候选重取票据时站点要等很久,本轮放弃这些候选,直接提交仍可用的候选。
"""
scan, _ = _restored_ticket_scan(owner, monkeypatch, [torrent(1, version)
for version in range(1, ticket_count + 2)], [])
@@ -1592,11 +1621,11 @@ def test_ticket_recovery_long_cooldown_is_bounded_across_checkpoints(owner, monk
def fetch(_sources, _key, page, **_params):
"""
模拟原页重取持续受站点长冷却阻塞。
模拟原页重取受站点长冷却阻塞。
"""
requests.append(page)
retry_at = (scan_clock["now"] + timedelta(hours=1)).isoformat(timespec="seconds")
return ScanPage([], SiteSearchObservation(), retry_at, "cooldown")
return ScanPage([], SiteSearchObservation(), retry_at, "rate_limit")
monkeypatch.setattr(SearchSources, "fetch", fetch)
submitted = []
@@ -1608,69 +1637,120 @@ def test_ticket_recovery_long_cooldown_is_bounded_across_checkpoints(owner, monk
submitted.extend(item.torrent_info.description for item in items)
return goals
for version in range(BUDGET_DEFERRAL_LIMIT - 1):
with pytest.raises(SubscriptionSearchDeferred) as error:
scan.run(submit)
assert error.value.wait_reason == "cooldown"
assert error.value.site_ids == (1,)
assert datetime.fromisoformat(error.value.retry_at) > scan_clock["now"] + timedelta(minutes=59)
snapshot = SearchSessionSnapshot("round", version, encode_search_state(scan.state()))
scan = SearchScan(owner=owner, plan=plan([1, 2]), full=True, snapshot=snapshot)
with pytest.raises(SubscriptionSearchDeferred):
scan.run(submit)
assert len(requests) == version + 1
scan_clock["now"] = datetime.fromisoformat(error.value.retry_at) + timedelta(seconds=1)
scan.run(submit)
assert len(requests) == BUDGET_DEFERRAL_LIMIT
assert len(requests) == 1
assert submitted == [f"1-{ticket_count + 1}"]
assert scan.ended
def test_successful_ticket_recovery_resets_budget_wait_count(owner, monkeypatch):
def test_legacy_checkpoint_with_wait_count_and_without_processed_candidates_restores(owner):
"""
原页成功取回票据后清零等待计数,后续页不继承此前冷却。
"""
scan, _ = _restored_ticket_scan(owner, monkeypatch, [torrent(1)], [torrent(1)])
scan.collection.sources["1:Show"].budget_deferrals = BUDGET_DEFERRAL_LIMIT - 1
scan.run(lambda _items, goals: goals)
assert scan.collection.sources["1:Show"].budget_deferrals == 0
assert scan.collection.settled == {"1:1"}
@pytest.mark.parametrize("wait_reason, seconds, expected", [
("cooldown", 11, 1), ("busy", 3600, 1), ("cooldown", 3600, 2),
])
def test_budget_wait_count_restores_and_success_resets(owner, monkeypatch, wait_reason, seconds, expected):
"""
检查点保留长冷却累计,短等待与忙碌不加计数,成功取页后清零。
"""
scan = SearchScan(owner=owner, plan=plan([1]))
scan.collection.sources["1:Show"].budget_deferrals = 1
snapshot = SearchSessionSnapshot("round", 0, encode_search_state(scan.state()))
scan = SearchScan(owner=owner, plan=plan([1]), snapshot=snapshot)
retry_at = (datetime.now(timezone.utc) + timedelta(seconds=seconds)).isoformat(timespec="seconds")
monkeypatch.setattr(SearchSources, "fetch", lambda *_args, **_kwargs: ScanPage(
[], SiteSearchObservation(), retry_at, wait_reason))
scan.step("1:Show")
assert scan.collection.sources["1:Show"].budget_deferrals == expected
pages(owner, monkeypatch, {0: []})
monkeypatch.setattr(SearchSources, "fetch", lambda *_args, **_kwargs: ScanPage(
[], SiteSearchObservation(True, "success", raw_count=0)))
scan.step("1:Show")
assert scan.collection.sources["1:Show"].budget_deferrals == 0
assert SearchScan(owner=owner, plan=plan([1])).collection.sources["1:Show"].budget_deferrals == 0
def test_legacy_checkpoint_defaults_budget_wait_count_and_processed_candidates(owner):
"""
缺少预算计数和候选处理记录的旧检查点仍可恢复,不会重启或提前结算来源。
旧检查点带已废弃的长等待计数、缺少候选处理记录时仍可恢复,不会重启或提前结算来源。
"""
state = SearchScan(owner=owner, plan=plan([1])).state()
state.pop("attempted")
for source in state["collection"]["sources"].values():
source.pop("budget_deferrals")
source["budget_deferrals"] = 2
snapshot = SearchSessionSnapshot("round", 0, encode_search_state(state))
scan = SearchScan(owner=owner, plan=plan([1]), snapshot=snapshot)
assert not scan.restart
assert not scan.attempted
assert all(source.budget_deferrals == 0 and not source.failed for source in scan.collection.sources.values())
assert all(not source.failed for source in scan.collection.sources.values())
def test_whole_season_submits_first_page_pack_without_paging_deeper(owner, monkeypatch):
"""
首次订阅缺整季时,各站第一页已凑齐全部缺集就立即提交,不再翻完所有页。
"""
monkeypatch.setattr(owner, "_sync_indexers", lambda _sites: [{"id": 1, "name": "A"}, {"id": 2, "name": "B"}])
season_plan = plan([])
season_plan.no_exists["tmdb:1"][1].total_episode = 3
pack = TorrentInfo(site=2, description="complete", title="Show 2026 S01 Complete 1080p",
enclosure="https://site.example/download?id=complete")
data = {"1:Show": {0: [torrent(1)], 1: [torrent(2)]}, "2:Show": {0: [pack], 1: [torrent(3)]}}
calls = []
def fetch(_sources, key, page, **_params):
"""
两个站点都还有后续页,第一页已经出现整季包。
"""
calls.append((key, page))
items = data.get(key, {}).get(page, [])
return ScanPage(items, SiteSearchObservation(True, "success", raw_count=len(items), has_more=True))
monkeypatch.setattr(SearchSources, "fetch", fetch)
submitted = []
scan = SearchScan(owner=owner, plan=season_plan, full=True, prefer_complete=True, complete_scan=False)
scan.run(lambda items, ready: submitted.append({item.torrent_info.description for item in items}) or ready)
assert submitted == [{"1-1", "complete"}]
assert ("1:Show", 1) not in calls and ("2:Show", 1) not in calls
assert scan.ended and not scan.collection.remaining
def test_complete_scan_strategy_keeps_collecting_after_first_pages(owner, monkeypatch):
"""
用户选了完整搜索时,第一页凑齐也要翻完再择优。
"""
calls = pages(owner, monkeypatch, {0: [torrent(1)], 1: [torrent(1, 2)]})
chosen = []
scan = SearchScan(owner=owner, plan=plan([1]), full=True, complete_scan=True)
scan.run(lambda items, ready: chosen.extend(item.torrent_info.description for item in items) or ready)
assert calls == [0, 1, 2]
assert set(chosen) == {"1-1", "1-2"}
def test_only_long_waits_left_waits_instead_of_failing(owner, monkeypatch, scan_clock):
"""
唯一的站只是访问次数暂时用完(不是出错)时,等它恢复,而不是把本轮判成搜索失败。
"""
retry_at = (scan_clock["now"] + timedelta(minutes=20)).isoformat(timespec="seconds")
monkeypatch.setattr(SearchSources, "fetch", lambda *_args, **_kwargs: ScanPage(
[], SiteSearchObservation(error="站点设置的访问次数已用完"), retry_at, "rate_limit"))
scan = SearchScan(owner=owner, plan=plan([1]))
scan.run(lambda _items, ready: ready)
scan.run(lambda _items, ready: ready)
assert not scan.ended
assert scan.retry["1:Show"] == retry_at
assert scan.collection.sources["1:Show"].failed is False
def test_first_pass_is_not_repeated_after_checkpoint_resume(owner, monkeypatch):
"""
第一页先交只发生一次并随检查点保存;续跑后深页资源按正常收尾择优,不会被单独提前提交。
"""
scan = SearchScan(owner=owner, plan=plan([1, 2]))
scan.first_pass_done = True
snapshot = SearchSessionSnapshot("round", 0, encode_search_state(scan.state()))
assert SearchScan(owner=owner, plan=plan([1, 2]), snapshot=snapshot).first_pass_done is True
state = scan.state()
state.pop("first_pass_done")
legacy = SearchSessionSnapshot("round", 0, encode_search_state(state))
assert SearchScan(owner=owner, plan=plan([1, 2]), snapshot=legacy).first_pass_done is False
def test_pack_first_keeps_looking_for_pack_when_first_page_only_has_singles(owner, monkeypatch):
"""
首次下载优先整包时,第一页凑齐全套散集也继续找,另一个站深页的整季包仍参与选择。
"""
monkeypatch.setattr(owner, "_sync_indexers", lambda _sites: [{"id": 1, "name": "A"}, {"id": 2, "name": "B"}])
season_plan = plan([])
season_plan.no_exists["tmdb:1"][1].total_episode = 3
pack = TorrentInfo(site=2, description="complete", title="Show 2026 S01 Complete 1080p",
enclosure="https://site.example/download?id=complete")
data = {"1:Show": {0: [torrent(1), torrent(2), torrent(3)]}, "2:Show": {0: [torrent(1, 2)], 1: [pack]}}
calls = []
def fetch(_sources, key, page, **_params):
"""
站点 A 第一页有全套散集,站点 B 第二页才有整季包。
"""
calls.append((key, page))
items = data.get(key, {}).get(page, [])
return ScanPage(items, SiteSearchObservation(True, "success", raw_count=len(items), has_more=bool(items)))
monkeypatch.setattr(SearchSources, "fetch", fetch)
submitted = []
scan = SearchScan(owner=owner, plan=season_plan, full=True, prefer_complete=True, complete_scan=False)
scan.run(lambda items, ready: submitted.append({item.torrent_info.description for item in items}) or ready)
assert ("2:Show", 1) in calls
assert any("complete" in batch for batch in submitted)

View File

@@ -311,3 +311,19 @@ def test_same_numeric_id_from_different_source_is_rejected(candidate_source: Med
assert context.media_info is candidate
assert context.candidate_recognized is True
assert context.media_info_is_target is False
def test_unrecognized_no_year_alias_resource_matches_subscription():
"""
资源自身识别不到作品时没有反证,无年份英文别名也按订阅目标处理,避免资源一直下不下来。
"""
from app.chain.subscribe.match import _is_title_match_allowed
from app.domain.metainfo import MetaInfo
target = MediaInfo(media_source=MediaSource.TMDB, media_id="96102", tmdb_id=96102, type=MediaType.TV,
title="机智的医生生活", original_title="슬기로운 의사생활", names=["Hospital Playlist"],
year="2020", season_years={1: "2020", 2: "2021"})
title = "Hospital Playlist S02 1080p NF WEB-DL DDP2.0 x264-DoA"
meta = MetaInfo(title=title)
assert _is_title_match_allowed(mediainfo=target, torrent_meta=meta,
torrent_info=TorrentInfo(title=title, site_name="测试站点"))

View File

@@ -126,7 +126,7 @@ def test_paged_search_keeps_new_missing_episodes_in_completion(subscription, mon
return items
monkeypatch.setattr(search, "search_site_torrents", request)
_process_paged_subscription(scope.chain, scope.initial, search, target, SystemConfigKey.SubscribeFilterRuleGroups, None)
assert calls == [0, 1]
assert calls == [0]
scope.download.download_single.assert_called_once()
scope.finished.assert_not_called()
assert _missing(scope, scope.current["value"])["tmdb:1"][1].episodes == [2]

View File

@@ -365,4 +365,4 @@ def test_global_identity_remains_valid_cross_type_evidence(conflict_case: Simple
)
assert matched is True
assert "tt1234567" in evidence
assert evidence == "识别结果就是订阅的作品"

View File

@@ -160,7 +160,7 @@ def test_manual_notification_and_old_context_keep_no_source(monkeypatch):
def test_batch_reads_persisted_activity_counts_and_safe_items(tmp_path):
"""
真实异步数据库投影按 task_id 关联活动,列表和详情计数一致并保留完整权限检查。
真实异步数据库投影按 task_id 关联活动,列表、详情和订阅卡片计数一致并保留完整权限检查。
"""
async def verify():
"""
@@ -179,6 +179,10 @@ def test_batch_reads_persisted_activity_counts_and_safe_items(tmp_path):
session.add(SubscriptionSearchTask(task_id=task_id, batch_id=batch_id, subscription_id=17, source="manual",
priority=1, position=0, state="completed", phase="completed",
attempt_count=1, cancel_requested=False, created_at="now", updated_at="now"))
session.add(SubscriptionSearchTask(task_id=uuid4().hex, batch_id=uuid4().hex, subscription_id=18,
source="manual", priority=1, position=0, state="completed",
phase="completed", attempt_count=1, cancel_requested=False,
created_at="now", updated_at="now"))
run = SubscriptionActivityRun(_subscription(note=[1], best_version_full=1), "search", "manual", task_id)
context = _context([1, 2])
annotate_subscription_download(run.subscribe, context)
@@ -198,6 +202,9 @@ def test_batch_reads_persisted_activity_counts_and_safe_items(tmp_path):
assert details.downloads[0]["subscription_download"] == context.subscription_download
assert details.downloads_omitted == 0
assert "private" not in str(details)
statuses = await service.for_subscriptions((17, 18))
assert statuses[17].submitted_count == 1
assert statuses[18].submitted_count == 0
finally:
await engine.dispose()
asyncio.run(verify())

View File

@@ -453,7 +453,7 @@ 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

View File

@@ -431,7 +431,7 @@ def test_deferred_site_cooldown_blocks_other_subscriptions_without_failure_growt
waiting = repository.claim_site(site_id=30, owner="worker-b", lease_seconds=900)
assert waiting.acquired is False
assert waiting.wait_reason == "cooldown"
assert waiting.wait_reason == "rate_limit"
assert waiting.retry_at == future
with Session(engine) as session:
record = session.execute(
@@ -924,7 +924,7 @@ def test_retry_provider_only_searches_pending_enabled_sites(monkeypatch):
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
@@ -933,7 +933,74 @@ def test_deferral_reports_cooldown_separately_from_busy():
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:
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
def test_claim_tells_failing_site_from_normal_request_interval(tmp_path):
"""
出错后的暂停报告为 failing 并带上失败原因,正常请求间隔仍报告为 cooldown。
"""
repository, _engine = _repository(tmp_path)
future = (datetime.now(timezone.utc) + timedelta(minutes=30)).isoformat(timespec="seconds")
claim = repository.claim_site(site_id=40, owner="worker-a", lease_seconds=900)
assert repository.finish_site(site_id=40, lease_token=claim.lease_token, outcome="error",
next_allowed_at=future, error="返回登录或权限提示页") is True
failing = repository.claim_site(site_id=40, owner="worker-b", lease_seconds=900)
assert failing.acquired is False
assert failing.wait_reason == "failing"
assert failing.last_error == "返回登录或权限提示页"
assert failing.consecutive_failures == 1
claim = repository.claim_site(site_id=41, owner="worker-a", lease_seconds=900)
assert repository.finish_site(site_id=41, lease_token=claim.lease_token, outcome="success",
next_allowed_at=future) is True
paced = repository.claim_site(site_id=41, owner="worker-b", lease_seconds=900, minimum_interval=10)
assert paced.acquired is False
assert paced.wait_reason == "cooldown"
class _FailingSiteRepository(_WaitingRepository):
"""
模拟登录失效、连续出错后暂停访问的站点。
"""
def claim_site(self, *, site_id: int, owner: str, lease_seconds: int) -> SiteBudgetClaim:
"""
返回出错暂停中的预算。
"""
del owner, lease_seconds
return SiteBudgetClaim(
site_id=site_id, acquired=False,
retry_at=(datetime.now(timezone.utc) + timedelta(hours=1)).isoformat(timespec="seconds"),
consecutive_failures=30, wait_reason="failing", last_error="返回登录或权限提示页",
)
def test_failing_site_is_skipped_without_deferring_subscription(monkeypatch):
"""
出错暂停中的站点这一轮直接跳过,不让整个订阅重新排队等它;连续失败多时每天只警告一次。
"""
from app.chain.search import provider as provider_module
warnings = []
monkeypatch.setattr(provider_module.logger, "warning", lambda message, *_args, **_kwargs: warnings.append(message))
monkeypatch.setattr(provider_module, "_site_failing_noticed_at", {})
budget = SubscriptionSiteBudget(repository=_FailingSiteRepository(), owner="worker", cancelled=lambda: False,
stop_state=ProcessStopState())
chain = object.__new__(SearchChain)
chain.configure_subscription_site_budget(budget)
chain.search_site_torrents = lambda **_kwargs: ["unexpected"]
for _ in range(2):
# 直接覆盖带预算的站点搜索入口,隔离上层订阅编排。
result = chain._search_site_torrents_with_budget( # pylint: disable=protected-access
site={"id": 16, "name": "我堡"}, keyword="show", mtype=None, page=0,
)
assert result == []
assert chain.consume_subscription_site_budget_deferrals() == ()
assert len(warnings) == 1
assert "我堡" in warnings[0] and "连续 30 次" in warnings[0] and "返回登录或权限提示页" in warnings[0]