diff --git a/app/application/search/session.py b/app/application/search/session.py index 3070ab353..a2ab94462 100644 --- a/app/application/search/session.py +++ b/app/application/search/session.py @@ -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) diff --git a/app/application/subscription/execution.py b/app/application/subscription/execution.py index 7a0063226..b731329a5 100644 --- a/app/application/subscription/execution.py +++ b/app/application/subscription/execution.py @@ -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( diff --git a/app/application/subscription/sitebudget.py b/app/application/subscription/sitebudget.py index a42ac1319..28d7f602b 100644 --- a/app/application/subscription/sitebudget.py +++ b/app/application/subscription/sitebudget.py @@ -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: diff --git a/app/application/subscription/status.py b/app/application/subscription/status.py index 4b693bb20..65c1454f6 100644 --- a/app/application/subscription/status.py +++ b/app/application/subscription/status.py @@ -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 diff --git a/app/application/torrent/download.py b/app/application/torrent/download.py index 2b517226a..f232f62f6 100644 --- a/app/application/torrent/download.py +++ b/app/application/torrent/download.py @@ -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, diff --git a/app/chain/search/provider.py b/app/chain/search/provider.py index 4ff2a38ee..896371eae 100644 --- a/app/chain/search/provider.py +++ b/app/chain/search/provider.py @@ -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( diff --git a/app/chain/search/result.py b/app/chain/search/result.py index 0a94c2088..41e9c044e 100644 --- a/app/chain/search/result.py +++ b/app/chain/search/result.py @@ -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 diff --git a/app/chain/search/scan.py b/app/chain/search/scan.py index 6504c8b1b..2dc6e2d39 100644 --- a/app/chain/search/scan.py +++ b/app/chain/search/scan.py @@ -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() diff --git a/app/chain/subscribe/match.py b/app/chain/subscribe/match.py index 0b3378a79..47a00ed1d 100644 --- a/app/chain/subscribe/match.py +++ b/app/chain/subscribe/match.py @@ -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: diff --git a/app/chain/subscribe/search.py b/app/chain/subscribe/search.py index fd1505bfe..f06f633d3 100644 --- a/app/chain/subscribe/search.py +++ b/app/chain/subscribe/search.py @@ -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 diff --git a/app/db/adapters/subscriptionsearch.py b/app/db/adapters/subscriptionsearch.py index 027b9864d..1763bd220 100644 --- a/app/db/adapters/subscriptionsearch.py +++ b/app/db/adapters/subscriptionsearch.py @@ -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) diff --git a/app/db/adapters/subscriptionstatus.py b/app/db/adapters/subscriptionstatus.py index e1e2d8a22..90517a6df 100644 --- a/app/db/adapters/subscriptionstatus.py +++ b/app/db/adapters/subscriptionstatus.py @@ -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]: """ diff --git a/app/domain/search.py b/app/domain/search.py index 56e39bfd8..0ff2c4b2c 100644 --- a/app/domain/search.py +++ b/app/domain/search.py @@ -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]: """ 完整模式等待所有来源结束；智能模式可按关闭证据或站点未收录判断提前就绪。 diff --git a/app/schemas/subscribe.py b/app/schemas/subscribe.py index 945ab812d..971ce6a4c 100644 --- a/app/schemas/subscribe.py +++ b/app/schemas/subscribe.py @@ -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) diff --git a/scripts/validation/subscription_governance_scale.py b/scripts/validation/subscription_governance_scale.py index 9a8783562..bc8adb26e 100644 --- a/scripts/validation/subscription_governance_scale.py +++ b/scripts/validation/subscription_governance_scale.py @@ -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 diff --git a/tests/test_search_query_guard.py b/tests/test_search_query_guard.py index eced41916..4cd1fed91 100644 --- a/tests/test_search_query_guard.py +++ b/tests/test_search_query_guard.py @@ -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 diff --git a/tests/test_search_same_name_disambiguation.py b/tests/test_search_same_name_disambiguation.py index 9efb23b36..f73e6be36 100644 --- a/tests/test_search_same_name_disambiguation.py +++ b/tests/test_search_same_name_disambiguation.py @@ -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 diff --git a/tests/test_search_scan.py b/tests/test_search_scan.py index ce937a53a..7581bfb1b 100644 --- a/tests/test_search_scan.py +++ b/tests/test_search_scan.py @@ -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) diff --git a/tests/test_subscribe_media_id_conflict.py b/tests/test_subscribe_media_id_conflict.py index b62040c88..5da840f69 100644 --- a/tests/test_subscribe_media_id_conflict.py +++ b/tests/test_subscribe_media_id_conflict.py @@ -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="测试站点")) diff --git a/tests/test_subscribe_paged_search.py b/tests/test_subscribe_paged_search.py index ca7fa6ae2..ffa6c3825 100644 --- a/tests/test_subscribe_paged_search.py +++ b/tests/test_subscribe_paged_search.py @@ -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] diff --git a/tests/test_subscribe_type_conflict.py b/tests/test_subscribe_type_conflict.py index 78c32e9d7..9909a3f81 100644 --- a/tests/test_subscribe_type_conflict.py +++ b/tests/test_subscribe_type_conflict.py @@ -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 == "识别结果就是订阅的作品" diff --git a/tests/test_subscription_download_source.py b/tests/test_subscription_download_source.py index 3fc8ef7f4..e79b3bc06 100644 --- a/tests/test_subscription_download_source.py +++ b/tests/test_subscription_download_source.py @@ -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()) diff --git a/tests/test_subscription_search_governance.py b/tests/test_subscription_search_governance.py index ca7e3bd6a..cd3aa1c19 100644 --- a/tests/test_subscription_search_governance.py +++ b/tests/test_subscription_search_governance.py @@ -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 diff --git a/tests/test_subscription_site_budget.py b/tests/test_subscription_site_budget.py index 9490075a4..72a026ad1 100644 --- a/tests/test_subscription_site_budget.py +++ b/tests/test_subscription_site_budget.py @@ -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]