diff --git a/app/application/chain/context.py b/app/application/chain/context.py index ac68d0110..5d8e40935 100644 --- a/app/application/chain/context.py +++ b/app/application/chain/context.py @@ -12,6 +12,7 @@ from app.application.configuration import ChainRuntimeConfig from app.runtime.stop import StopState, runtime_stop_state if TYPE_CHECKING: + from app.application.download.admission import SubscriptionDownloadRepository from app.application.download.failures import DownloadFailureRepository from app.application.history import ( DownloadHistoryRepository, @@ -29,11 +30,11 @@ if TYPE_CHECKING: DeleteSubscribeScope, SyncDeleteSubscribeScope, ) + from app.application.subscription.execution import SubscriptionSearchRepository from app.application.subscription.mutation import ( SubscriptionMutationScope, SyncSubscriptionMutationScope, ) - from app.application.subscription.execution import SubscriptionSearchRepository from app.application.transfer.execution import TransferExecutionRepository from app.application.transfer.workflow import TransferAdmissionRepository @@ -76,6 +77,7 @@ class ChainRuntimeContext: download_failure_repository: DownloadFailureRepository user_repository: ChainUserRepository subscription_search_repository: Optional[SubscriptionSearchRepository] = None + subscription_download_repository: Optional[SubscriptionDownloadRepository] = None legacy_transfer_command: Optional[LegacyTransferCommand] = None durable_event_writer: Optional[ChainDurableEventWriter] = None configuration: ChainRuntimeConfig = field( diff --git a/app/application/download/admission.py b/app/application/download/admission.py new file mode 100644 index 000000000..2e4665316 --- /dev/null +++ b/app/application/download/admission.py @@ -0,0 +1,124 @@ +"""订阅下载提交的持久幂等与不确定终态合同。""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Callable, Optional, Protocol + + +class DownloadReconciliationRequired(RuntimeError): + """表示下载器可能已接受任务,必须先对账才能继续自动提交。""" + + +@dataclass(frozen=True, slots=True) +class SubscriptionDownloadGovernance: + """把订阅身份、入口任务与取消检查传入下载提交边界。""" + + subscription_id: int + mode: str + task_id: Optional[str] = None + cancelled: Optional[Callable[[], bool]] = None + mark_started: Optional[Callable[[], None]] = None + + +@dataclass(frozen=True, slots=True) +class SubscriptionDownloadRequest: + """一次订阅下载提交认领所需的规范身份。""" + + idempotency_key: str + subscription_id: int + task_id: Optional[str] + logical_identity: str + resource_key: str + coverage: str + mode: str + + +@dataclass(frozen=True, slots=True) +class SubscriptionDownloadSnapshot: + """脱离 Session 的订阅下载提交状态快照。""" + + idempotency_key: str + subscription_id: int + task_id: Optional[str] + state: str + attempt_count: int + attempt_token: Optional[str] + downloader: Optional[str] + download_hash: Optional[str] + available_at: Optional[str] + last_error: Optional[str] + created_at: str + updated_at: str + + +@dataclass(frozen=True, slots=True) +class SubscriptionDownloadClaim: + """返回本次是否取得唯一提交权以及当前持久状态。""" + + acquired: bool + snapshot: SubscriptionDownloadSnapshot + + +class SubscriptionDownloadRepository(Protocol): + """订阅下载幂等账本所需的最小持久化端口。""" + + def claim(self, request: SubscriptionDownloadRequest) -> SubscriptionDownloadClaim: + """按唯一键认领提交;仅到期 retryable/cancelled 状态允许重新认领。""" + ... + + def mark_accepted( + self, + *, + idempotency_key: str, + attempt_token: str, + downloader: Optional[str], + download_hash: str, + ) -> bool: + """记录下载器已明确接受任务,后续任何失败都不得自动重试。""" + ... + + def mark_succeeded( + self, + *, + idempotency_key: str, + attempt_token: str, + ) -> bool: + """在 canonical 本地结算完成后写入成功终态。""" + ... + + def mark_retryable( + self, + *, + idempotency_key: str, + attempt_token: str, + available_at: str, + error: Optional[str], + ) -> bool: + """记录下载器明确拒绝、尚未产生外部副作用的可重试状态。""" + ... + + def mark_reconcile_required( + self, + *, + idempotency_key: str, + attempt_token: str, + error: Optional[str], + downloader: Optional[str] = None, + download_hash: Optional[str] = None, + ) -> bool: + """冻结可能已产生外部副作用的提交,等待下载器对账。""" + ... + + def mark_cancelled( + self, + *, + idempotency_key: str, + attempt_token: str, + ) -> bool: + """仅在进入下载器副作用前收口取消。""" + ... + + def has_started_for_task(self, task_id: str) -> bool: + """判断搜索任务是否已有不能按未执行处理的下载提交。""" + ... diff --git a/app/application/subscription/candidates.py b/app/application/subscription/candidates.py index 814cfd014..5cb1b9c24 100644 --- a/app/application/subscription/candidates.py +++ b/app/application/subscription/candidates.py @@ -3,15 +3,14 @@ import copy from dataclasses import dataclass, field from datetime import datetime, timezone -from typing import Dict, List, Optional +from typing import Any, Dict, List, Optional, cast from uuid import uuid4 from app.application.subscription.contract import SubscriptionSnapshot from app.domain.context import Context, MediaInfo from app.foundation import text as text_tools from app.schemas.media import resolve_media_identity -from app.schemas.types import MediaType - +from app.schemas.types import MediaSource, MediaType CandidateGroups = Dict[str, List[Context]] @@ -196,11 +195,11 @@ class CandidateIndex: context.candidate_recognized = False context.media_info_is_target = True context.media_info = MediaInfo( - type=subscribe.type, + type=cast(MediaType, subscribe.type), title=subscribe.name, - media_source=subscribe.media_source, - media_id=subscribe.media_id, - season=subscribe.season, + media_source=cast(MediaSource, subscribe.media_source), + media_id=cast(str, subscribe.media_id), + season=cast(int, subscribe.season), ) @classmethod @@ -236,7 +235,7 @@ class CandidateIndex: return media_season is None or target_season == media_season @classmethod - def meta_seasons(cls, meta_info) -> set[int]: + def meta_seasons(cls, meta_info: Any) -> set[int]: """提取标题解析出的显式季范围。""" meta_fields = vars(meta_info) if meta_info else {} if "season_list" in meta_fields: @@ -271,7 +270,7 @@ class CandidateIndex: return {identity for identity in identities if identity} @staticmethod - def media_identity(media) -> Optional[tuple[str, str]]: + def media_identity(media: Any) -> Optional[tuple[str, str]]: """把动态媒体对象的身份归一为可索引键。""" source, media_id = resolve_media_identity(media=media) if not source or not media_id: @@ -279,7 +278,7 @@ class CandidateIndex: return str(source), media_id @staticmethod - def normalize_int(value) -> Optional[int]: + def normalize_int(value: Any) -> Optional[int]: """将季号等动态字段转为整数。""" if value is None: return None @@ -289,15 +288,15 @@ class CandidateIndex: return None @staticmethod - def normalize_media_type(value) -> Optional[str]: + def normalize_media_type(value: Any) -> Optional[str]: """统一媒体类型枚举与字符串形态。""" if isinstance(value, MediaType): value = value.value if value == MediaType.UNKNOWN.value: return None - return value + return str(value) if value is not None else None @staticmethod - def normalize_title(value) -> str: + def normalize_title(value: Any) -> str: """归一标题用于低置信标题匹配。""" return (text_tools.normalize_upper(value or "") or "").strip() diff --git a/app/application/subscription/sitebudget.py b/app/application/subscription/sitebudget.py index e89f4a5b7..7e9c56f86 100644 --- a/app/application/subscription/sitebudget.py +++ b/app/application/subscription/sitebudget.py @@ -148,7 +148,7 @@ class SubscriptionSiteBudget: "login_invalid": (900.0, 21600.0), "timeout": (300.0, 7200.0), }.get(outcome, (180.0, 3600.0)) - return min(base * (2**exponent), ceiling) + return float(min(base * (2**exponent), ceiling)) def _raise_if_cancelled(self) -> None: """在不持有业务锁的等待边界传播取消或停机。""" diff --git a/app/chain/base.py b/app/chain/base.py index 23992b914..c92724073 100644 --- a/app/chain/base.py +++ b/app/chain/base.py @@ -72,6 +72,7 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin, self.transfer_execution_repository = context.transfer_execution_repository self.media_server_repository = context.media_server_repository self.download_failure_repository = context.download_failure_repository + self.subscription_download_repository = context.subscription_download_repository self.user_repository = context.user_repository self.runtime_config = context.configuration self.stop_state = context.stop_state diff --git a/app/chain/download/admission.py b/app/chain/download/admission.py new file mode 100644 index 000000000..51f845d76 --- /dev/null +++ b/app/chain/download/admission.py @@ -0,0 +1,201 @@ +"""订阅下载提交幂等身份与状态转换 owner。""" + +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timedelta, timezone +from typing import Optional, Set, cast + +from app.application.download.admission import ( + DownloadReconciliationRequired, + SubscriptionDownloadClaim, + SubscriptionDownloadGovernance, + SubscriptionDownloadRepository, + SubscriptionDownloadRequest, +) +from app.chain.download.contract import _DownloadOwnerBase +from app.domain import episode as episode_rules +from app.domain.context import Context +from app.schemas.media import build_media_key, resolve_media_identity +from app.schemas.types import MediaType + + +class DownloadAdmissionOwner(_DownloadOwnerBase): + """计算订阅提交唯一键并通过持久账本控制外部副作用。""" + + def _build_subscription_download_request( + self, + *, + context: Context, + episodes: Optional[Set[int]], + governance: SubscriptionDownloadGovernance, + ) -> SubscriptionDownloadRequest: + """组合订阅、torrent、季集覆盖与模式生成规范幂等请求。""" + media = context.media_info + meta = context.meta_info + torrent = context.torrent_info + media_source, media_id = resolve_media_identity(media=media) + media_key = build_media_key(media_source, media_id) or ( + f"{getattr(media, 'title', '')}:{getattr(media, 'year', '')}" + ) + seasons = sorted(set(getattr(meta, "season_list", None) or [])) + selected = sorted(set(episodes or getattr(meta, "episode_list", None) or [])) + if selected: + coverage = f"episodes:{episode_rules.format_ranges(selected)}" + elif seasons: + coverage = "seasons:" + ",".join(str(season) for season in seasons) + ":full" + else: + coverage = "full" + media_type = getattr(media, "type", None) + media_type_value = getattr(media_type, "value", media_type) + media_season = getattr(media, "season", None) + meta_season = getattr(meta, "season", None) + logical_identity = json.dumps( + { + "subscription_id": governance.subscription_id, + "media_key": str(media_key or ""), + "media_type": str(media_type_value or ""), + "season": media_season if media_season is not None else meta_season, + "episode_group": getattr(media, "episode_group", None), + }, + ensure_ascii=False, + sort_keys=True, + ) + resource_key = self._torrent_resource_key(torrent) + canonical = json.dumps( + { + "logical_identity": logical_identity, + "resource_key": resource_key, + "coverage": coverage, + "mode": governance.mode, + }, + ensure_ascii=False, + sort_keys=True, + ) + return SubscriptionDownloadRequest( + idempotency_key=hashlib.sha256(canonical.encode("utf-8")).hexdigest(), + subscription_id=governance.subscription_id, + task_id=governance.task_id, + logical_identity=logical_identity, + resource_key=resource_key, + coverage=coverage, + mode=governance.mode, + ) + + def _claim_subscription_download( + self, + *, + context: Context, + episodes: Optional[Set[int]], + governance: Optional[SubscriptionDownloadGovernance], + ) -> tuple[Optional[SubscriptionDownloadClaim], Optional[str]]: + """在下载器调用前认领唯一提交权,并返回已成功提交的历史 hash。""" + if governance is None: + return None, None + legacy_hash = self._legacy_subscription_download_hash( + context=context, + episodes=episodes, + governance=governance, + ) + if legacy_hash: + return None, legacy_hash + repository = getattr(self, "subscription_download_repository", None) + if repository is None: + raise RuntimeError("订阅下载幂等仓储尚未配置") + request = self._build_subscription_download_request( + context=context, + episodes=episodes, + governance=governance, + ) + claim = repository.claim(request) + snapshot = claim.snapshot + if claim.acquired: + if not snapshot.attempt_token: + raise RuntimeError("订阅下载提交已认领但缺少尝试令牌") + return claim, None + if snapshot.state == "succeeded" and snapshot.download_hash: + return claim, snapshot.download_hash + if snapshot.state in {"submitting", "accepted", "reconcile_required"}: + raise DownloadReconciliationRequired( + f"订阅下载提交 {snapshot.idempotency_key} 当前为 {snapshot.state},需要先对账下载器" + ) + return claim, None + + def _legacy_subscription_download_hash( + self, + *, + context: Context, + episodes: Optional[Set[int]], + governance: SubscriptionDownloadGovernance, + ) -> Optional[str]: + """重读迁移前下载历史,兼容识别同订阅同 torrent 同覆盖的成功提交。""" + media = context.media_info + meta = context.meta_info + torrent = context.torrent_info + repository = getattr(self, "download_history_repository", None) + media_source, media_id = resolve_media_identity(media=media) + if repository is None or not media_source or not media_id or not torrent: + return None + histories = repository.get_by_media_identity( + media_source=media_source, + media_id=str(media_id), + music_type=getattr(media, "music_type", None), + ) + expected_episodes = episode_rules.format_ranges( + sorted(set(episodes or getattr(meta, "episode_list", None) or [])) + ) + expected_season = str(getattr(meta, "season", None) or "") + for history in histories: + if not history.download_hash: + continue + if history.torrent_name != torrent.title or history.torrent_site != torrent.site_name: + continue + if not self._history_matches_subscription(history.note, governance.subscription_id): + continue + if getattr(history, "episode_group", None) != getattr(media, "episode_group", None): + continue + if expected_episodes: + if history.episodes == expected_episodes: + return cast(str, history.download_hash) + continue + if getattr(media, "type", None) == MediaType.TV and str(history.seasons or "") != expected_season: + continue + return cast(str, history.download_hash) + return None + + @staticmethod + def _history_matches_subscription(note: object, subscription_id: int) -> bool: + """从下载历史来源快照确认记录属于同一订阅,而非同媒体其他记录。""" + if not isinstance(note, dict): + return False + source = note.get("source") + if not isinstance(source, str) or not source.startswith("Subscribe|"): + return False + try: + payload = json.loads(source.split("|", 1)[1]) + return int(payload.get("id")) == subscription_id + except (AttributeError, TypeError, ValueError, json.JSONDecodeError): + return False + + @staticmethod + def _subscription_download_cancelled( + governance: Optional[SubscriptionDownloadGovernance], + ) -> bool: + """在可安全取消边界调用入口提供的取消检查。""" + return bool(governance and governance.cancelled and governance.cancelled()) + + @staticmethod + def _subscription_download_retry_at(error: Optional[str], ttl_seconds: int) -> str: + """把明确拒绝的下载提交推迟到失败冷却到期后。""" + del error + return ( + datetime.now(timezone.utc) + timedelta(seconds=max(1, ttl_seconds)) + ).isoformat(timespec="seconds") + + def _subscription_download_repository(self) -> SubscriptionDownloadRepository: + """返回已配置的订阅下载仓储,缺失时拒绝越过外部副作用边界。""" + repository = getattr(self, "subscription_download_repository", None) + if repository is None: + raise RuntimeError("订阅下载幂等仓储尚未配置") + return cast(SubscriptionDownloadRepository, repository) diff --git a/app/chain/download/batch.py b/app/chain/download/batch.py index 292d6b886..2a860f9ff 100644 --- a/app/chain/download/batch.py +++ b/app/chain/download/batch.py @@ -4,6 +4,7 @@ import copy from typing import Callable, Dict, List, Optional, Set, Tuple, cast from app.application.download import selection as _selection +from app.application.download.admission import SubscriptionDownloadGovernance from app.application.torrent.download import TorrentHelper from app.chain.download.contract import _DownloadOwnerBase from app.domain import episode as episode_rules @@ -39,7 +40,8 @@ class DownloadBatchOwner(_DownloadOwnerBase): userid: Optional[str] = None, username: Optional[str] = None, downloader: Optional[str] = None, - custom_words: Optional[str] = None + custom_words: Optional[str] = None, + governance: Optional[SubscriptionDownloadGovernance] = None, ) -> Tuple[ List[Context], Optional[Dict[str, Dict[int, NotExistMediaInfo]]], @@ -60,6 +62,7 @@ class DownloadBatchOwner(_DownloadOwnerBase): username=username, downloader=downloader, custom_words=custom_words, + governance=governance, ) def _execute_batch_download(self, @@ -71,7 +74,8 @@ class DownloadBatchOwner(_DownloadOwnerBase): userid: Optional[str] = None, username: Optional[str] = None, downloader: Optional[str] = None, - custom_words: Optional[str] = None + custom_words: Optional[str] = None, + governance: Optional[SubscriptionDownloadGovernance] = None, ) -> Tuple[ List[Context], Optional[Dict[str, Dict[int, NotExistMediaInfo]]], @@ -87,6 +91,7 @@ class DownloadBatchOwner(_DownloadOwnerBase): :param username: 调用下载的用户名/插件名 :param downloader: 下载器 :param custom_words: 下载来源(如订阅)的完整自定义识别词文本,随下载记录存档,供整理时原样复现识别 + :param governance: 订阅下载幂等、入口任务和取消边界;非订阅调用保持为空 :return: 已下载资源列表及剩余缺集,键格式为 no_exists[source:id] """ no_exists_was_none = no_exists is None @@ -186,6 +191,7 @@ class DownloadBatchOwner(_DownloadOwnerBase): username=username, downloader=downloader, custom_words=custom_words, + governance=governance, ) # 电视剧整季匹配 @@ -303,7 +309,8 @@ class DownloadBatchOwner(_DownloadOwnerBase): userid=userid, username=username, downloader=downloader, - custom_words=custom_words + custom_words=custom_words, + governance=governance, ) else: # 下载 @@ -312,7 +319,8 @@ class DownloadBatchOwner(_DownloadOwnerBase): channel=channel, source=source, userid=userid, username=username, downloader=downloader, - custom_words=custom_words) + custom_words=custom_words, + governance=governance) if download_id: # 下载成功 @@ -401,7 +409,8 @@ class DownloadBatchOwner(_DownloadOwnerBase): channel=channel, source=source, userid=userid, username=username, downloader=downloader, - custom_words=custom_words) + custom_words=custom_words, + governance=governance) if download_id: # 下载成功 if __requires_complete_coverage(missing_info): @@ -518,7 +527,8 @@ class DownloadBatchOwner(_DownloadOwnerBase): userid=userid, username=username, downloader=downloader, - custom_words=custom_words + custom_words=custom_words, + governance=governance, ) if not download_id: __remember_context_failure(context) diff --git a/app/chain/download/contract.py b/app/chain/download/contract.py index 8fb08d2c6..c4f9c5a3f 100644 --- a/app/chain/download/contract.py +++ b/app/chain/download/contract.py @@ -9,6 +9,7 @@ if TYPE_CHECKING: """向类型检查器声明同一 Facade 上的跨 owner 方法。""" _active_download_failure_fingerprints: Callable[..., Any] + _build_subscription_download_request: Callable[..., Any] _append_no_exists: Callable[..., None] _build_download_notification: Callable[..., Any] _build_download_failure_fingerprint: Callable[..., Any] @@ -22,14 +23,20 @@ if TYPE_CHECKING: _is_job_active: Callable[..., bool] _is_subscribe_source: Callable[..., bool] _log_download_failure_cooldown: Callable[..., None] + _legacy_subscription_download_hash: Callable[..., Any] + _history_matches_subscription: Callable[..., bool] _media_identity_keys: Callable[..., set[str]] _matches_media_identity: Callable[..., bool] _prepare_batch_download_contexts: Callable[..., Any] + _claim_subscription_download: Callable[..., Any] _record_download_failure: Callable[..., Any] _resolve_media_download_dir: Callable[..., Any] _settle_download_success: Callable[..., None] _submit_download_added_task: Callable[..., None] _torrent_resource_key: Callable[..., str] + _subscription_download_cancelled: Callable[..., bool] + _subscription_download_repository: Callable[..., Any] + _subscription_download_retry_at: Callable[..., str] _validate_music_album_resource: Callable[..., Any] batch_download: Callable[..., Any] download_single: Callable[..., Any] diff --git a/app/chain/download/facade.py b/app/chain/download/facade.py index 780cde049..1681c68a7 100644 --- a/app/chain/download/facade.py +++ b/app/chain/download/facade.py @@ -3,6 +3,7 @@ from collections.abc import Callable from typing import Any, TypeVar +from app.chain.download.admission import DownloadAdmissionOwner from app.chain.download.batch import DownloadBatchOwner from app.chain.download.existence import DownloadExistenceOwner from app.chain.download.failure import DownloadFailureOwner @@ -29,6 +30,7 @@ class DownloadChain( DownloadSubtitleOwner, DownloadSelectionOwner, DownloadFailureOwner, + DownloadAdmissionOwner, DownloadSubmissionOwner, DownloadBatchOwner, DownloadExistenceOwner, diff --git a/app/chain/download/selection.py b/app/chain/download/selection.py index 1db401615..5ff6365a0 100644 --- a/app/chain/download/selection.py +++ b/app/chain/download/selection.py @@ -4,6 +4,7 @@ from pathlib import Path from typing import Callable, Dict, List, Optional, Set, Tuple, Union, cast from app.application.configuration import get_chain_runtime_config_snapshot +from app.application.download.admission import SubscriptionDownloadGovernance from app.application.download.failures import ( DownloadFailureSnapshot, ) @@ -156,6 +157,7 @@ class DownloadSelectionOwner(_DownloadOwnerBase): username: Optional[str], downloader: Optional[str], custom_words: Optional[str], + governance: Optional[SubscriptionDownloadGovernance], ) -> None: """ 处理电影与音乐的直接候选下载。 @@ -201,6 +203,7 @@ class DownloadSelectionOwner(_DownloadOwnerBase): username=username, downloader=downloader, custom_words=custom_words, + governance=governance, ): logger.info(f"{context.torrent_info.title} 添加下载成功") downloaded_list.append(context) diff --git a/app/chain/download/submission.py b/app/chain/download/submission.py index b35297856..0ede8c0f1 100644 --- a/app/chain/download/submission.py +++ b/app/chain/download/submission.py @@ -9,6 +9,10 @@ from urllib.parse import urlencode, urljoin, urlparse from app.application.configuration import get_chain_runtime_config_snapshot from app.application.directory import validate_download_save_path +from app.application.download.admission import ( + DownloadReconciliationRequired, + SubscriptionDownloadGovernance, +) from app.application.torrent.download import TorrentHelper from app.chain.download.contract import _DownloadOwnerBase from app.chain.download.ports import ( @@ -270,7 +274,9 @@ class DownloadSubmissionOwner(_DownloadOwnerBase): username: Optional[str] = None, label: Optional[str] = None, return_detail: bool = False, - custom_words: Optional[str] = None) -> Union[Optional[str], Tuple[Optional[str], Optional[str]]]: + custom_words: Optional[str] = None, + governance: Optional[SubscriptionDownloadGovernance] = None, + ) -> Union[Optional[str], Tuple[Optional[str], Optional[str]]]: """ 下载单个资源并发送结果通知。 @@ -290,6 +296,7 @@ class DownloadSubmissionOwner(_DownloadOwnerBase): label=label, return_detail=return_detail, custom_words=custom_words, + governance=governance, ) def _execute_download_single(self, context: Context, @@ -304,7 +311,9 @@ class DownloadSubmissionOwner(_DownloadOwnerBase): username: Optional[str] = None, label: Optional[str] = None, return_detail: bool = False, - custom_words: Optional[str] = None) -> Union[Optional[str], Tuple[Optional[str], Optional[str]]]: + custom_words: Optional[str] = None, + governance: Optional[SubscriptionDownloadGovernance] = None, + ) -> Union[Optional[str], Tuple[Optional[str], Optional[str]]]: """ 下载及发送通知 :param context: 资源上下文 @@ -320,6 +329,7 @@ class DownloadSubmissionOwner(_DownloadOwnerBase): :param label: 自定义标签 :param return_detail: 是否返回详细结果;False 时返回下载任务 hash 或 None,True 时返回 (hash, error_msg) :param custom_words: 下载来源(如订阅)的完整自定义识别词文本,随下载记录存档,供整理时原样复现识别 + :param governance: 订阅级提交幂等、入口任务与取消检查;非订阅调用保持为空 :return: return_detail=False 时返回下载任务 hash 或 None;return_detail=True 时返回 (hash, error_msg) """ _torrent = context.torrent_info @@ -408,41 +418,138 @@ class DownloadSubmissionOwner(_DownloadOwnerBase): file_uri = FileURI(storage=storage, path=download_dir.as_posix()) download_dir = Path(file_uri.uri) + if self._subscription_download_cancelled(governance): + cancel_error = "订阅下载在提交前已取消" + return (None, cancel_error) if return_detail else None + admission, duplicate_hash = self._claim_subscription_download( + context=context, + episodes=episodes, + governance=governance, + ) + if duplicate_hash: + if governance and governance.mark_started: + governance.mark_started() + logger.info(f"{_torrent.title} 已由重叠订阅入口提交,复用任务 {duplicate_hash}") + return (duplicate_hash, "下载任务已由重叠入口提交") if return_detail else duplicate_hash + if admission is not None and not admission.acquired: + wait_error = ( + f"订阅下载提交当前为 {admission.snapshot.state}," + f"最早可重试:{admission.snapshot.available_at or '待下一轮'}" + ) + return (None, wait_error) if return_detail else None + attempt_token = admission.snapshot.attempt_token if admission is not None else None + admission_key = admission.snapshot.idempotency_key if admission is not None else None + if admission is not None and self._subscription_download_cancelled(governance): + self._subscription_download_repository().mark_cancelled( + idempotency_key=admission.snapshot.idempotency_key, + attempt_token=admission.snapshot.attempt_token or "", + ) + cancel_error = "订阅下载在下载器调用前已取消" + return (None, cancel_error) if return_detail else None + # 添加下载 - result: Optional[Tuple[Optional[str], Optional[str], Optional[str], str]] = self.download(content=torrent_content, - cookie=_torrent.site_cookie, - episodes=cast(Set[int], episodes), - download_dir=download_dir, - category=_media.category, - label=label, - downloader=downloader or _site_downloader) + if governance and governance.mark_started: + governance.mark_started() + try: + result: Optional[Tuple[Optional[str], Optional[str], Optional[str], str]] = self.download( + content=torrent_content, + cookie=_torrent.site_cookie, + episodes=cast(Set[int], episodes), + download_dir=download_dir, + category=_media.category, + label=label, + downloader=downloader or _site_downloader, + ) + except Exception as err: + if admission_key and attempt_token: + self._subscription_download_repository().mark_reconcile_required( + idempotency_key=admission_key, + attempt_token=attempt_token, + error=f"下载器调用异常:{str(err)}", + ) + raise DownloadReconciliationRequired( + f"{_torrent.title} 下载器结果不确定,已冻结自动重试" + ) from err + raise if result: _downloader, _hash, _layout, error_msg = result else: _downloader, _hash, _layout, error_msg = None, None, None, "未找到下载器" if _hash: - self._settle_download_success( - context=context, - media=_media, - meta=_meta, - torrent=_torrent, - folder_name=_folder_name, - file_list=_file_list, - download_dir=download_dir, - layout=_layout, - downloader=_downloader, - download_hash=_hash, - download_episodes=download_episodes, - episodes=episodes, - channel=channel, - source=source, - userid=userid, - username=username, - torrent_content=torrent_content, - custom_words=custom_words, - ) + if admission_key and attempt_token: + accepted = self._subscription_download_repository().mark_accepted( + idempotency_key=admission_key, + attempt_token=attempt_token, + downloader=_downloader, + download_hash=_hash, + ) + if not accepted: + raise DownloadReconciliationRequired( + f"{_torrent.title} 已被下载器接受,但本地接受状态写入失败" + ) + try: + self._settle_download_success( + context=context, + media=_media, + meta=_meta, + torrent=_torrent, + folder_name=_folder_name, + file_list=_file_list, + download_dir=download_dir, + layout=_layout, + downloader=_downloader, + download_hash=_hash, + download_episodes=download_episodes, + episodes=episodes, + channel=channel, + source=source, + userid=userid, + username=username, + torrent_content=torrent_content, + custom_words=custom_words, + ) + except Exception as err: + if admission_key and attempt_token: + self._subscription_download_repository().mark_reconcile_required( + idempotency_key=admission_key, + attempt_token=attempt_token, + error=f"下载器已接受但本地结算失败:{str(err)}", + downloader=_downloader, + download_hash=_hash, + ) + raise DownloadReconciliationRequired( + f"{_torrent.title} 已被下载器接受但本地结算失败,已转待对账" + ) from err + raise + if admission_key and attempt_token: + succeeded = self._subscription_download_repository().mark_succeeded( + idempotency_key=admission_key, + attempt_token=attempt_token, + ) + if not succeeded: + self._subscription_download_repository().mark_reconcile_required( + idempotency_key=admission_key, + attempt_token=attempt_token, + error="本地结算完成但幂等成功终态写入失败", + downloader=_downloader, + download_hash=_hash, + ) + raise DownloadReconciliationRequired( + f"{_torrent.title} 本地结算完成但提交终态未确认,已转待对账" + ) else: + if admission_key and attempt_token: + retry_at = self._subscription_download_retry_at( + error_msg, + self._download_failure_ttl(error_msg), + ) + self._subscription_download_repository().mark_retryable( + idempotency_key=admission_key, + attempt_token=attempt_token, + available_at=retry_at, + error=error_msg, + ) # 下载失败 logger.error(f"{_media.title_year} 添加下载任务失败:" f"{_torrent.title} - {_torrent.enclosure},{error_msg}") diff --git a/app/chain/search/facade.py b/app/chain/search/facade.py index 3ad76144d..1b2d19549 100644 --- a/app/chain/search/facade.py +++ b/app/chain/search/facade.py @@ -4,6 +4,7 @@ import threading from collections.abc import AsyncIterator, Callable from typing import Any, Optional, TypeVar, cast +from app.application.subscription.sitebudget import SubscriptionSiteBudget from app.chain.base import ChainBase from app.chain.search.cache import SearchCacheOwner from app.chain.search.media import SearchMediaOwner @@ -16,7 +17,6 @@ from app.chain.search.result import SearchResultOwner from app.chain.search.site import SearchSiteOwner from app.chain.search.subtitle import SearchSubtitleOwner from app.chain.search.title import SearchTitleOwner -from app.application.subscription.sitebudget import SubscriptionSiteBudget from app.domain.context import Context, MediaInfo, SubtitleInfo from app.runtime.events import Event, eventmanager from app.schemas.mediaserver import NotExistMediaInfo diff --git a/app/chain/search/result.py b/app/chain/search/result.py index f6f6e415e..6740b7ce3 100644 --- a/app/chain/search/result.py +++ b/app/chain/search/result.py @@ -203,8 +203,7 @@ def _match_torrents( def _context_media(mediainfo: MediaInfo) -> MediaInfo: """复制并裁剪上下文媒体信息,避免修改调用方持有的目标对象。""" context_media = copy.copy(mediainfo) - clear_context_media = cast(Callable[[], None], context_media.clear) - clear_context_media() + context_media.clear() return context_media diff --git a/app/chain/subscribe/completion.py b/app/chain/subscribe/completion.py index 7b21c0628..7944c1639 100644 --- a/app/chain/subscribe/completion.py +++ b/app/chain/subscribe/completion.py @@ -189,7 +189,7 @@ class SubscribeCompletionOwner(_SubscribeOwnerBase): mediainfo=mediainfo, mediakey=mediakey, ) - return completed + return bool(completed) def _SubscribeChain__update_subscribe_note( self, diff --git a/app/chain/subscribe/policy.py b/app/chain/subscribe/policy.py index b9cd18c8a..253b3e606 100644 --- a/app/chain/subscribe/policy.py +++ b/app/chain/subscribe/policy.py @@ -1,8 +1,9 @@ """订阅优先级、剧集范围与来源编码策略""" import json -from typing import Any, Dict, List, Optional, Tuple, Union, cast +from typing import Any, Callable, Dict, List, Optional, Tuple, Union, cast +from app.application.download.admission import SubscriptionDownloadGovernance from app.application.subscription import priority as _priority from app.application.subscription.contract import ( SubscriptionSnapshot, @@ -258,6 +259,14 @@ class SubscribePolicyOwner(_SubscribeOwnerBase): ) return result + @staticmethod + def _SubscribeChain__load_current_subscription( + loader: Callable[[int], Optional[SubscriptionSnapshot]], + subscription_id: int, + ) -> Optional[SubscriptionSnapshot]: + """通过显式可调用边界读取提交前的最新订阅快照。""" + return loader(subscription_id) + def _SubscribeChain__download_best_version_with_full_pack_first( self, contexts: List[Context], @@ -272,6 +281,60 @@ class SubscribePolicyOwner(_SubscribeOwnerBase): """ TV 分集洗版先尝试覆盖目标范围的全集资源,失败后回退到按集下载。 """ + governance: Optional[SubscriptionDownloadGovernance] = None + repository = getattr(self, "subscription_repository", None) + get_current = getattr(repository, "get", None) + if callable(get_current): + current_get = cast( + Callable[[int], Optional[SubscriptionSnapshot]], + get_current, + ) + current = self._SubscribeChain__load_current_subscription( + current_get, + subscribe.id, + ) + if current is None: + logger.info(f"订阅 {subscribe.id} 已删除,放弃本轮下载提交") + return [], no_exists + if current.state == "S": + logger.info(f"订阅 {current.name} 已暂停,放弃本轮下载提交") + return [], no_exists + if self._SubscribeChain__candidate_contract_changed(subscribe, current): + logger.info(f"订阅 {current.name} 的筛选或媒体身份已变化,放弃旧候选并等待下一轮") + return [], no_exists + if not contexts or not contexts[0].meta_info or not contexts[0].media_info: + return [], no_exists + + exists, fresh_no_exists = self.check_and_handle_existing_media( + subscribe=current, + meta=contexts[0].meta_info, + mediainfo=contexts[0].media_info, + mediakey=mediakey, + ) + if exists: + return [], fresh_no_exists + contexts = self._SubscribeChain__revalidate_download_contexts(current, contexts) + if not contexts: + return [], fresh_no_exists + subscribe = current + no_exists = fresh_no_exists + username = current.username + save_path = current.save_path + downloader = current.downloader + source = self.get_subscribe_source_keyword(current) + governance = SubscriptionDownloadGovernance( + subscription_id=current.id, + mode=self._SubscribeChain__download_governance_mode(current), + task_id=cast(Optional[str], getattr(self, "_subscription_download_task_id", None)), + cancelled=cast( + Optional[Callable[[], bool]], + getattr(self, "_subscription_download_cancelled", None), + ), + mark_started=cast( + Optional[Callable[[], None]], + getattr(self, "_subscription_download_mark_started", None), + ), + ) full_pack_no_exists = self._SubscribeChain__build_full_pack_first_no_exists( subscribe=subscribe, mediakey=mediakey ) @@ -316,6 +379,7 @@ class SubscribePolicyOwner(_SubscribeOwnerBase): downloader=downloader, source=source, custom_words=subscribe.custom_words, + governance=governance, ) if downloads: return downloads, lefts @@ -330,10 +394,84 @@ class SubscribePolicyOwner(_SubscribeOwnerBase): downloader=downloader, source=source, custom_words=subscribe.custom_words, + governance=governance, ) ) return result + @staticmethod + def _SubscribeChain__candidate_contract_changed( + prepared: SubscriptionSnapshot, + current: SubscriptionSnapshot, + ) -> bool: + """检测会让准备阶段候选失效的订阅身份和筛选字段变化。""" + fields = ( + "type", + "media_source", + "media_id", + "music_type", + "season", + "episode_group", + "keyword", + "sites", + "include", + "exclude", + "quality", + "resolution", + "effect", + "audio_quality", + "audio_format", + "min_bitrate", + "min_bit_depth", + "min_sample_rate", + "filter_groups", + "custom_words", + ) + return any(getattr(prepared, field) != getattr(current, field) for field in fields) + + def _SubscribeChain__revalidate_download_contexts( + self, + subscribe: SubscriptionSnapshot, + contexts: List[Context], + ) -> List[Context]: + """按当前洗版模式和优先级重新过滤准备阶段候选。""" + if not subscribe.best_version: + return contexts + accepted: List[Context] = [] + for context in contexts: + media = context.media_info + meta = context.meta_info + torrent = context.torrent_info + if not media or not meta or not torrent: + continue + if media.type == MediaType.TV: + if self._SubscribeChain__is_full_best_version_enabled(subscribe) \ + and not self._SubscribeChain__is_full_season_resource(meta, subscribe): + continue + if not self._is_episode_range_covered(meta, subscribe): + continue + if not self._SubscribeChain__prepare_best_version_tv_candidate( + subscribe=subscribe, + context=context, + priority=torrent.pri_order, + ): + continue + elif subscribe.current_priority and torrent.pri_order <= subscribe.current_priority: + continue + accepted.append(context) + return accepted + + @staticmethod + def _SubscribeChain__download_governance_mode(subscribe: SubscriptionSnapshot) -> str: + """把订阅模式归一为稳定幂等键组成部分。""" + if not subscribe.best_version: + return "normal" + if subscribe.type == MediaType.TV.value and subscribe.best_version_full: + return "best_version_full" + if subscribe.type == MediaType.TV.value: + return "best_version_episode" + return "best_version" + @classmethod def _is_episode_range_covered(cls, meta: MetaBase, subscribe: SubscriptionSnapshot) -> bool: """ diff --git a/app/chain/subscribe/search.py b/app/chain/subscribe/search.py index 8edd9e03c..f45f23278 100644 --- a/app/chain/subscribe/search.py +++ b/app/chain/subscribe/search.py @@ -3,6 +3,7 @@ import random import time from datetime import datetime, timedelta, timezone +from functools import partial from typing import Any, Callable, Optional, cast from uuid import uuid4 @@ -13,8 +14,8 @@ from app.application.subscription.contract import ( build_subscribe_meta, subscribe_media_key, ) -from app.application.subscription.query import SubscriptionQueryService from app.application.subscription.execution import SearchBatchSnapshot, SubscriptionSearchRepository +from app.application.subscription.query import SubscriptionQueryService from app.application.subscription.sitebudget import ( SubscriptionSearchCancelled, SubscriptionSiteBudget, @@ -118,9 +119,8 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): :param sids: 订阅ID集合,有值时按给定顺序处理 :return: 更新订阅状态为R或删除订阅 """ - queue = cast( - Optional[SubscriptionSearchRepository], - getattr(self, "subscription_search_repository", None), + queue: Optional[SubscriptionSearchRepository] = getattr( + self, "subscription_search_repository", None ) if queue is not None: return self._execute_queued_search( @@ -246,7 +246,7 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): "coalesced": enqueued.coalesced_count, }, ) - return enqueued.batch.batch_id + return str(enqueued.batch.batch_id) def _drain_search_queue( self, @@ -273,6 +273,8 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): if not task.lease_token: logger.error(f"订阅搜索任务 {task.task_id} 缺少租约令牌,跳过执行") continue + task_id = str(task.task_id) + cancelled = partial(queue.is_cancel_requested, task_id) if queue.is_cancel_requested(task.task_id): queue.release_task( task_id=task.task_id, @@ -301,19 +303,31 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): searchchain.configure_subscription_site_budget( SubscriptionSiteBudget( repository=queue, - owner=f"{owner}:{task.task_id}", - cancelled=lambda task_id=task.task_id: queue.is_cancel_requested(task_id), + owner=f"{owner}:{task_id}", + cancelled=cancelled, stop_state=getattr(self, "stop_state", runtime_stop_state), ) ) + self._subscription_download_task_id = task_id + self._subscription_download_cancelled = cancelled + self._subscription_download_crossed_boundary = False + self._subscription_download_mark_started = self._mark_subscription_download_started try: current = self._process_search_subscription(subscribe, searchchain) if queue.is_cancel_requested(task.task_id): - queue.release_task( - task_id=task.task_id, - lease_token=task.lease_token, - cancelled=True, - ) + if self._subscription_download_started_for_task(task.task_id): + queue.finish_task( + task_id=task.task_id, + lease_token=task.lease_token, + state="completed", + error="取消请求晚于下载提交边界,已按实际结果完成", + ) + else: + queue.release_task( + task_id=task.task_id, + lease_token=task.lease_token, + cancelled=True, + ) else: queue.finish_task( task_id=task.task_id, @@ -336,6 +350,10 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): error=str(err), ) finally: + delattr(self, "_subscription_download_task_id") + delattr(self, "_subscription_download_cancelled") + delattr(self, "_subscription_download_mark_started") + self._subscription_download_crossed_boundary = False searchchain.configure_subscription_site_budget(None) if current and current.state == "N": try: @@ -355,15 +373,25 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): queue_lock.release() return processed + def _subscription_download_started_for_task(self, task_id: str) -> bool: + """判断取消是否已晚于下载器副作用边界。""" + if bool(getattr(self, "_subscription_download_crossed_boundary", False)): + return True + repository = getattr(self, "subscription_download_repository", None) + return bool(repository and repository.has_started_for_task(task_id)) + + def _mark_subscription_download_started(self) -> None: + """记录当前搜索任务已提交或复用了真实下载结果。""" + self._subscription_download_crossed_boundary = True + def resume_search_queue( self, progress_callback: Optional[Callable[..., None]] = None, limit: int = 50, ) -> None: """短周期恢复排队或租约过期任务,不创建新的 24 小时兜底批次。""" - queue = cast( - Optional[SubscriptionSearchRepository], - getattr(self, "subscription_search_repository", None), + queue: Optional[SubscriptionSearchRepository] = getattr( + self, "subscription_search_repository", None ) if queue is None: return @@ -413,17 +441,15 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): def cancel_search_batch(self, batch_id: str) -> bool: """请求取消持久搜索批次;未注入队列时返回失败。""" - queue = cast( - Optional[SubscriptionSearchRepository], - getattr(self, "subscription_search_repository", None), + queue: Optional[SubscriptionSearchRepository] = getattr( + self, "subscription_search_repository", None ) return bool(queue and queue.request_cancel(batch_id)) def get_search_batch(self, batch_id: str) -> Optional[SearchBatchSnapshot]: """返回持久搜索批次状态;未注入队列时返回空。""" - queue = cast( - Optional[SubscriptionSearchRepository], - getattr(self, "subscription_search_repository", None), + queue: Optional[SubscriptionSearchRepository] = getattr( + self, "subscription_search_repository", None ) return queue.get_batch(batch_id) if queue else None diff --git a/app/chain/torrents.py b/app/chain/torrents.py index 60e23ddca..bd2b1984a 100644 --- a/app/chain/torrents.py +++ b/app/chain/torrents.py @@ -12,6 +12,7 @@ from app.chain.base import ChainBase from app.chain.media import MediaChain from app.domain import site as site_rules from app.domain.context import Context, MediaInfo, MusicInfo, TorrentInfo +from app.domain.meta.metabase import MetaBase from app.domain.meta.metamusic import MetaMusic from app.domain.metainfo import MetaInfo from app.runtime.log import logger @@ -496,6 +497,8 @@ class TorrentsChain(ChainBase): def _build_refresh_context(self, torrent: TorrentInfo, stype: str) -> Context: """识别单个种子并构造缓存上下文。""" logger.info(f'处理资源:{torrent.title} ...') + meta: MetaBase + mediainfo: MediaInfo | MusicInfo if torrent.category == MediaType.MUSIC.value: meta = MetaMusic.parse_query(torrent.title) mediainfo = MusicInfo( @@ -508,15 +511,23 @@ class TorrentsChain(ChainBase): candidate_recognized = False match_source = "unknown" else: - meta = MetaInfo(title=torrent.title, subtitle=torrent.description) - if torrent.title != meta.org_string: - logger.info(f'种子名称应用识别词后发生改变:{torrent.title} => {meta.org_string}') - if meta.type != MediaType.TV and torrent.category == MediaType.TV.value: - meta.type = MediaType.TV - mediainfo = MediaChain().recognize_by_meta(meta, obtain_images=False) or MediaInfo() - mediainfo.clear() - candidate_recognized = bool(mediainfo and all(resolve_media_identity(media=mediainfo))) - match_source = self._get_media_id_match_source(mediainfo) + video_meta = MetaInfo(title=torrent.title, subtitle=torrent.description) + if torrent.title != video_meta.org_string: + logger.info( + f'种子名称应用识别词后发生改变:{torrent.title} => {video_meta.org_string}' + ) + if video_meta.type != MediaType.TV and torrent.category == MediaType.TV.value: + video_meta.type = MediaType.TV + video_mediainfo = ( + MediaChain().recognize_by_meta(video_meta, obtain_images=False) or MediaInfo() + ) + video_mediainfo.clear() + candidate_recognized = bool( + video_mediainfo and all(resolve_media_identity(media=video_mediainfo)) + ) + match_source = self._get_media_id_match_source(video_mediainfo) + meta = video_meta + mediainfo = video_mediainfo context = Context( meta_info=meta, media_info=mediainfo, diff --git a/app/db/adapters/subscriptiondownload.py b/app/db/adapters/subscriptiondownload.py new file mode 100644 index 000000000..896acce2c --- /dev/null +++ b/app/db/adapters/subscriptiondownload.py @@ -0,0 +1,137 @@ +"""订阅下载幂等账本的短事务适配器。""" + +from collections.abc import Callable +from typing import Optional, TypeVar + +from sqlalchemy.orm import Session + +from app.application.download.admission import ( + SubscriptionDownloadClaim, + SubscriptionDownloadRequest, + SubscriptionDownloadSnapshot, +) +from app.db.models.subscriptiondownload import SubscriptionDownloadSubmission +from app.db.oper.subscriptiondownload import SubscriptionDownloadOper +from app.db.uow import SqlAlchemyUnitOfWork + +T = TypeVar("T") + + +def _snapshot(record: SubscriptionDownloadSubmission) -> SubscriptionDownloadSnapshot: + """在 Session 内投影不可变提交快照。""" + return SubscriptionDownloadSnapshot( + idempotency_key=record.idempotency_key, + subscription_id=record.subscription_id, + task_id=record.task_id, + state=record.state, + attempt_count=record.attempt_count, + attempt_token=record.attempt_token, + downloader=record.downloader, + download_hash=record.download_hash, + available_at=record.available_at, + last_error=record.last_error, + created_at=record.created_at, + updated_at=record.updated_at, + ) + + +class TransactionalSubscriptionDownloadRepository: + """为每次提交认领和状态变更创建独立短事务。""" + + def __init__(self, session_factory: Callable[[], Session]) -> None: + """保存由组合根提供的同步 Session 工厂。""" + self._session_factory = session_factory + + def _read(self, operation: Callable[[SubscriptionDownloadOper], T]) -> T: + """在短 Session 中执行一次只读查询。""" + with self._session_factory() as session: + return operation(SubscriptionDownloadOper(session)) + + def _write(self, operation: Callable[[SubscriptionDownloadOper], T]) -> T: + """在显式短事务中执行一次状态变更。""" + with self._session_factory() as session: + unit_of_work = SqlAlchemyUnitOfWork(session) + try: + result = operation(SubscriptionDownloadOper(session)) + unit_of_work.commit() + return result + except Exception: + unit_of_work.rollback() + raise + + def claim(self, request: SubscriptionDownloadRequest) -> SubscriptionDownloadClaim: + """认领唯一提交权并返回脱离 Session 的当前状态。""" + def operation(repository: SubscriptionDownloadOper) -> SubscriptionDownloadClaim: + """在同一事务内完成唯一插入或 fenced 重领。""" + record, acquired = repository.claim(request) + return SubscriptionDownloadClaim(acquired=acquired, snapshot=_snapshot(record)) + + return self._write(operation) + + def mark_accepted( + self, + *, + idempotency_key: str, + attempt_token: str, + downloader: Optional[str], + download_hash: str, + ) -> bool: + """记录下载器已接受任务。""" + return self._write(lambda repository: repository.mark_accepted( + idempotency_key=idempotency_key, + attempt_token=attempt_token, + downloader=downloader, + download_hash=download_hash, + )) + + def mark_succeeded(self, *, idempotency_key: str, attempt_token: str) -> bool: + """记录 canonical 本地结算已完成。""" + return self._write(lambda repository: repository.mark_succeeded( + idempotency_key=idempotency_key, + attempt_token=attempt_token, + )) + + def mark_retryable( + self, + *, + idempotency_key: str, + attempt_token: str, + available_at: str, + error: Optional[str], + ) -> bool: + """记录未产生外部副作用的延迟重试状态。""" + return self._write(lambda repository: repository.mark_retryable( + idempotency_key=idempotency_key, + attempt_token=attempt_token, + available_at=available_at, + error=error, + )) + + def mark_reconcile_required( + self, + *, + idempotency_key: str, + attempt_token: str, + error: Optional[str], + downloader: Optional[str] = None, + download_hash: Optional[str] = None, + ) -> bool: + """冻结可能已被下载器接受的提交。""" + return self._write(lambda repository: repository.mark_reconcile_required( + idempotency_key=idempotency_key, + attempt_token=attempt_token, + error=error, + downloader=downloader, + download_hash=download_hash, + )) + + def mark_cancelled(self, *, idempotency_key: str, attempt_token: str) -> bool: + """在外部提交前收口取消。""" + return self._write(lambda repository: repository.mark_cancelled( + idempotency_key=idempotency_key, + attempt_token=attempt_token, + )) + + def has_started_for_task(self, task_id: str) -> bool: + """查询搜索任务是否已越过可安全取消边界。""" + return self._read(lambda repository: repository.has_started_for_task(task_id)) diff --git a/app/db/adapters/subscriptionsearch.py b/app/db/adapters/subscriptionsearch.py index 468135c05..61995bf9e 100644 --- a/app/db/adapters/subscriptionsearch.py +++ b/app/db/adapters/subscriptionsearch.py @@ -12,7 +12,6 @@ from app.application.subscription.execution import ( ) from app.application.subscription.sitebudget import SiteBudgetClaim from app.db.models.subscriptionsearch import ( - SubscriptionSiteBudget, SubscriptionSearchBatch, SubscriptionSearchTask, ) diff --git a/app/db/models/__init__.py b/app/db/models/__init__.py index 6287ec479..d1a3c2b96 100644 --- a/app/db/models/__init__.py +++ b/app/db/models/__init__.py @@ -5,7 +5,6 @@ from typing import Any from . import _identity # noqa: F401 注册全局媒体身份写入不变量 - _MODEL_EXPORTS = { "AgentChat": ("app.db.models.agentchat", "AgentChat"), "AgentTask": ("app.db.models.agenttask", "AgentTask"), @@ -47,6 +46,10 @@ _MODEL_EXPORTS = { "app.db.models.subscriptionsearch", "SubscriptionSiteBudget", ), + "SubscriptionDownloadSubmission": ( + "app.db.models.subscriptiondownload", + "SubscriptionDownloadSubmission", + ), "SystemConfig": ("app.db.models.systemconfig", "SystemConfig"), "TransferHistory": ("app.db.models.transferhistory", "TransferHistory"), "TransferExecutionStep": ( diff --git a/app/db/models/subscriptiondownload.py b/app/db/models/subscriptiondownload.py new file mode 100644 index 000000000..68d8f3c2a --- /dev/null +++ b/app/db/models/subscriptiondownload.py @@ -0,0 +1,52 @@ +"""订阅下载提交幂等账本模型。""" + +from typing import Optional + +from sqlalchemy import Index, Integer, String, Text, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.base import Base, get_id_column + + +class SubscriptionDownloadSubmission(Base): + """记录订阅下载唯一提交权、下载器接受事实和待对账终态。""" + + id = get_id_column() + idempotency_key: Mapped[str] = mapped_column(String(64), nullable=False) + subscription_id: Mapped[int] = mapped_column(Integer, nullable=False) + task_id: Mapped[Optional[str]] = mapped_column(String(64)) + logical_identity: Mapped[str] = mapped_column(Text, nullable=False) + resource_key: Mapped[str] = mapped_column(Text, nullable=False) + coverage: Mapped[str] = mapped_column(Text, nullable=False) + mode: Mapped[str] = mapped_column(String(32), nullable=False) + state: Mapped[str] = mapped_column(String(32), nullable=False) + attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + attempt_token: Mapped[Optional[str]] = mapped_column(String(64)) + downloader: Mapped[Optional[str]] = mapped_column(String(128)) + download_hash: Mapped[Optional[str]] = mapped_column(String(256)) + available_at: Mapped[Optional[str]] = mapped_column(String(40)) + last_error: Mapped[Optional[str]] = mapped_column(Text) + created_at: Mapped[str] = mapped_column(String(40), nullable=False) + updated_at: Mapped[str] = mapped_column(String(40), nullable=False) + started_at: Mapped[Optional[str]] = mapped_column(String(40)) + finished_at: Mapped[Optional[str]] = mapped_column(String(40)) + + __table_args__ = ( + UniqueConstraint( + "idempotency_key", + name="uq_subscriptiondownloadsubmission_idempotency_key", + ), + Index( + "ix_subscriptiondownloadsubmission_task_state", + "task_id", + "state", + "id", + ), + Index( + "ix_subscriptiondownloadsubmission_subscription_state", + "subscription_id", + "state", + "updated_at", + "id", + ), + ) diff --git a/app/db/oper/subscriptiondownload.py b/app/db/oper/subscriptiondownload.py new file mode 100644 index 000000000..97f7292e8 --- /dev/null +++ b/app/db/oper/subscriptiondownload.py @@ -0,0 +1,231 @@ +"""订阅下载提交账本的事务内状态转换。""" + +from datetime import datetime, timezone +from typing import Optional +from uuid import uuid4 + +from sqlalchemy import and_, or_, select, update +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from app.application.download.admission import SubscriptionDownloadRequest +from app.db.base import DbOper, execute_dml +from app.db.models.subscriptiondownload import SubscriptionDownloadSubmission + + +def utc_now_text() -> str: + """返回可按字符串稳定排序的 UTC ISO 时间。""" + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +class SubscriptionDownloadOper(DbOper): + """在调用方事务内维护唯一下载提交及其 fenced 终态。""" + + def claim(self, request: SubscriptionDownloadRequest) -> tuple[SubscriptionDownloadSubmission, bool]: + """创建或原子重领允许重试的提交记录。""" + if not isinstance(self._db, Session): + raise RuntimeError("订阅下载提交认领需要调用方提供同步 Session") + now = utc_now_text() + attempt_token = uuid4().hex + record = self._db.execute( + select(SubscriptionDownloadSubmission).where( + SubscriptionDownloadSubmission.idempotency_key == request.idempotency_key + ) + ).scalars().first() + if record is None: + try: + with self._db.begin_nested(): + record = SubscriptionDownloadSubmission( + idempotency_key=request.idempotency_key, + subscription_id=request.subscription_id, + task_id=request.task_id, + logical_identity=request.logical_identity, + resource_key=request.resource_key, + coverage=request.coverage, + mode=request.mode, + state="submitting", + attempt_count=1, + attempt_token=attempt_token, + created_at=now, + updated_at=now, + started_at=now, + ) + self._db.add(record) + self._db.flush() + return record, True + except IntegrityError: + self._db.expire_all() + record = self._db.execute( + select(SubscriptionDownloadSubmission).where( + SubscriptionDownloadSubmission.idempotency_key == request.idempotency_key + ) + ).scalar_one() + + reclaimable = record.state in {"retryable", "cancelled"} and ( + record.available_at is None or record.available_at <= now + ) + if not reclaimable: + return record, False + updated = execute_dml( + self._db, + update(SubscriptionDownloadSubmission) + .where( + SubscriptionDownloadSubmission.id == record.id, + SubscriptionDownloadSubmission.state.in_(("retryable", "cancelled")), + or_( + SubscriptionDownloadSubmission.available_at.is_(None), + SubscriptionDownloadSubmission.available_at <= now, + ), + ) + .values( + state="submitting", + task_id=request.task_id, + attempt_count=SubscriptionDownloadSubmission.attempt_count + 1, + attempt_token=attempt_token, + downloader=None, + download_hash=None, + available_at=None, + last_error=None, + updated_at=now, + started_at=now, + finished_at=None, + ), + execution_options={"synchronize_session": False}, + ) + self._db.flush() + self._db.expire_all() + current = self._db.execute( + select(SubscriptionDownloadSubmission).where( + SubscriptionDownloadSubmission.id == record.id + ) + ).scalar_one() + return current, bool(updated and current.attempt_token == attempt_token) + + def mark_accepted( + self, + *, + idempotency_key: str, + attempt_token: str, + downloader: Optional[str], + download_hash: str, + ) -> bool: + """把当前 submitting 尝试推进为下载器已接受。""" + return self._transition( + idempotency_key=idempotency_key, + attempt_token=attempt_token, + source_states=("submitting",), + values={ + "state": "accepted", + "downloader": downloader, + "download_hash": download_hash, + }, + ) + + def mark_succeeded(self, *, idempotency_key: str, attempt_token: str) -> bool: + """把当前 accepted 尝试推进为本地结算成功。""" + return self._transition( + idempotency_key=idempotency_key, + attempt_token=attempt_token, + source_states=("accepted",), + values={"state": "succeeded", "finished_at": utc_now_text()}, + ) + + def mark_retryable( + self, + *, + idempotency_key: str, + attempt_token: str, + available_at: str, + error: Optional[str], + ) -> bool: + """把下载器明确拒绝的当前尝试收口为延迟可重试。""" + return self._transition( + idempotency_key=idempotency_key, + attempt_token=attempt_token, + source_states=("submitting",), + values={ + "state": "retryable", + "available_at": available_at, + "last_error": error, + "finished_at": utc_now_text(), + }, + ) + + def mark_reconcile_required( + self, + *, + idempotency_key: str, + attempt_token: str, + error: Optional[str], + downloader: Optional[str], + download_hash: Optional[str], + ) -> bool: + """冻结 submitting/accepted 尝试,禁止自动重新提交。""" + values: dict[str, object] = { + "state": "reconcile_required", + "last_error": error, + "finished_at": utc_now_text(), + } + if downloader is not None: + values["downloader"] = downloader + if download_hash is not None: + values["download_hash"] = download_hash + return self._transition( + idempotency_key=idempotency_key, + attempt_token=attempt_token, + source_states=("submitting", "accepted"), + values=values, + ) + + def mark_cancelled(self, *, idempotency_key: str, attempt_token: str) -> bool: + """在外部提交前把当前尝试收口为已取消。""" + return self._transition( + idempotency_key=idempotency_key, + attempt_token=attempt_token, + source_states=("submitting",), + values={"state": "cancelled", "finished_at": utc_now_text()}, + ) + + def has_started_for_task(self, task_id: str) -> bool: + """判断任务是否进入过外部副作用边界或已成功。""" + if not isinstance(self._db, Session): + raise RuntimeError("订阅下载提交查询需要调用方提供同步 Session") + return bool(self._db.execute( + select(SubscriptionDownloadSubmission.id) + .where( + SubscriptionDownloadSubmission.task_id == task_id, + SubscriptionDownloadSubmission.state.in_(( + "submitting", + "accepted", + "succeeded", + "reconcile_required", + )), + ) + .limit(1) + ).scalar()) + + def _transition( + self, + *, + idempotency_key: str, + attempt_token: str, + source_states: tuple[str, ...], + values: dict[str, object], + ) -> bool: + """以当前尝试令牌原子推进状态,拒绝过期执行者写回。""" + if not isinstance(self._db, Session): + raise RuntimeError("订阅下载提交写入需要调用方提供同步 Session") + now = utc_now_text() + return bool(execute_dml( + self._db, + update(SubscriptionDownloadSubmission) + .where( + and_( + SubscriptionDownloadSubmission.idempotency_key == idempotency_key, + SubscriptionDownloadSubmission.attempt_token == attempt_token, + SubscriptionDownloadSubmission.state.in_(source_states), + ) + ) + .values(updated_at=now, **values), + execution_options={"synchronize_session": False}, + )) diff --git a/app/db/oper/subscriptionsearch.py b/app/db/oper/subscriptionsearch.py index 1e780cd9e..9c69f5e4a 100644 --- a/app/db/oper/subscriptionsearch.py +++ b/app/db/oper/subscriptionsearch.py @@ -10,9 +10,9 @@ from sqlalchemy.orm import Session from app.db.base import DbOper, execute_dml from app.db.models.subscriptionsearch import ( - SubscriptionSiteBudget, SubscriptionSearchBatch, SubscriptionSearchTask, + SubscriptionSiteBudget, ) @@ -198,12 +198,13 @@ class SubscriptionSearchOper(DbOper): ) self._db.flush() self._db.expire_all() - return self._db.execute( + claimed_task: Optional[SubscriptionSearchTask] = self._db.execute( select(SubscriptionSearchTask).where( SubscriptionSearchTask.id == candidate.id, SubscriptionSearchTask.lease_token == lease_token, ) ).scalars().first() + return claimed_task return None def finish_task( @@ -359,9 +360,10 @@ class SubscriptionSearchOper(DbOper): """按稳定批次 ID 读取聚合记录。""" if not isinstance(self._db, Session): raise RuntimeError("订阅搜索批次查询需要调用方提供同步 Session") - return self._db.execute( + batch: Optional[SubscriptionSearchBatch] = self._db.execute( select(SubscriptionSearchBatch).where(SubscriptionSearchBatch.batch_id == batch_id) ).scalars().first() + return batch def claim_site( self, @@ -477,7 +479,10 @@ class SubscriptionSearchOper(DbOper): def _ensure_site_budget(self, *, site_id: int, now: str) -> SubscriptionSiteBudget: """并发安全地创建站点预算初始记录。""" - record = self._db.execute( + db = self._db + if not isinstance(db, Session): + raise RuntimeError("订阅站点预算初始化需要调用方提供同步 Session") + record: Optional[SubscriptionSiteBudget] = db.execute( select(SubscriptionSiteBudget).where( SubscriptionSiteBudget.site_id == site_id ) @@ -485,24 +490,28 @@ class SubscriptionSearchOper(DbOper): if record is not None: return record try: - with self._db.begin_nested(): - self._db.add(SubscriptionSiteBudget( + with db.begin_nested(): + db.add(SubscriptionSiteBudget( site_id=site_id, next_allowed_at=now, updated_at=now, )) - self._db.flush() + db.flush() except IntegrityError: - self._db.expire_all() - return self._db.execute( + db.expire_all() + created: SubscriptionSiteBudget = db.execute( select(SubscriptionSiteBudget).where( SubscriptionSiteBudget.site_id == site_id ) ).scalar_one() + return created def _batch_cancel_requested(self, batch_id: str) -> bool: """在当前事务中读取批次取消标记。""" - return bool(self._db.execute( + db = self._db + if not isinstance(db, Session): + raise RuntimeError("订阅搜索批次取消查询需要调用方提供同步 Session") + return bool(db.execute( select(SubscriptionSearchBatch.cancel_requested).where( SubscriptionSearchBatch.batch_id == batch_id ) @@ -510,7 +519,10 @@ class SubscriptionSearchOper(DbOper): def _refresh_batch(self, batch_id: str, *, now: str, error: Optional[str]) -> None: """依据所属任务终态重新计算批次计数和聚合状态。""" - rows = self._db.execute( + db = self._db + if not isinstance(db, Session): + raise RuntimeError("订阅搜索批次刷新需要调用方提供同步 Session") + rows = db.execute( select(SubscriptionSearchTask.state, func.count()) # pylint: disable=not-callable .where(SubscriptionSearchTask.batch_id == batch_id) .group_by(SubscriptionSearchTask.state) diff --git a/app/domain/context.py b/app/domain/context.py index 852c8d1e8..0c22fcb3a 100644 --- a/app/domain/context.py +++ b/app/domain/context.py @@ -1524,7 +1524,7 @@ class MediaInfo: dicts["media_id"] = str(self.media_id) if self.media_id is not None else None return dicts - def clear(self): + def clear(self) -> None: """ 去除多余数据,减小体积 """ diff --git a/app/startup/composition/chain.py b/app/startup/composition/chain.py index 3be18bd01..4dcfd49b6 100644 --- a/app/startup/composition/chain.py +++ b/app/startup/composition/chain.py @@ -36,6 +36,7 @@ from app.chain.transfer.filter import ( from app.db.adapters.chain import TransactionalChainDurableEventWriter from app.db.adapters.download import TransactionalDownloadFailureRepository from app.db.adapters.mediaserver import TransactionalMediaServerRepository +from app.db.adapters.subscriptiondownload import TransactionalSubscriptionDownloadRepository from app.db.adapters.transfer.admission import TransactionalTransferAdmissionRepository from app.db.oper.message import MessageOper from app.db.oper.systemconfig import SystemConfigOper @@ -111,6 +112,7 @@ def build_chain_runtime_context( transfer_execution_repository=dependencies.transfer_execution, media_server_repository=TransactionalMediaServerRepository(SessionFactory), download_failure_repository=TransactionalDownloadFailureRepository(SessionFactory), + subscription_download_repository=TransactionalSubscriptionDownloadRepository(SessionFactory), user_repository=build_transactional_user_repository(), legacy_transfer_command=execute_legacy_transfer_command, configuration=configuration(), diff --git a/app/startup/composition/runtime.py b/app/startup/composition/runtime.py index c71ff5771..e5434ec0b 100644 --- a/app/startup/composition/runtime.py +++ b/app/startup/composition/runtime.py @@ -48,8 +48,8 @@ from app.startup.composition.context import ( ) if TYPE_CHECKING: - from app.application.subscription.execution import SubscriptionSearchRepository from app.application.messaging.message import MessageHelper, MessageQueueManager + from app.application.subscription.execution import SubscriptionSearchRepository from app.startup.composition.agent import AgentComposition from app.startup.composition.configuration import ConfigurationComposition from app.startup.composition.database import DatabaseComposition diff --git a/database/versions/e1b6d4f8a2c7_3_0_21.py b/database/versions/e1b6d4f8a2c7_3_0_21.py new file mode 100644 index 000000000..a53337b8a --- /dev/null +++ b/database/versions/e1b6d4f8a2c7_3_0_21.py @@ -0,0 +1,72 @@ +"""3.0.21 增加订阅下载幂等与待对账提交账本。 + +Revision ID: e1b6d4f8a2c7 +Revises: d2a7c5e9f1b4 +Create Date: 2026-09-01 +""" + +# Alembic 的 op 是运行期代理,静态分析无法看到实际操作方法。 +# pylint: disable=no-member + +import sqlalchemy as sa +from alembic import op + +revision = "e1b6d4f8a2c7" +down_revision = "d2a7c5e9f1b4" +branch_labels = None +depends_on = None + +_TABLE = "subscriptiondownloadsubmission" + + +def _table_names() -> set[str]: + """返回当前数据库表名集合。""" + return set(sa.inspect(op.get_bind()).get_table_names()) + + +def upgrade() -> None: + """创建向后兼容的独立订阅下载提交账本。""" + if _TABLE in _table_names(): + return + op.create_table( + _TABLE, + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("idempotency_key", sa.String(length=64), nullable=False), + sa.Column("subscription_id", sa.Integer(), nullable=False), + sa.Column("task_id", sa.String(length=64), nullable=True), + sa.Column("logical_identity", sa.Text(), nullable=False), + sa.Column("resource_key", sa.Text(), nullable=False), + sa.Column("coverage", sa.Text(), nullable=False), + sa.Column("mode", sa.String(length=32), nullable=False), + sa.Column("state", sa.String(length=32), nullable=False), + sa.Column("attempt_count", sa.Integer(), nullable=False), + sa.Column("attempt_token", sa.String(length=64), nullable=True), + sa.Column("downloader", sa.String(length=128), nullable=True), + sa.Column("download_hash", sa.String(length=256), nullable=True), + sa.Column("available_at", sa.String(length=40), nullable=True), + sa.Column("last_error", sa.Text(), nullable=True), + sa.Column("created_at", sa.String(length=40), nullable=False), + sa.Column("updated_at", sa.String(length=40), nullable=False), + sa.Column("started_at", sa.String(length=40), nullable=True), + sa.Column("finished_at", sa.String(length=40), nullable=True), + sa.UniqueConstraint( + "idempotency_key", + name="uq_subscriptiondownloadsubmission_idempotency_key", + ), + ) + op.create_index( + "ix_subscriptiondownloadsubmission_task_state", + _TABLE, + ["task_id", "state", "id"], + ) + op.create_index( + "ix_subscriptiondownloadsubmission_subscription_state", + _TABLE, + ["subscription_id", "state", "updated_at", "id"], + ) + + +def downgrade() -> None: + """移除订阅下载提交账本,不改动旧下载历史。""" + if _TABLE in _table_names(): + op.drop_table(_TABLE) diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index a4683ffd5..b556fbd6c 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -754,8 +754,8 @@ flowchart LR | 指标 | 当前值 | |---|---:| -| Python 模块 | 927 | -| 内部导入边 | 7,745 | +| Python 模块 | 932 | +| 内部导入边 | 7,784 | | 非平凡 SCC | 1(精确 containment 的 TMDB 移植包环) | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 53(债务已清零,53 条精确 containment) | diff --git a/docs/architecture/optimization-checklist.md b/docs/architecture/optimization-checklist.md index c72a46493..86064ddb4 100644 --- a/docs/architecture/optimization-checklist.md +++ b/docs/architecture/optimization-checklist.md @@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 927 / 7,745 | `dependency-baseline.json` 当前快照 | +| 宿主 Python 模块 / 内部依赖边 | 932 / 7,784 | `dependency-baseline.json` 当前快照 | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | @@ -102,8 +102,8 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | Event Contract | 53 | 均已有 payload model,但当前全部是 diagnostic enforcement | | Python 源码量 | 305,884 行 | 排除 `app/plugins/**`;61 个文件超过 1,000 行,11 个超过 2,000 行 | | 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`;65 个超过 150 行,21 个超过 250 行 | -| 全量 mypy 历史债务 | 9,603 / 517 文件 | Agent API 重构后的现状基线;canonical Facade 与 endpoint 类型边界已补齐,低水位只允许继续下降 | -| Ruff 历史诊断 | 569 | 低水位门禁通过,但规则集只覆盖 `E4/E7/E9/F/I` | +| 全量 mypy 历史债务 | 9,589 / 517 文件 | Agent API 重构后的现状基线;canonical Facade 与 endpoint 类型边界已补齐,低水位只允许继续下降 | +| Ruff 历史诊断 | 568 | 低水位门禁通过,但规则集只覆盖 `E4/E7/E9/F/I` | | 覆盖率固定基线 | Application 80.00%,Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 | ### 3.3 热点文件 diff --git a/docs/refactor/moviepilot-subscription-governance-roadmap.md b/docs/refactor/moviepilot-subscription-governance-roadmap.md index 0a9f9a001..e131184e2 100644 --- a/docs/refactor/moviepilot-subscription-governance-roadmap.md +++ b/docs/refactor/moviepilot-subscription-governance-roadmap.md @@ -1,7 +1,7 @@ # MoviePilot 订阅执行治理 > 状态:`active(2026-09-01 已由 MoviePilot v3 接管)` -> 当前叶:`SUB-GOV-003A` +> 当前叶:`SUB-GOV-003B` > 迁移来源:`V3-RDY-009A`、`V3-RDY-009A1`、`V3-RDY-009A1B`、 > `V3-RDY-009A1C`、`V3-RDY-009A1D` > 适用范围:V3 订阅刷新、匹配、兜底搜索、下载提交与用户状态闭环 @@ -235,11 +235,11 @@ RSS/Spider 保持现有逐站点串行刷新,不通过并发化换取几秒收 | `SUB-GOV-002A` | 将 24 小时兜底搜索移出 Match/提交长锁,建立批次、订阅 single-flight、可取消等待和恢复游标 | 001C | `completed(2026-09-01)` | | `SUB-GOV-002B` | 为兜底搜索建立每站点并发、间隔和错误冷却预算;RSS/Spider 保持基线压力 | 002A | `completed(2026-09-01)` | | `SUB-GOV-002C` | 联合运行日常 Match、手工搜索和兜底搜索,验证公平性、锁等待、站点压力和失败恢复 | 001D, 002B | `completed(2026-09-01)` | -| `SUB-GOV-003A` | 建立跨入口的订阅级下载幂等、下载器不确定终态和取消补偿 | 001C, 002A | `in_progress` | -| `SUB-GOV-003B` | 后端提供订阅及批次业务状态,前端展示排队、匹配、搜索、提交、完成、失败和取消 | 002A, 003A | `pending` | +| `SUB-GOV-003A` | 建立跨入口的订阅级下载幂等、下载器不确定终态和取消补偿 | 001C, 002A | `completed(2026-09-01)` | +| `SUB-GOV-003B` | 后端提供订阅及批次业务状态,前端展示排队、匹配、搜索、提交、完成、失败和取消 | 002A, 003A | `in_progress` | | `SUB-GOV-004` | 多条订阅记录指向同一媒体时的跨记录季集去重和产品规则 | 003A | `pending(最低优先级)` | -当前只激活 `SUB-GOV-003A`。001B–D 是日常 Match 主线;002A–C 是 24 小时兜底搜索主线;003A/B +当前只激活 `SUB-GOV-003B`。001B–D 是日常 Match 主线;002A–C 是 24 小时兜底搜索主线;003A/B 只有在前两条链路的身份和终态稳定后实施。每个叶子完成验收并更新本表后,才激活下一个满足依赖的叶子。 ### 6.1 SUB-GOV-001A 验收证据 @@ -343,6 +343,23 @@ RSS/Spider 保持现有逐站点串行刷新,不通过并发化换取几秒收 - 验证:联合队列/站点预算/搜索状态 `35 passed`,订阅/搜索/调度架构 `58 passed`,错误级 Pylint 为 0, Host 架构基线与 `git diff --check` 通过。 +### 6.9 SUB-GOV-003A 验收证据 + +- 新增持久 `SubscriptionDownloadSubmission` 账本和 `SubscriptionDownloadGovernance` 应用合同;幂等键包含 + 订阅 ID、媒体逻辑身份、torrent 稳定身份、精确季集覆盖和普通/洗版模式,唯一约束覆盖 RSS、Spider、 + 手工搜索与兜底搜索的重叠入口; +- 下载器调用前以 fenced attempt token 原子认领;明确拒绝进入带冷却的 retryable,已接受但本地结算失败或 + 下载器超时进入 `reconcile_required`,不得自动重试;成功记录可复用既有 hash,迁移前同订阅同 torrent + 同覆盖的下载历史也参与兼容去重; +- 生产订阅下载路径在提交前重新读取订阅、缺集、下载历史、洗版优先级和当前下载参数;订阅删除、暂停或筛选 + 条件变化时停止提交,准备阶段快照不越过外部副作用边界; +- 取消在下载器调用前写入 cancelled 且不产生外部提交;越过提交边界后的取消按真实下载结果完成并保留说明, + 不伪造 cancelled 终态;任务重启通过持久账本判断是否已经开始; +- Alembic `e1b6d4f8a2c7` 为唯一 head,迁移支持 SQLite/PostgreSQL、可逆且重复执行安全;公开 + `DownloadChain.download_single()` / `batch_download()` 只在末尾增加可选治理参数,旧插件 ABI 保持兼容; +- 验证:下载幂等、取消补偿、订阅提交与搜索治理 `131 passed`;架构合同、依赖、复杂度和质量 ratchet + `284 passed`;Ruff 低水位 `568`、mypy 低水位 `9,589`,Host 架构基线与 `git diff --check` 通过。 + ## 7. 上线前验证与验收 ### 7.1 场景 diff --git a/tests/conftest.py b/tests/conftest.py index 5ff277b84..a368f9224 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -283,6 +283,7 @@ def configure_plugin_system_services(): from app.db.adapters.mediaserver import TransactionalMediaServerRepository from app.db.adapters.site import TransactionalSiteRepository from app.db.adapters.subscription import TransactionalSubscriptionRepository + from app.db.adapters.subscriptiondownload import TransactionalSubscriptionDownloadRepository from app.db.adapters.transaction import TransactionalWriteRunner from app.db.adapters.transfer.admission import TransactionalTransferAdmissionRepository from app.db.adapters.transfer.execution import ( @@ -410,6 +411,7 @@ def configure_plugin_system_services(): transfer_execution_repository=TransactionalTransferExecutionRepository(SessionFactory), media_server_repository=TransactionalMediaServerRepository(SessionFactory), download_failure_repository=TransactionalDownloadFailureRepository(SessionFactory), + subscription_download_repository=TransactionalSubscriptionDownloadRepository(SessionFactory), user_repository=user_repository(), configuration=build_chain_runtime_config(settings), ) diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index e21957238..1b8d20386 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1089,8 +1089,8 @@ "runtime_only": true } }, - "edge_count": 7745, - "edge_sha256": "4c722884c3a228e2fc51d5389d4fec3e10436e797c4ae168b9306154a9179475", + "edge_count": 7784, + "edge_sha256": "602f1b401b10d5a26a0782fee6ca31b1276dd9861ec9006be35d1372c4974e59", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -3694,8 +3694,21 @@ "app.chain.douban -> app.schemas", "app.chain.douban -> app.schemas.context", "app.chain.douban -> app.schemas.types", + "app.chain.download.admission -> app.application", + "app.chain.download.admission -> app.application.download", + "app.chain.download.admission -> app.application.download.admission", + "app.chain.download.admission -> app.chain", + "app.chain.download.admission -> app.chain.download", + "app.chain.download.admission -> app.chain.download.contract", + "app.chain.download.admission -> app.domain", + "app.chain.download.admission -> app.domain.context", + "app.chain.download.admission -> app.domain.episode", + "app.chain.download.admission -> app.schemas", + "app.chain.download.admission -> app.schemas.media", + "app.chain.download.admission -> app.schemas.types", "app.chain.download.batch -> app.application", "app.chain.download.batch -> app.application.download", + "app.chain.download.batch -> app.application.download.admission", "app.chain.download.batch -> app.application.download.selection", "app.chain.download.batch -> app.application.torrent", "app.chain.download.batch -> app.application.torrent.download", @@ -3729,6 +3742,7 @@ "app.chain.download.existence -> app.schemas.types", "app.chain.download.facade -> app.chain", "app.chain.download.facade -> app.chain.download", + "app.chain.download.facade -> app.chain.download.admission", "app.chain.download.facade -> app.chain.download.batch", "app.chain.download.facade -> app.chain.download.existence", "app.chain.download.facade -> app.chain.download.failure", @@ -3793,6 +3807,7 @@ "app.chain.download.selection -> app.application", "app.chain.download.selection -> app.application.configuration", "app.chain.download.selection -> app.application.download", + "app.chain.download.selection -> app.application.download.admission", "app.chain.download.selection -> app.application.download.failures", "app.chain.download.selection -> app.application.torrent", "app.chain.download.selection -> app.application.torrent.download", @@ -3815,6 +3830,8 @@ "app.chain.download.submission -> app.application", "app.chain.download.submission -> app.application.configuration", "app.chain.download.submission -> app.application.directory", + "app.chain.download.submission -> app.application.download", + "app.chain.download.submission -> app.application.download.admission", "app.chain.download.submission -> app.application.torrent", "app.chain.download.submission -> app.application.torrent.download", "app.chain.download.submission -> app.chain", @@ -4521,6 +4538,8 @@ "app.chain.subscribe.notify -> app.schemas.message", "app.chain.subscribe.notify -> app.schemas.types", "app.chain.subscribe.policy -> app.application", + "app.chain.subscribe.policy -> app.application.download", + "app.chain.subscribe.policy -> app.application.download.admission", "app.chain.subscribe.policy -> app.application.subscription", "app.chain.subscribe.policy -> app.application.subscription.contract", "app.chain.subscribe.policy -> app.application.subscription.priority", @@ -4656,6 +4675,7 @@ "app.chain.torrents -> app.domain", "app.chain.torrents -> app.domain.context", "app.chain.torrents -> app.domain.meta", + "app.chain.torrents -> app.domain.meta.metabase", "app.chain.torrents -> app.domain.meta.metamusic", "app.chain.torrents -> app.domain.metainfo", "app.chain.torrents -> app.domain.site", @@ -5146,6 +5166,15 @@ "app.db.adapters.subscription -> app.schemas", "app.db.adapters.subscription -> app.schemas.common", "app.db.adapters.subscription -> app.schemas.types", + "app.db.adapters.subscriptiondownload -> app.application", + "app.db.adapters.subscriptiondownload -> app.application.download", + "app.db.adapters.subscriptiondownload -> app.application.download.admission", + "app.db.adapters.subscriptiondownload -> app.db", + "app.db.adapters.subscriptiondownload -> app.db.models", + "app.db.adapters.subscriptiondownload -> app.db.models.subscriptiondownload", + "app.db.adapters.subscriptiondownload -> app.db.oper", + "app.db.adapters.subscriptiondownload -> app.db.oper.subscriptiondownload", + "app.db.adapters.subscriptiondownload -> app.db.uow", "app.db.adapters.subscriptionsearch -> app.application", "app.db.adapters.subscriptionsearch -> app.application.subscription", "app.db.adapters.subscriptionsearch -> app.application.subscription.execution", @@ -5294,6 +5323,8 @@ "app.db.models.subscribehistory -> app.db.models._constraints", "app.db.models.subscribehistory -> app.schemas", "app.db.models.subscribehistory -> app.schemas.types", + "app.db.models.subscriptiondownload -> app.db", + "app.db.models.subscriptiondownload -> app.db.base", "app.db.models.subscriptionsearch -> app.db", "app.db.models.subscriptionsearch -> app.db.base", "app.db.models.systemconfig -> app.db", @@ -5402,6 +5433,13 @@ "app.db.oper.subscribehistory -> app.schemas.common", "app.db.oper.subscribehistory -> app.schemas.query", "app.db.oper.subscribehistory -> app.schemas.types", + "app.db.oper.subscriptiondownload -> app.application", + "app.db.oper.subscriptiondownload -> app.application.download", + "app.db.oper.subscriptiondownload -> app.application.download.admission", + "app.db.oper.subscriptiondownload -> app.db", + "app.db.oper.subscriptiondownload -> app.db.base", + "app.db.oper.subscriptiondownload -> app.db.models", + "app.db.oper.subscriptiondownload -> app.db.models.subscriptiondownload", "app.db.oper.subscriptionsearch -> app.db", "app.db.oper.subscriptionsearch -> app.db.base", "app.db.oper.subscriptionsearch -> app.db.models", @@ -8044,6 +8082,7 @@ "app.startup.composition.chain -> app.db.adapters.chain", "app.startup.composition.chain -> app.db.adapters.download", "app.startup.composition.chain -> app.db.adapters.mediaserver", + "app.startup.composition.chain -> app.db.adapters.subscriptiondownload", "app.startup.composition.chain -> app.db.adapters.transfer", "app.startup.composition.chain -> app.db.adapters.transfer.admission", "app.startup.composition.chain -> app.db.oper", @@ -8838,7 +8877,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 927, + "module_count": 932, "modules": [ "app", "app.adapters", @@ -9042,6 +9081,7 @@ "app.application.database", "app.application.directory", "app.application.download", + "app.application.download.admission", "app.application.download.failures", "app.application.download.selection", "app.application.download.tasks", @@ -9161,6 +9201,7 @@ "app.chain.dashboard", "app.chain.douban", "app.chain.download", + "app.chain.download.admission", "app.chain.download.batch", "app.chain.download.contract", "app.chain.download.existence", @@ -9266,6 +9307,7 @@ "app.db.adapters.query", "app.db.adapters.site", "app.db.adapters.subscription", + "app.db.adapters.subscriptiondownload", "app.db.adapters.subscriptionsearch", "app.db.adapters.transaction", "app.db.adapters.transfer", @@ -9300,6 +9342,7 @@ "app.db.models.siteuserdata", "app.db.models.subscribe", "app.db.models.subscribehistory", + "app.db.models.subscriptiondownload", "app.db.models.subscriptionsearch", "app.db.models.systemconfig", "app.db.models.transferexecutionstep", @@ -9323,6 +9366,7 @@ "app.db.oper.site", "app.db.oper.subscribe", "app.db.oper.subscribehistory", + "app.db.oper.subscriptiondownload", "app.db.oper.subscriptionsearch", "app.db.oper.systemconfig", "app.db.oper.transferexecutionstep", diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index 2013dd9ea..8c244a64b 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -968,8 +968,8 @@ }, "app/chain/subscribe/match.py": { "arg-type": 4, - "assignment": 2, - "no-untyped-call": 4, + "assignment": 1, + "no-untyped-call": 1, "union-attr": 7 }, "app/chain/subscribe/notify.py": { @@ -990,7 +990,7 @@ "app/chain/subscribe/refresh.py": { "arg-type": 3, "assignment": 2, - "no-redef": 2, + "no-redef": 1, "var-annotated": 2 }, "app/chain/subscribe/search.py": { @@ -1009,9 +1009,8 @@ "type-arg": 2 }, "app/chain/torrents.py": { - "arg-type": 28, - "assignment": 4, - "no-any-return": 1, + "arg-type": 22, + "assignment": 3, "no-untyped-call": 1, "no-untyped-def": 15, "type-arg": 5, @@ -1275,7 +1274,7 @@ "assignment": 86, "no-any-return": 1, "no-untyped-call": 4, - "no-untyped-def": 24, + "no-untyped-def": 23, "operator": 1, "type-arg": 36 }, diff --git a/tests/fixtures/architecture/ruff-baseline.json b/tests/fixtures/architecture/ruff-baseline.json index 1436c7421..d8b491700 100644 --- a/tests/fixtures/architecture/ruff-baseline.json +++ b/tests/fixtures/architecture/ruff-baseline.json @@ -96,9 +96,6 @@ "app/db/diagnostics.py": { "I001": 1 }, - "app/db/models/__init__.py": { - "I001": 1 - }, "app/db/models/agentchat.py": { "I001": 1 }, diff --git a/tests/test_download_architecture.py b/tests/test_download_architecture.py index 30ba6fbc0..8b49dade2 100644 --- a/tests/test_download_architecture.py +++ b/tests/test_download_architecture.py @@ -39,6 +39,7 @@ def test_download_monolith_is_retired_and_package_owners_are_complete() -> None: assert not (CHAIN_ROOT / "download.py").exists() assert {path.name for path in DOWNLOAD_PACKAGE.glob("*.py")} == { "__init__.py", + "admission.py", "batch.py", "contract.py", "existence.py", diff --git a/tests/test_download_compat.py b/tests/test_download_compat.py index 3dda154a0..7973e768d 100644 --- a/tests/test_download_compat.py +++ b/tests/test_download_compat.py @@ -56,6 +56,7 @@ def test_download_single_accepts_existing_plugin_call_shapes() -> None: ) assert signature.parameters["return_detail"].default is False assert signature.parameters["custom_words"].default is None + assert signature.parameters["governance"].default is None def test_batch_download_accepts_existing_plugin_call_shapes() -> None: @@ -73,6 +74,7 @@ def test_batch_download_accepts_existing_plugin_call_shapes() -> None: downloader="qbittorrent", custom_words="S04E05 => S01E170", ) + assert signature.parameters["governance"].default is None def test_get_no_exists_info_accepts_existing_plugin_call_shapes() -> None: diff --git a/tests/test_subscription_download_governance.py b/tests/test_subscription_download_governance.py new file mode 100644 index 000000000..502107cf4 --- /dev/null +++ b/tests/test_subscription_download_governance.py @@ -0,0 +1,464 @@ +"""订阅下载跨入口幂等、不确定终态与取消补偿测试。""" + +from __future__ import annotations + +import importlib +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +import app.chain.download.submission as download_submission +from app.application.download.admission import ( + DownloadReconciliationRequired, + SubscriptionDownloadGovernance, + SubscriptionDownloadRequest, +) +from app.application.subscription.contract import SubscriptionSnapshot +from app.chain.download import DownloadChain +from app.chain.subscribe import policy as subscribe_policy +from app.chain.subscribe.facade import SubscribeChain +from app.db.adapters.subscriptiondownload import TransactionalSubscriptionDownloadRepository +from app.db.base import Base +from app.db.models.subscriptiondownload import SubscriptionDownloadSubmission +from app.domain.context import Context, MediaInfo, TorrentInfo +from app.domain.metainfo import MetaInfo +from app.schemas.types import MediaSource, MediaType + + +def _request(key: str = "key-1", task_id: str | None = "task-1") -> SubscriptionDownloadRequest: + """构造固定身份的提交认领请求。""" + return SubscriptionDownloadRequest( + idempotency_key=key, + subscription_id=7, + task_id=task_id, + logical_identity='{"subscription_id":7}', + resource_key="example.com:id=42", + coverage="episodes:E01-E03", + mode="normal", + ) + + +def _repository(tmp_path) -> tuple[TransactionalSubscriptionDownloadRepository, object]: + """创建独立 SQLite 幂等账本仓储与 Session 工厂。""" + engine = create_engine( + f"sqlite:///{tmp_path / 'subscription-download.db'}", + connect_args={"check_same_thread": False}, + ) + Base.metadata.create_all(engine) + factory = sessionmaker(bind=engine) + return TransactionalSubscriptionDownloadRepository(factory), factory + + +def test_repository_claims_same_submission_once_across_workers(tmp_path) -> None: + """并发入口对同一幂等键只能有一个取得下载器提交权。""" + repository, _factory = _repository(tmp_path) + + with ThreadPoolExecutor(max_workers=2) as executor: + claims = list(executor.map(lambda _index: repository.claim(_request()), range(2))) + + assert sum(claim.acquired for claim in claims) == 1 + assert {claim.snapshot.state for claim in claims} == {"submitting"} + assert {claim.snapshot.attempt_count for claim in claims} == {1} + + +def test_repository_fences_retry_and_preserves_uncertain_terminal(tmp_path) -> None: + """明确拒绝可延迟重试,过期令牌和待对账状态均不得重新提交。""" + repository, _factory = _repository(tmp_path) + first = repository.claim(_request()) + token = first.snapshot.attempt_token + assert token + assert repository.mark_retryable( + idempotency_key="key-1", + attempt_token=token, + available_at="1970-01-01T00:00:00+00:00", + error="downloader rejected", + ) + + second = repository.claim(_request(task_id="task-2")) + second_token = second.snapshot.attempt_token + assert second.acquired + assert second_token and second_token != token + assert not repository.mark_succeeded( + idempotency_key="key-1", + attempt_token=token, + ) + assert repository.mark_reconcile_required( + idempotency_key="key-1", + attempt_token=second_token, + error="transport timeout", + ) + + blocked = repository.claim(_request(task_id="task-3")) + assert not blocked.acquired + assert blocked.snapshot.state == "reconcile_required" + assert blocked.snapshot.attempt_count == 2 + assert repository.has_started_for_task("task-2") + + +class _FakeTorrentHelper: + """返回固定种子目录和文件清单,隔离真实 bencode 解析。""" + + @staticmethod + def get_fileinfo_from_torrent_content(_content): + """返回单文件种子结构。""" + return "Demo.Show.S01", ["Demo.Show.S01E01.mkv"] + + +def _download_chain(repository) -> DownloadChain: + """构造只执行下载提交边界的 Chain 测试实例。""" + chain = DownloadChain.__new__(DownloadChain) + chain.subscription_download_repository = repository + chain.download_history_repository = MagicMock() + chain.download_history_repository.get_by_media_identity.return_value = [] + chain.download_failure_repository = MagicMock() + chain.eventmanager = MagicMock() + chain.eventmanager.send_event.return_value = None + chain.post_message = MagicMock() + chain._settle_download_success = MagicMock() + chain._resolve_media_download_dir = MagicMock( + return_value=("local", Path("/downloads"), None) + ) + chain.runtime_config = SimpleNamespace(media_extensions=(".mkv",)) + return chain + + +def _context() -> Context: + """构造具有稳定媒体和 torrent 身份的电视剧候选。""" + return Context( + meta_info=MetaInfo("Demo Show S01E01"), + media_info=MediaInfo( + media_source=MediaSource.TMDB, + media_id="77", + type=MediaType.TV, + title="Demo Show", + year="2026", + tmdb_id=77, + genre_ids=[18], + ), + torrent_info=TorrentInfo( + site=12, + site_name="TestSite", + title="Demo Show S01E01 1080p", + enclosure="https://example.com/download.php?id=42", + ), + ) + + +@pytest.fixture(autouse=True) +def _submission_dependencies(monkeypatch): + """固定目录、媒体补全和种子解析,避免触发外部模块。""" + monkeypatch.setattr( + "app.application.directory.DirectoryHelper.get_download_dirs", + lambda _self: [SimpleNamespace( + storage="local", + download_path="/downloads", + category=None, + media_type=None, + )], + ) + monkeypatch.setattr(download_submission, "TorrentHelper", _FakeTorrentHelper) + monkeypatch.setattr( + download_submission.MediaChain, + "supplement_tmdb_info", + lambda _self, media, _meta: media, + ) + + +def test_download_chain_reuses_success_without_second_downloader_call(tmp_path) -> None: + """重叠入口在首个提交成功后复用 hash,不再次调用下载器。""" + repository, _factory = _repository(tmp_path) + chain = _download_chain(repository) + chain.download = MagicMock(return_value=("qb", "hash-1", "Original", "accepted")) + governance = SubscriptionDownloadGovernance(subscription_id=7, mode="normal") + + first = chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads", + governance=governance, + ) + second = chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads", + governance=governance, + ) + + assert first == second == "hash-1" + chain.download.assert_called_once() + chain._settle_download_success.assert_called_once() + + +def test_download_chain_freezes_when_local_settlement_fails(tmp_path) -> None: + """下载器接受而历史结算失败时进入待对账,后续入口不得盲重试。""" + repository, _factory = _repository(tmp_path) + chain = _download_chain(repository) + chain.download = MagicMock(return_value=("qb", "hash-2", "Original", "accepted")) + chain._settle_download_success.side_effect = RuntimeError("history unavailable") + governance = SubscriptionDownloadGovernance(subscription_id=7, mode="normal") + + with pytest.raises(DownloadReconciliationRequired): + chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads", + governance=governance, + ) + with pytest.raises(DownloadReconciliationRequired): + chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads", + governance=governance, + ) + + chain.download.assert_called_once() + + +def test_download_chain_freezes_when_downloader_result_is_uncertain(tmp_path) -> None: + """下载器调用抛错可能已产生副作用,重启后的新实例仍不得自动重试。""" + repository, _factory = _repository(tmp_path) + chain = _download_chain(repository) + chain.download = MagicMock(side_effect=TimeoutError("response timeout")) + governance = SubscriptionDownloadGovernance(subscription_id=7, mode="normal") + + with pytest.raises(DownloadReconciliationRequired): + chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads", + governance=governance, + ) + + restarted = _download_chain(repository) + restarted.download = MagicMock() + with pytest.raises(DownloadReconciliationRequired): + restarted.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads", + governance=governance, + ) + restarted.download.assert_not_called() + + +def test_download_chain_delays_retry_after_explicit_rejection(tmp_path) -> None: + """下载器明确拒绝且无 hash 时进入冷却,不把不确定和已拒绝混为一谈。""" + repository, factory = _repository(tmp_path) + chain = _download_chain(repository) + chain.download = MagicMock(return_value=("qb", None, "Original", "downloader rejected")) + governance = SubscriptionDownloadGovernance(subscription_id=7, mode="normal") + + first = chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads", + governance=governance, + ) + second = chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads", + governance=governance, + ) + + assert first is None and second is None + chain.download.assert_called_once() + with factory() as session: + record = session.query(SubscriptionDownloadSubmission).one() + assert record.state == "retryable" + assert record.available_at > record.updated_at + + +def test_download_chain_reads_legacy_history_before_new_ledger(tmp_path) -> None: + """迁移前同订阅同 torrent 同覆盖的成功历史仍能阻止重复提交。""" + repository, factory = _repository(tmp_path) + chain = _download_chain(repository) + chain.download = MagicMock() + chain.download_history_repository.get_by_media_identity.return_value = [ + SimpleNamespace( + download_hash="legacy-hash", + torrent_name="Demo Show S01E01 1080p", + torrent_site="TestSite", + episodes="E01", + seasons="S01", + episode_group=None, + note={"source": 'Subscribe|{"id": 7}'}, + ) + ] + + result = chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads", + governance=SubscriptionDownloadGovernance(subscription_id=7, mode="normal"), + ) + + assert result == "legacy-hash" + chain.download.assert_not_called() + with factory() as session: + assert session.query(SubscriptionDownloadSubmission).count() == 0 + + +def test_download_chain_cancels_before_external_side_effect(tmp_path) -> None: + """取消在下载器边界前生效时不得创建下载任务或成功事实。""" + repository, _factory = _repository(tmp_path) + chain = _download_chain(repository) + chain.download = MagicMock() + governance = SubscriptionDownloadGovernance( + subscription_id=7, + mode="normal", + task_id="cancel-task", + cancelled=lambda: True, + ) + + result = chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads", + governance=governance, + ) + + assert result is None + chain.download.assert_not_called() + assert not repository.has_started_for_task("cancel-task") + + +def test_subscription_policy_reloads_facts_and_threads_governance(monkeypatch) -> None: + """提交策略必须使用当前订阅和重算缺集,而不是直接消费准备阶段快照。""" + prepared = SubscriptionSnapshot( + id=7, + name="Demo Show", + type=MediaType.TV.value, + media_source=MediaSource.TMDB, + media_id="77", + season=1, + state="R", + save_path="/old", + downloader="old", + ) + current = SubscriptionSnapshot( + **{ + **prepared.to_dict(), + "save_path": "/current", + "downloader": "current", + "lack_episode": 1, + } + ) + context = _context() + stale_missing = {"themoviedb:77": {1: SimpleNamespace(episodes=[1, 2])}} + fresh_missing = {"themoviedb:77": {1: SimpleNamespace(episodes=[1])}} + captured = {} + + class _FakeDownloadChain: + """捕获提交策略传入的当前事实与治理上下文。""" + + def batch_download(self, **kwargs): + """记录一次批量提交并返回当前缺集。""" + captured.update(kwargs) + return [], kwargs["no_exists"] + + chain = SubscribeChain.__new__(SubscribeChain) + chain.subscription_repository = SimpleNamespace(get=lambda _subscribe_id: current) + chain.check_and_handle_existing_media = MagicMock(return_value=(False, fresh_missing)) + monkeypatch.setattr( + chain, + "_SubscribeChain__revalidate_download_contexts", + lambda _subscribe, contexts: contexts, + ) + monkeypatch.setattr(subscribe_policy, "DownloadChain", _FakeDownloadChain) + + _downloads, lefts = chain._SubscribeChain__download_best_version_with_full_pack_first( + contexts=[context], + no_exists=stale_missing, + subscribe=prepared, + mediakey="themoviedb:77", + save_path="/old", + downloader="old", + ) + + assert lefts is fresh_missing + chain.check_and_handle_existing_media.assert_called_once_with( + subscribe=current, + meta=context.meta_info, + mediainfo=context.media_info, + mediakey="themoviedb:77", + ) + assert captured["no_exists"] is fresh_missing + assert captured["save_path"] == "/current" + assert captured["downloader"] == "current" + assert captured["governance"].subscription_id == 7 + assert captured["governance"].mode == "normal" + + +def test_subscription_policy_discards_candidates_after_filter_change(monkeypatch) -> None: + """准备后筛选合同变化时必须放弃旧候选,不允许用旧快照提交。""" + prepared = SubscriptionSnapshot( + id=7, + name="Demo Show", + type=MediaType.TV.value, + media_source=MediaSource.TMDB, + media_id="77", + season=1, + state="R", + quality="1080p", + ) + current = SubscriptionSnapshot(**{**prepared.to_dict(), "quality": "2160p"}) + chain = SubscribeChain.__new__(SubscribeChain) + chain.subscription_repository = SimpleNamespace(get=lambda _subscribe_id: current) + batch_download = MagicMock(side_effect=AssertionError("stale candidate submitted")) + monkeypatch.setattr(subscribe_policy, "DownloadChain", lambda: SimpleNamespace( + batch_download=batch_download, + )) + + downloads, lefts = chain._SubscribeChain__download_best_version_with_full_pack_first( + contexts=[_context()], + no_exists={"themoviedb:77": {}}, + subscribe=prepared, + mediakey="themoviedb:77", + ) + + assert downloads == [] + assert lefts == {"themoviedb:77": {}} + batch_download.assert_not_called() + + +def test_subscription_download_migration_is_idempotent_and_reversible(tmp_path, monkeypatch) -> None: + """3.0.21 迁移可重复升级,并只移除自身新增的兼容表。""" + engine = create_engine(f"sqlite:///{tmp_path / 'migration.db'}") + migration = importlib.import_module("database.versions.e1b6d4f8a2c7_3_0_21") + with engine.begin() as connection: + operations = Operations(MigrationContext.configure(connection)) + monkeypatch.setattr(migration, "op", operations) + migration.upgrade() + migration.upgrade() + inspector = sa.inspect(connection) + assert "subscriptiondownloadsubmission" in inspector.get_table_names() + indexes = {item["name"] for item in inspector.get_indexes("subscriptiondownloadsubmission")} + assert "ix_subscriptiondownloadsubmission_task_state" in indexes + migration.downgrade() + assert "subscriptiondownloadsubmission" not in sa.inspect(connection).get_table_names() + + +def test_model_metadata_registers_submission_table() -> None: + """显式模型注册必须让 fresh create_all 包含订阅提交账本。""" + assert SubscriptionDownloadSubmission.__tablename__ == "subscriptiondownloadsubmission" + assert "subscriptiondownloadsubmission" in Base.metadata.tables diff --git a/tests/test_subscription_governance_replay.py b/tests/test_subscription_governance_replay.py index ecc1f9cf6..e2f8410fd 100644 --- a/tests/test_subscription_governance_replay.py +++ b/tests/test_subscription_governance_replay.py @@ -14,7 +14,6 @@ from app.domain.metainfo import MetaInfo from app.schemas.mediaserver import NotExistMediaInfo from app.schemas.types import MediaSource, MediaType - _FIXTURE_PATH = Path(__file__).parent / "fixtures" / "subscription_governance_match_replay.json" diff --git a/tests/test_subscription_search_governance.py b/tests/test_subscription_search_governance.py index 1826f90a3..9d27c8779 100644 --- a/tests/test_subscription_search_governance.py +++ b/tests/test_subscription_search_governance.py @@ -2,6 +2,7 @@ from dataclasses import replace from datetime import datetime, timedelta +from types import SimpleNamespace from unittest.mock import Mock, patch from sqlalchemy import create_engine @@ -116,3 +117,30 @@ def test_fallback_queue_continues_after_one_subscription_failure(tmp_path, monke assert batch.finished_count == 1 assert batch.failed_count == 1 assert batch.last_error == "provider timeout" + + +def test_late_cancel_completes_when_download_submission_already_started(tmp_path, monkeypatch): + """取消晚于下载器副作用边界时按真实结果完成,不能伪装成未执行取消。""" + subscribe = _subscribe(5) + chain = _chain(tmp_path, [subscribe]) + monkeypatch.setattr(chain, "_search_batch_available_at", lambda _source: "1970-01-01T00:00:00+00:00") + chain.subscription_download_repository = SimpleNamespace( + has_started_for_task=lambda _task_id: False, + ) + queue = chain.subscription_search_repository + cancel_checks = iter((False, True)) + monkeypatch.setattr(queue, "is_cancel_requested", lambda _task_id: next(cancel_checks)) + def process(item, _searchchain): + """模拟当前任务复用或完成下载后才收到取消。""" + chain._mark_subscription_download_started() + return item + + monkeypatch.setattr(chain, "_process_search_subscription", process) + + with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()): + batch_id = chain.search(state="R") + + batch = chain.get_search_batch(batch_id) + assert batch.state == "completed" + assert batch.finished_count == 1 + assert batch.cancelled_count == 0 diff --git a/tests/test_subscription_site_budget.py b/tests/test_subscription_site_budget.py index 3487d07aa..f783fd426 100644 --- a/tests/test_subscription_site_budget.py +++ b/tests/test_subscription_site_budget.py @@ -15,10 +15,10 @@ from app.application.subscription.sitebudget import ( SubscriptionSearchCancelled, SubscriptionSiteBudget, ) +from app.chain.search.facade import SearchChain from app.db.adapters.subscriptionsearch import TransactionalSubscriptionSearchRepository from app.db.base import Base from app.db.models.subscriptionsearch import SubscriptionSiteBudget as SiteBudgetRecord -from app.chain.search.facade import SearchChain from app.modules.indexer import _classify_search_failure from app.runtime.stop import ProcessStopState