diff --git a/app/application/download/admission.py b/app/application/download/admission.py index 2e4665316..13174e144 100644 --- a/app/application/download/admission.py +++ b/app/application/download/admission.py @@ -26,12 +26,14 @@ class SubscriptionDownloadRequest: """一次订阅下载提交认领所需的规范身份。""" idempotency_key: str + legacy_idempotency_key: Optional[str] subscription_id: int task_id: Optional[str] logical_identity: str resource_key: str coverage: str mode: str + delivery_scope: str @dataclass(frozen=True, slots=True) @@ -67,6 +69,10 @@ class SubscriptionDownloadRepository(Protocol): """按唯一键认领提交；仅到期 retryable/cancelled 状态允许重新认领。""" ... + def get(self, idempotency_key: str) -> Optional[SubscriptionDownloadSnapshot]: + """按幂等键读取现有提交，供键版本兼容检查。""" + ... + def mark_accepted( self, *, diff --git a/app/chain/download/admission.py b/app/chain/download/admission.py index 51f845d76..ea31abfcc 100644 --- a/app/chain/download/admission.py +++ b/app/chain/download/admission.py @@ -30,8 +30,9 @@ class DownloadAdmissionOwner(_DownloadOwnerBase): context: Context, episodes: Optional[Set[int]], governance: SubscriptionDownloadGovernance, + delivery_scope: str, ) -> SubscriptionDownloadRequest: - """组合订阅、torrent、季集覆盖与模式生成规范幂等请求。""" + """组合逻辑媒体、资源、覆盖、模式和交付目标生成规范幂等请求。""" media = context.media_info meta = context.meta_info torrent = context.torrent_info @@ -53,11 +54,11 @@ class DownloadAdmissionOwner(_DownloadOwnerBase): 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), + "music_type": getattr(media, "music_type", None), }, ensure_ascii=False, sort_keys=True, @@ -69,18 +70,44 @@ class DownloadAdmissionOwner(_DownloadOwnerBase): "resource_key": resource_key, "coverage": coverage, "mode": governance.mode, + "delivery_scope": delivery_scope, + }, + ensure_ascii=False, + sort_keys=True, + ) + legacy_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, + ) + legacy_canonical = json.dumps( + { + "logical_identity": legacy_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(), + legacy_idempotency_key=hashlib.sha256( + legacy_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, + delivery_scope=delivery_scope, ) def _claim_subscription_download( @@ -89,6 +116,8 @@ class DownloadAdmissionOwner(_DownloadOwnerBase): context: Context, episodes: Optional[Set[int]], governance: Optional[SubscriptionDownloadGovernance], + downloader: Optional[str], + download_uri: str, ) -> tuple[Optional[SubscriptionDownloadClaim], Optional[str]]: """在下载器调用前认领唯一提交权，并返回已成功提交的历史 hash。""" if governance is None: @@ -107,7 +136,19 @@ class DownloadAdmissionOwner(_DownloadOwnerBase): context=context, episodes=episodes, governance=governance, + delivery_scope=self._subscription_delivery_scope( + downloader=downloader, + download_uri=download_uri, + ), ) + if request.legacy_idempotency_key: + legacy = repository.get(request.legacy_idempotency_key) + if legacy and legacy.state == "succeeded" and legacy.download_hash: + return None, legacy.download_hash + if legacy and legacy.state in {"submitting", "accepted", "reconcile_required"}: + raise DownloadReconciliationRequired( + f"订阅下载提交 {legacy.idempotency_key} 当前为 {legacy.state}，需要先对账下载器" + ) claim = repository.claim(request) snapshot = claim.snapshot if claim.acquired: @@ -122,6 +163,22 @@ class DownloadAdmissionOwner(_DownloadOwnerBase): ) return claim, None + @staticmethod + def _subscription_delivery_scope( + *, + downloader: Optional[str], + download_uri: str, + ) -> str: + """规范化实际下载器和保存目标，限定跨记录去重的产品边界。""" + return json.dumps( + { + "downloader": downloader or "auto", + "download_uri": download_uri, + }, + ensure_ascii=False, + sort_keys=True, + ) + def _legacy_subscription_download_hash( self, *, diff --git a/app/chain/download/submission.py b/app/chain/download/submission.py index 0ede8c0f1..221dbc2c8 100644 --- a/app/chain/download/submission.py +++ b/app/chain/download/submission.py @@ -425,6 +425,8 @@ class DownloadSubmissionOwner(_DownloadOwnerBase): context=context, episodes=episodes, governance=governance, + downloader=downloader or _site_downloader, + download_uri=file_uri.uri, ) if duplicate_hash: if governance and governance.mark_started: diff --git a/app/db/adapters/subscriptiondownload.py b/app/db/adapters/subscriptiondownload.py index 896acce2c..975e063a5 100644 --- a/app/db/adapters/subscriptiondownload.py +++ b/app/db/adapters/subscriptiondownload.py @@ -68,6 +68,16 @@ class TransactionalSubscriptionDownloadRepository: return self._write(operation) + def get(self, idempotency_key: str) -> Optional[SubscriptionDownloadSnapshot]: + """读取一个已存在的提交快照。""" + return self._read( + lambda repository: ( + _snapshot(record) + if (record := repository.get(idempotency_key)) is not None + else None + ) + ) + def mark_accepted( self, *, diff --git a/app/db/models/subscriptiondownload.py b/app/db/models/subscriptiondownload.py index 68d8f3c2a..d666d0ae4 100644 --- a/app/db/models/subscriptiondownload.py +++ b/app/db/models/subscriptiondownload.py @@ -19,6 +19,7 @@ class SubscriptionDownloadSubmission(Base): 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) + delivery_scope: Mapped[str] = mapped_column(Text, nullable=False, default="legacy") 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)) diff --git a/app/db/oper/subscriptiondownload.py b/app/db/oper/subscriptiondownload.py index 97f7292e8..1d5e32272 100644 --- a/app/db/oper/subscriptiondownload.py +++ b/app/db/oper/subscriptiondownload.py @@ -43,6 +43,7 @@ class SubscriptionDownloadOper(DbOper): resource_key=request.resource_key, coverage=request.coverage, mode=request.mode, + delivery_scope=request.delivery_scope, state="submitting", attempt_count=1, attempt_token=attempt_token, @@ -101,6 +102,16 @@ class SubscriptionDownloadOper(DbOper): ).scalar_one() return current, bool(updated and current.attempt_token == attempt_token) + def get(self, idempotency_key: str) -> Optional[SubscriptionDownloadSubmission]: + """按稳定幂等键读取提交记录。""" + if not isinstance(self._db, Session): + raise RuntimeError("订阅下载提交查询需要调用方提供同步 Session") + return self._db.execute( + select(SubscriptionDownloadSubmission).where( + SubscriptionDownloadSubmission.idempotency_key == idempotency_key + ) + ).scalars().first() + def mark_accepted( self, *, diff --git a/database/versions/a7d9e2c4f6b1_3_0_23.py b/database/versions/a7d9e2c4f6b1_3_0_23.py new file mode 100644 index 000000000..d95cb8f88 --- /dev/null +++ b/database/versions/a7d9e2c4f6b1_3_0_23.py @@ -0,0 +1,48 @@ +"""3.0.23 增加订阅下载交付目标范围。 + +Revision ID: a7d9e2c4f6b1 +Revises: f3c8a1d6b2e9 +Create Date: 2026-09-01 +""" + +# Alembic 的 op 是运行期代理，静态分析无法看到实际操作方法。 +# pylint: disable=no-member + +import sqlalchemy as sa +from alembic import op + +revision = "a7d9e2c4f6b1" +down_revision = "f3c8a1d6b2e9" +branch_labels = None +depends_on = None + +_TABLE = "subscriptiondownloadsubmission" + + +def _column_names() -> set[str]: + """返回当前订阅下载提交列名集合。""" + inspector = sa.inspect(op.get_bind()) + if _TABLE not in set(inspector.get_table_names()): + return set() + return {column["name"] for column in inspector.get_columns(_TABLE)} + + +def upgrade() -> None: + """为存量提交增加兼容交付范围，新键会写入实际下载目标。""" + columns = _column_names() + if columns and "delivery_scope" not in columns: + op.add_column( + _TABLE, + sa.Column( + "delivery_scope", + sa.Text(), + nullable=False, + server_default="legacy", + ), + ) + + +def downgrade() -> None: + """移除交付范围字段，保留原提交账本与唯一键。""" + if "delivery_scope" in _column_names(): + op.drop_column(_TABLE, "delivery_scope") diff --git a/tests/test_subscription_download_governance.py b/tests/test_subscription_download_governance.py index 502107cf4..b92ff4763 100644 --- a/tests/test_subscription_download_governance.py +++ b/tests/test_subscription_download_governance.py @@ -37,12 +37,14 @@ def _request(key: str = "key-1", task_id: str | None = "task-1") -> Subscription """构造固定身份的提交认领请求。""" return SubscriptionDownloadRequest( idempotency_key=key, + legacy_idempotency_key=None, subscription_id=7, task_id=task_id, logical_identity='{"subscription_id":7}', resource_key="example.com:id=42", coverage="episodes:E01-E03", mode="normal", + delivery_scope='{"download_uri":"local:/downloads","downloader":"auto"}', ) @@ -199,6 +201,173 @@ def test_download_chain_reuses_success_without_second_downloader_call(tmp_path) chain._settle_download_success.assert_called_once() +def test_download_chain_deduplicates_same_media_across_subscription_rows(tmp_path) -> None: + """同媒体同覆盖同交付目标的重复订阅只允许一个下载器提交。""" + repository, _factory = _repository(tmp_path) + chain = _download_chain(repository) + chain.download = MagicMock(return_value=("qb", "hash-cross-row", "Original", "accepted")) + + first = chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads", + governance=SubscriptionDownloadGovernance(subscription_id=7, mode="normal"), + ) + second = chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads", + governance=SubscriptionDownloadGovernance(subscription_id=8, mode="normal"), + ) + + assert first == second == "hash-cross-row" + chain.download.assert_called_once() + chain._settle_download_success.assert_called_once() + + +def test_download_chain_keeps_distinct_delivery_targets_separate(tmp_path) -> None: + """不同保存目标属于独立产品意图，不得跨记录误去重。""" + repository, _factory = _repository(tmp_path) + chain = _download_chain(repository) + chain._resolve_media_download_dir.side_effect = ( + lambda *, save_path, **_kwargs: ("local", Path(save_path), None) + ) + chain.download = MagicMock(side_effect=[ + ("qb", "hash-a", "Original", "accepted"), + ("qb", "hash-b", "Original", "accepted"), + ]) + + first = chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads/a", + governance=SubscriptionDownloadGovernance(subscription_id=7, mode="normal"), + ) + second = chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads/b", + governance=SubscriptionDownloadGovernance(subscription_id=8, mode="normal"), + ) + + assert (first, second) == ("hash-a", "hash-b") + assert chain.download.call_count == 2 + + +def test_download_chain_keeps_distinct_downloaders_separate(tmp_path) -> None: + """不同下载器属于独立交付策略，不得跨记录误去重。""" + repository, _factory = _repository(tmp_path) + chain = _download_chain(repository) + chain.download = MagicMock(side_effect=[ + ("qb-a", "hash-a", "Original", "accepted"), + ("qb-b", "hash-b", "Original", "accepted"), + ]) + + first = chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads", + downloader="qb-a", + governance=SubscriptionDownloadGovernance(subscription_id=7, mode="normal"), + ) + second = chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads", + downloader="qb-b", + governance=SubscriptionDownloadGovernance(subscription_id=8, mode="normal"), + ) + + assert (first, second) == ("hash-a", "hash-b") + assert chain.download.call_count == 2 + + +def test_download_chain_keeps_distinct_episode_coverage_separate(tmp_path) -> None: + """跨记录仅复用精确覆盖，新增目标集不得被已有子集吞掉。""" + repository, _factory = _repository(tmp_path) + chain = _download_chain(repository) + chain.download = MagicMock(side_effect=[ + ("qb", "hash-e1", "Original", "accepted"), + ("qb", "hash-e12", "Original", "accepted"), + ]) + + first = chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads", + governance=SubscriptionDownloadGovernance(subscription_id=7, mode="normal"), + ) + second = chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1, 2}, + save_path="/downloads", + governance=SubscriptionDownloadGovernance(subscription_id=8, mode="normal"), + ) + + assert (first, second) == ("hash-e1", "hash-e12") + assert chain.download.call_count == 2 + + +def test_download_chain_reuses_pre_004_ledger_key(tmp_path) -> None: + """键升级后仍应读取 003A 同记录成功账本，避免版本升级造成重复下载。""" + repository, _factory = _repository(tmp_path) + chain = _download_chain(repository) + governance = SubscriptionDownloadGovernance(subscription_id=7, mode="normal") + request = chain._build_subscription_download_request( + context=_context(), + episodes={1}, + governance=governance, + delivery_scope=chain._subscription_delivery_scope( + downloader=None, + download_uri="local:/downloads", + ), + ) + legacy = repository.claim( + SubscriptionDownloadRequest( + idempotency_key=request.legacy_idempotency_key or "", + legacy_idempotency_key=None, + subscription_id=7, + task_id=None, + logical_identity='{"subscription_id":7}', + resource_key=request.resource_key, + coverage=request.coverage, + mode=request.mode, + delivery_scope="legacy", + ) + ) + token = legacy.snapshot.attempt_token or "" + assert repository.mark_accepted( + idempotency_key=legacy.snapshot.idempotency_key, + attempt_token=token, + downloader="qb", + download_hash="legacy-ledger-hash", + ) + assert repository.mark_succeeded( + idempotency_key=legacy.snapshot.idempotency_key, + attempt_token=token, + ) + chain.download = MagicMock() + + result = chain.download_single( + context=_context(), + torrent_content=b"torrent", + episodes={1}, + save_path="/downloads", + governance=governance, + ) + + assert result == "legacy-ledger-hash" + chain.download.assert_not_called() + + def test_download_chain_freezes_when_local_settlement_fails(tmp_path) -> None: """下载器接受而历史结算失败时进入待对账，后续入口不得盲重试。""" repository, _factory = _repository(tmp_path) @@ -458,6 +627,40 @@ def test_subscription_download_migration_is_idempotent_and_reversible(tmp_path, assert "subscriptiondownloadsubmission" not in sa.inspect(connection).get_table_names() +def test_subscription_delivery_scope_migration_is_idempotent_and_reversible( + tmp_path, + monkeypatch, +) -> None: + """3.0.23 为存量提交填充兼容范围，并支持重复升级和完整回滚。""" + engine = create_engine(f"sqlite:///{tmp_path / 'delivery-migration.db'}") + base_migration = importlib.import_module("database.versions.e1b6d4f8a2c7_3_0_21") + migration = importlib.import_module("database.versions.a7d9e2c4f6b1_3_0_23") + with engine.begin() as connection: + operations = Operations(MigrationContext.configure(connection)) + monkeypatch.setattr(base_migration, "op", operations) + monkeypatch.setattr(migration, "op", operations) + base_migration.upgrade() + connection.execute(sa.text( + "INSERT INTO subscriptiondownloadsubmission " + "(idempotency_key, subscription_id, logical_identity, resource_key, coverage, mode, " + "state, attempt_count, created_at, updated_at) VALUES " + "('legacy-key', 7, '{}', 'resource', 'full', 'normal', 'succeeded', 1, 'now', 'now')" + )) + + migration.upgrade() + migration.upgrade() + + assert connection.execute(sa.text( + "SELECT delivery_scope FROM subscriptiondownloadsubmission WHERE idempotency_key='legacy-key'" + )).scalar_one() == "legacy" + migration.downgrade() + columns = { + column["name"] + for column in sa.inspect(connection).get_columns("subscriptiondownloadsubmission") + } + assert "delivery_scope" not in columns + + def test_model_metadata_registers_submission_table() -> None: """显式模型注册必须让 fresh create_all 包含订阅提交账本。""" assert SubscriptionDownloadSubmission.__tablename__ == "subscriptiondownloadsubmission"