fix(subscribe): restore strict media identity checks

app/application/subscription/candidates.py

@@ -44,7 +44,7 @@ class CandidateIndex:
                 elif meta_identity:
                     self._explicit_identity[position] = meta_identity
                 else:
-                    # 标题解析未携带显式 ID 的识别冲突仍可能通过同作品证据复核。
+                    # 无显式标题 ID 的候选保守路由，最终匹配仍须严格核对识别身份。
                     self._reconcilable.add(position)
 
     def select_cache_candidates(
@@ -77,7 +77,7 @@ class CandidateIndex:
         """
         保守路由可能命中的候选，并在外部事实查询前应用站点范围。
 
-        无显式 ID 的候选识别类型可由后续严格复核纠正，不能据此提前排除；
+        路由只收集候选，不代表允许下载；媒体身份和类型由最终匹配严格核对。
         标题解析出的类型和季范围仍须符合订阅要求。
         """
         if subscribe.custom_words:

app/chain/subscribe/match.py

@@ -192,84 +192,35 @@ def _reconcile_candidate_media(
     context: Context,
 ) -> Optional[MediaInfo]:
     """
-    在推断媒体 ID 或类型冲突时，以订阅目标对候选标题做严格复核。
+    严格核对候选自身的媒体身份和类型，冲突时直接拒绝。
 
-    标题携带明确媒体身份或目标缺少可比较身份时保持严格拒绝；只有普通
-    标题推断产生冲突且标题、别名、类型和同作品证据复核通过时才回填订阅目标。
-    电影和电视剧可能共用数字 ID，因此身份相同时仍须复核类型冲突。
+    同名真人、动漫及翻拍作品可能拥有相同标题、别名或年份，不能据此覆盖
+    已识别出的不同媒体 ID。TMDB 的电影和电视剧数字 ID 也不可跨类型比较。
+
+    :param target_mediainfo: 订阅目标媒体信息
+    :param candidate_mediainfo: 候选自身的识别结果，通过时保留原对象
+    :param torrent_meta: 为既有调用入口保留的标题元数据，不再用于覆盖候选身份
+    :param torrent_info: 候选资源信息，用于记录拒绝原因
+    :param context: 为既有调用入口保留的上下文，本方法不修改其匹配依据或媒体信息
+    :return: 身份和类型未冲突的候选；冲突时返回 None
     """
+    if candidate_mediainfo.type != target_mediainfo.type:
+        logger.debug(
+            f"{torrent_info.site_name} - {torrent_info.title} "
+            f"媒体类型 {(candidate_mediainfo.type or MediaType.UNKNOWN).value} != "
+            f"{(target_mediainfo.type or MediaType.UNKNOWN).value}，已跳过"
+        )
+        return None
+
     candidate_source, candidate_id = resolve_media_identity(media=candidate_mediainfo)
     target_source, target_id = resolve_media_identity(media=target_mediainfo)
-    type_conflict = candidate_mediainfo.type != target_mediainfo.type
-    conflicts = []
-    if candidate_source and candidate_id and (
-        candidate_source != target_source or candidate_id != target_id or type_conflict
-    ):
-        conflicts.append(
-            (
-                str(candidate_source),
-                candidate_id,
-                f"{target_source}:{target_id}" if target_source and target_id else None,
-            )
-        )
-    if not conflicts:
-        return candidate_mediainfo
-
-    conflict_text = "、".join(
-        f"{source} {candidate_id} != {target_id if target_id is not None else '缺失'}"
-        for source, candidate_id, target_id in conflicts
-    )
-    if type_conflict:
-        conflict_text = (
-            f"媒体类型 {(candidate_mediainfo.type or MediaType.UNKNOWN).value} != "
-            f"{(target_mediainfo.type or MediaType.UNKNOWN).value}；{conflict_text}"
-        )
-    if any(target_id is None for _, _, target_id in conflicts):
+    if candidate_id and (candidate_source, candidate_id) != (target_source, target_id):
         logger.debug(
-            f"{torrent_info.site_name} - {torrent_info.title} 候选媒体ID与订阅目标不可比较：{conflict_text}"
+            f"{torrent_info.site_name} - {torrent_info.title} 候选媒体ID与订阅目标不匹配："
+            f"{candidate_source}:{candidate_id} != {target_source}:{target_id}，已跳过"
         )
         return None
-
-    explicit_source, explicit_id = resolve_media_identity(media=torrent_meta)
-    if explicit_source and explicit_id:
-        logger.debug(
-            f"{torrent_info.site_name} - {torrent_info.title} 标题含明确媒体ID "
-            f"{explicit_source}:{explicit_id}，保持严格校验：{conflict_text}"
-        )
-        return None
-
-    if not TorrentHelper.match_torrent(
-        mediainfo=target_mediainfo,
-        torrent_meta=torrent_meta,
-        torrent=torrent_info,
-    ):
-        logger.debug(
-            f"{torrent_info.site_name} - {torrent_info.title} 候选媒体ID冲突且标题复核失败：{conflict_text}"
-        )
-        return None
-
-    evidence_matched, evidence = TorrentHelper.match_same_work_evidence(
-        target_mediainfo=target_mediainfo,
-        candidate_mediainfo=candidate_mediainfo,
-        torrent_meta=torrent_meta,
-    )
-    if not evidence_matched:
-        logger.debug(
-            f"{torrent_info.site_name} - {torrent_info.title} 候选媒体ID冲突且缺少同作品证据："
-            f"{evidence}；{conflict_text}"
-        )
-        return None
-
-    context.media_info = target_mediainfo
-    context.match_source = "title"
-    context.candidate_recognized = False
-    context.media_info_is_target = True
-    logger.debug(
-        f"{target_mediainfo.title_year} 候选媒体ID冲突（{conflict_text}），"
-        f"经标题或别名及{evidence}复核匹配到订阅目标："
-        f"{torrent_info.site_name} - {torrent_info.title}"
-    )
-    return target_mediainfo
+    return candidate_mediainfo
 
 
 def _prepare_subscription_match(

docs/testing.md

@@ -173,6 +173,11 @@ with patch.object(SomeModule, "fetch", new=AsyncMock(return_value=FAKE)):
 验证双语作品名及艺人署名只有在括号两侧均属于同一实体的可信名称时才匹配；
 未知别名、副标题、不同录音版本及歌曲所属专辑继续保留各自的身份边界。
 
+`tests/test_subscribe_media_id_conflict.py` 与 `tests/test_subscribe_type_conflict.py` 验证
+订阅候选已识别出的媒体来源、ID 或类型冲突时直接拒绝，不以标题、别名或年份覆盖身份。
+《剑来》同名真人与动漫的离线回归覆盖资源年份一致、共同首播年份及缺失年份，断言动漫
+不会进入下载且共享候选不被回填；身份一致的资源和未识别出身份的标题兜底仍可下载。
+
 普通电视剧首次下载的整包优先回归见 `tests/test_subscribe_paged_search.py`：订阅没有下载记录，
 且最新媒体库缺失全部目标集时，先收集所选搜索范围内的候选再尝试完整包；单页策略仍只查每个
 来源的第一页。整包继续通过规则过滤、优先级门槛和种子文件完整覆盖校验，无可用整包时按集

tests/test_subscribe_media_id_conflict.py

@@ -4,6 +4,8 @@
 
 from types import SimpleNamespace
 
+import pytest
+
 from app.chain.subscribe.facade import SubscribeChain
 from app.domain.context import Context, MediaInfo, TorrentInfo
 from app.schemas.types import MediaSource, MediaType
@@ -97,9 +99,9 @@ def _reconcile(target, candidate, meta, torrent, context):
     )
 
 
-def test_inferred_tmdb_conflict_falls_back_to_strict_alias_match():
+def test_inferred_tmdb_conflict_rejects_matching_alias_and_year():
     """
-    标题推断到重复 TMDB 条目时应允许订阅目标别名复核。
+    别名和首播年份一致也不能覆盖候选自身的不同 TMDB ID。
     """
     target = _target_media()
     candidate = _candidate_media()
@@ -113,11 +115,11 @@ def test_inferred_tmdb_conflict_falls_back_to_strict_alias_match():
 
     result = _reconcile(target, candidate, meta, torrent, context)
 
-    assert result is target
-    assert context.media_info is target
-    assert context.match_source == "title"
-    assert context.candidate_recognized is False
-    assert context.media_info_is_target is True
+    assert result is None
+    assert context.media_info is candidate
+    assert context.match_source == "tmdbid"
+    assert context.candidate_recognized is True
+    assert context.media_info_is_target is False
 
 
 def test_no_year_alias_rejects_candidate_with_different_first_air_year():
@@ -153,9 +155,9 @@ def test_no_year_alias_rejects_candidate_with_different_first_air_year():
     assert context.match_source == "tmdbid"
 
 
-def test_explicit_target_year_can_override_wrong_same_name_candidate():
+def test_explicit_target_year_cannot_override_other_media_identity():
     """
-    资源明确携带目标年份时应允许纠正同名候选的错误识别。
+    资源年份与订阅一致时，仍须拒绝已识别为不同作品的候选。
     """
     target = _target_media(
         tmdb_id=236356,
@@ -182,9 +184,9 @@ def test_explicit_target_year_can_override_wrong_same_name_candidate():
     )
     context = _context(meta, candidate, torrent)
 
-    assert _reconcile(target, candidate, meta, torrent, context) is target
-    assert context.media_info is target
-    assert context.match_source == "title"
+    assert _reconcile(target, candidate, meta, torrent, context) is None
+    assert context.media_info is candidate
+    assert context.match_source == "tmdbid"
 
 
 def test_inferred_tmdb_conflict_rejects_nonmatching_title():
@@ -208,9 +210,9 @@ def test_inferred_tmdb_conflict_rejects_nonmatching_title():
     assert context.media_info_is_target is False
 
 
-def test_inferred_douban_conflict_falls_back_to_strict_alias_match():
+def test_inferred_douban_conflict_rejects_matching_alias():
     """
-    豆瓣 ID 推断冲突时应使用同一套严格标题复核。
+    豆瓣 ID 冲突同样不能由相同别名覆盖。
     """
     target = _target_media(tmdb_id=None, douban_id="30170816")
     candidate = _candidate_media(tmdb_id=None, douban_id="36612345")
@@ -224,9 +226,9 @@ def test_inferred_douban_conflict_falls_back_to_strict_alias_match():
 
     result = _reconcile(target, candidate, meta, torrent, context)
 
-    assert result is target
-    assert context.media_info is target
-    assert context.match_source == "title"
+    assert result is None
+    assert context.media_info is candidate
+    assert context.match_source == "tmdbid"
 
 
 def test_explicit_tmdb_identity_keeps_strict_conflict_rejection():
@@ -287,3 +289,25 @@ def test_matching_media_id_preserves_candidate_identity():
     assert context.match_source == "tmdbid"
     assert context.candidate_recognized is True
     assert context.media_info_is_target is False
+
+
+@pytest.mark.parametrize("candidate_source", [MediaSource.Douban, MediaSource.Bangumi, MediaSource.AniList])
+def test_same_numeric_id_from_different_source_is_rejected(candidate_source: MediaSource) -> None:
+    """
+    不同媒体来源中的相同数字 ID 不是同一作品，标题一致也不能跨来源放行。
+    """
+    target = _target_media()
+    candidate = _candidate_media(tmdb_id=target.tmdb_id)
+    candidate.media_source = candidate_source
+    meta = _torrent_meta()
+    torrent = TorrentInfo(
+        title="A Record Of A Mortals Journey To Immortality S01E186",
+        site_name="测试站点",
+        category=MediaType.TV.value,
+    )
+    context = _context(meta, candidate, torrent)
+
+    assert _reconcile(target, candidate, meta, torrent, context) is None
+    assert context.media_info is candidate
+    assert context.candidate_recognized is True
+    assert context.media_info_is_target is False

tests/test_subscribe_type_conflict.py

@@ -1,5 +1,5 @@
 """
-验证订阅候选被推断成其他媒体类型时的严格复核与缓存隔离。
+验证订阅候选的媒体身份与类型冲突不会进入下载或污染缓存。
 """
 
 from collections.abc import Iterator
@@ -142,6 +142,73 @@ def _reconcile(case: SimpleNamespace) -> Optional[MediaInfo]:
     )
 
 
+@pytest.mark.parametrize(
+    ("torrent_year", "candidate_year"),
+    [("2026", "2024"), ("2026", "2026"), (None, "2026"), (None, None), (None, "2024")],
+)
+def test_live_action_subscription_rejects_same_title_animation(
+    conflict_case: SimpleNamespace,
+    torrent_year: Optional[str],
+    candidate_year: Optional[str],
+) -> None:
+    """
+    《剑来》同名真人与动漫使用不同身份时，标题和年份均不能使动漫进入下载。
+
+    ID 和年份为离线样例，覆盖资源年份相同、共同首播年份及缺失年份的放行路径。
+    """
+    case = conflict_case
+    case.target.type = MediaType.TV
+    case.target.title = case.target.original_title = "剑来"
+    case.target.season_years = {1: "2026"}
+    case.target.genre_ids = [18]
+    case.candidate.type = MediaType.TV
+    case.candidate.title = case.candidate.original_title = "剑来"
+    case.candidate.year = candidate_year
+    case.candidate.genre_ids = [16]
+    case.torrent.title = f"剑来 {torrent_year or ''} S01E01 1080p WEB-DL"
+    case.torrent.description = "剑来 | 动画版 | 国语中字"
+    case.torrent.category = MediaType.TV.value
+    case.meta = MetaInfo(title=case.torrent.title, subtitle=case.torrent.description)
+    case.context.meta_info = case.meta
+    case.subscribe = replace(case.subscribe, name="剑来", type=MediaType.TV.value, season=1)
+
+    assert _execute_match(case) == "completed"
+
+    case.download.assert_not_called()
+    assert case.context.media_info is case.candidate
+    assert case.context.media_info.media_id != case.target.media_id
+    assert case.context.candidate_recognized is True
+    assert case.context.media_info_is_target is False
+
+
+@pytest.mark.parametrize("recognized", [True, False])
+def test_matching_identity_and_unrecognized_title_still_reach_download(
+    conflict_case: SimpleNamespace, recognized: bool
+) -> None:
+    """
+    身份一致的正常资源及尚未识别出身份的标题兜底继续下载，并保留各自匹配依据。
+    """
+    case = conflict_case
+    candidate = replace(case.target) if recognized else None
+    case.context.media_info = candidate
+    case.context.match_source = MediaSource.TMDB.value if recognized else "unknown"
+    case.context.candidate_recognized = recognized
+
+    assert _execute_match(case) == "completed"
+
+    case.download.assert_called_once()
+    matched = case.download.call_args.kwargs["contexts"][0]
+    assert matched.media_info.media_source == case.target.media_source
+    assert matched.media_info.media_id == case.target.media_id
+    assert matched.media_info.type == case.target.type
+    if recognized:
+        assert matched.media_info is candidate
+    assert matched.match_source == (MediaSource.TMDB.value if recognized else "title")
+    assert matched.candidate_recognized is recognized
+    assert matched.media_info_is_target is (not recognized)
+    assert case.context.media_info is candidate
+
+
 def test_inferred_type_conflict_reaches_subscription_review(conflict_case: SimpleNamespace) -> None:
     """
     只由识别结果产生的类型冲突不能在候选路由阶段丢失。
@@ -154,11 +221,11 @@ def test_inferred_type_conflict_reaches_subscription_review(conflict_case: Simpl
 
 
 @pytest.mark.parametrize("candidate_id", ["21494", "1567937"])
-def test_movie_subtitle_reconciles_inferred_tv_without_changing_cache(
+def test_movie_subtitle_and_year_cannot_override_inferred_tv(
     conflict_case: SimpleNamespace, candidate_id: str
 ) -> None:
     """
-    副标题及精确年份可纠正跨类型推断，相同数字 ID 也必须复核且不污染 RSS 缓存。
+    副标题和精确年份不能覆盖类型冲突，相同数字 ID 也不允许进入下载。
     """
     case = conflict_case
     case.candidate.media_id = candidate_id
@@ -166,14 +233,7 @@ def test_movie_subtitle_reconciles_inferred_tv_without_changing_cache(
 
     assert _execute_match(case) == "completed"
 
-    case.download.assert_called_once()
-    matched = case.download.call_args.kwargs["contexts"]
-    assert len(matched) == 1
-    assert matched[0].media_info.media_id == case.target.media_id
-    assert matched[0].media_info.type == MediaType.MOVIE
-    assert matched[0].match_source == "title"
-    assert matched[0].candidate_recognized is False
-    assert matched[0].media_info_is_target is True
+    case.download.assert_not_called()
     assert case.context.media_info is case.candidate
     assert case.candidate.type == MediaType.TV
     assert case.candidate.media_id == candidate_id
@@ -183,11 +243,11 @@ def test_movie_subtitle_reconciles_inferred_tv_without_changing_cache(
 
 
 @pytest.mark.parametrize("candidate_id", ["21494", "1567937"])
-def test_type_conflict_with_same_numeric_id_still_requires_title_review(
+def test_type_conflict_with_same_numeric_id_is_rejected(
     conflict_case: SimpleNamespace, candidate_id: str
 ) -> None:
     """
-    TMDB 的电影与电视剧数字 ID 相同时仍不能绕过标题校验。
+    TMDB 的电影与电视剧数字 ID 相同也不能绕过类型校验。
     """
     case = conflict_case
     case.candidate.media_id = candidate_id
@@ -233,7 +293,7 @@ def test_no_year_tv_title_cannot_override_other_type_without_work_evidence(
     conflict_case: SimpleNamespace, candidate_id: str
 ) -> None:
     """
-    无年份剧集标题虽命中，仍需同作品证据才能覆盖错误类型。
+    无年份剧集标题命中也不能覆盖候选自身的媒体类型。
     """
     case = conflict_case
     case.target.type = MediaType.TV
@@ -252,11 +312,11 @@ def test_no_year_tv_title_cannot_override_other_type_without_work_evidence(
     case.download.assert_not_called()
 
 
-def test_cross_type_rejection_logs_type_and_review_reason(
+def test_cross_type_rejection_logs_conflicting_types(
     conflict_case: SimpleNamespace, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     """
-    冲突诊断须同时记录两种媒体类型和标题复核失败原因。
+    冲突诊断须同时记录候选和目标的媒体类型。
     """
     case = conflict_case
     case.torrent.description = "其他电影"
@@ -264,12 +324,12 @@ def test_cross_type_rejection_logs_type_and_review_reason(
     monkeypatch.setattr(subscribe_match.logger, "debug", messages.append)
 
     assert _reconcile(case) is None
-    assert any("电视剧" in message and "电影" in message and "标题复核失败" in message for message in messages)
+    assert any("电视剧" in message and "电影" in message and "已跳过" in message for message in messages)
 
 
-def test_tv_title_and_year_can_reconcile_inferred_movie(conflict_case: SimpleNamespace) -> None:
+def test_tv_title_and_year_cannot_override_inferred_movie(conflict_case: SimpleNamespace) -> None:
     """
-    电视剧被误识别为电影时，同样可经标题及目标季年份复核后提交。
+    电视剧标题及目标季年份命中，仍不能覆盖候选自身的电影类型。
     """
     case = conflict_case
     case.target.type = MediaType.TV
@@ -283,11 +343,9 @@ def test_tv_title_and_year_can_reconcile_inferred_movie(conflict_case: SimpleNam
 
     assert _execute_match(case) == "completed"
 
-    case.download.assert_called_once()
-    matched = case.download.call_args.kwargs["contexts"][0]
-    assert matched.media_info.type == MediaType.TV
-    assert matched.media_info.media_id == case.target.media_id
-    assert matched.match_source == "title"
+    case.download.assert_not_called()
+    assert case.context.media_info is case.candidate
+    assert case.context.media_info.type == MediaType.MOVIE
 
 
 def test_global_identity_remains_valid_cross_type_evidence(conflict_case: SimpleNamespace) -> None: