fix(dedup): check_duplicate支持exclude_video_id，修复recompute自匹配duplicate_of指向自己 (Issue #1702连带修复)

recompute-dedup重算时视频记录已在库中(正常生成流程新视频未落库所以没暴露),
check_duplicate候选列表包含自身,自匹配汉明距离0融合分最高,导致
is_duplicate=true但duplicate_of指向视频自己。
check_batch_duplicate/compute_duplicate_rate本就有排除自身逻辑,
给check_duplicate补exclude_video_id参数,recompute任务与helpers均传入。
补回归单测。

apps/worker/video_processing/dedup.py

@@ -782,6 +782,7 @@ class VideoDeduplicator:
         scope: str = "project",
         user_id: str = "",
         duration_sec: float = 0,
+        exclude_video_id: str | None = None,
     ) -> Optional[dict]:
         """检查视频是否与已有视频重复。
 
@@ -798,6 +799,9 @@ class VideoDeduplicator:
             scope: "project" 项目内查重（默认），"user" 跨项目全局查重
             user_id: 用户 ID（scope="user" 时使用）
             duration_sec: 视频时长（秒），用于时长预过滤 ±15%
+            exclude_video_id: 排除的视频 ID（查重自身时用）。recompute-dedup
+                重算时视频记录已存在，不排除会自匹配（距离 0 分最高）导致
+                duplicate_of 指向自己（Issue #1702 连带修复）。
 
         Returns:
             重复信息字典（含 duplicate, duplicate_of, reason, similarity, duplicate_segments），
@@ -815,6 +819,9 @@ class VideoDeduplicator:
         best_result: Optional[dict] = None
 
         for existing in existing_videos:
+            # 排除自身（recompute 时当前视频已在候选列表里，否则自匹配距离 0 必最高分）
+            if exclude_video_id and existing.id == exclude_video_id:
+                continue
             if not existing.video_fingerprint:
                 continue
 
@@ -1240,6 +1247,7 @@ def check_duplicate_task(self: Task, generated_video_id: str) -> dict:
             # Issue #1702: fingerprint.duration 单位已经是秒，旧代码 /1000 导致
             # ±15% 时长预过滤窗口缩到 ~0.013s，scope=user 的跨项目查重永远返回 None。
             duration_sec=fingerprint.duration if fingerprint.duration else 0,
+            exclude_video_id=generated_video_id,
         )
 
         video.video_fingerprint = fingerprint.to_dict()

apps/worker/video_processing/dedup_helpers.py

@@ -101,6 +101,7 @@ def create_video_record_and_dedup(
                 scope="user",
                 user_id=user_id,
                 duration_sec=duration_sec,
+                exclude_video_id=video_id,
             )
 
             # (b) 批次内查重（仅当有 batch_id 时）

tests/unit/test_dedup_1702_zero_rate_fix.py

@@ -387,3 +387,42 @@ class TestRecomputeDownloadPath:
             'download_key = f"projects/{video.project_id}/generated/{generated_video_id}/{generated_video_id}.mp4"',
             "",
         )
+
+
+# ── check_duplicate 排除自身（#1702 连带修复：recompute 自匹配） ─────
+
+
+class TestCheckDuplicateExcludesSelf:
+    def test_exclude_video_id_skips_self_match(self):
+        """recompute 时当前视频已在候选列表：自匹配距离 0 分会让 duplicate_of
+        指向自己。exclude_video_id 必须跳过自身，返回真实的其他匹配或 None。
+        """
+
+        ddp = VideoDeduplicator()
+        h = _h(0)
+        # 候选列表里同时放「自己」（完全相同）和一个异源视频
+        self_video = _video("v-self", [h], duration=10.0)
+        other_video = _video("v-other", [_h(40 + i) for i in range(3)], duration=10.0)
+        fp = _fingerprint([h], 10.0)
+
+        session = MagicMock()
+        session.query.return_value.filter.return_value.order_by.return_value.all.return_value = []
+
+        # 不传 exclude → 自匹配命中（错误行为复现）
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            MockRepo.return_value.list_by_project.return_value = [self_video, other_video]
+            result = ddp.check_duplicate(fp, "proj1", session)
+        assert result is not None and result["duplicate_of"] == "v-self"
+
+        # 传 exclude_video_id → 跳过自己，异源不匹配 → None
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            MockRepo.return_value.list_by_project.return_value = [self_video, other_video]
+            result = ddp.check_duplicate(fp, "proj1", session, exclude_video_id="v-self")
+        assert result is None
+
+        # 排除自己后，真实同源其他视频仍能检出
+        real_dup = _video("v-real", [h], duration=10.0)
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            MockRepo.return_value.list_by_project.return_value = [self_video, real_dup]
+            result = ddp.check_duplicate(fp, "proj1", session, exclude_video_id="v-self")
+        assert result is not None and result["duplicate_of"] == "v-real"