refactor(subscribe): deduplicate duplicate media records

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,
         *,

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,
         *,

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:

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,
         *,

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))

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,
         *,

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")

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"