fix: 修复旧测试文件兼容 #1658 融合算法

- test_dedup_engine.py: 5 个失败测试修复
  - 补充归一化颜色直方图数据使融合相似度能通过阈值
  - 更新 reason 字符串为 phash_histogram_fusion / batch_phash_histogram_fusion
  - 更新相似度期望值（融合公式 0.7*phash + 0.3*hist）
  - 更新 _make_existing_video 辅助函数支持 histograms 参数

- test_duplicate_rate.py: 2 个失败测试修复
  - 补充归一化直方图数据并更新期望值（97.81, 98.91）
  - 更新 _make_fingerprint 辅助函数支持 histograms 参数

全量 81 passed, 8 skipped（cv2 依赖测试跳过）

tests/unit/test_dedup_engine.py

@@ -81,6 +81,11 @@ from apps.worker.video_processing.dedup import (  # noqa: E402
     hamming_distance,
 )
 
+# Issue #1658: 归一化颜色直方图（96 维 = 3 通道 × 32 bins，sum=1.0）。
+# 相同的归一化直方图之间 Bhattacharyya 系数 = Σ√(a*b) = Σa = 1.0，
+# 代表"颜色完全一致"，用于测试融合逻辑中的直方图贡献。
+_NORM_HISTOGRAM = [1.0 / 96] * 96
+
 # ---------------------------------------------------------------------------
 # dedup 模块已导入完成，立即恢复 worker_app 真实包，避免污染后续测试文件
 # ---------------------------------------------------------------------------
@@ -208,8 +213,9 @@ class TestComputeColorHistogram:
 class TestVideoDeduplicatorCheckDuplicate:
     """VideoDeduplicator.check_duplicate() 测试。
 
-    当前实现仅使用 MD5 精确匹配和 pHash 距离判定，
-    不包含颜色直方图相似度计算。
+    Issue #1658 后判定逻辑：MD5 精确匹配，或 pHash + 颜色直方图融合
+    （0.7*phash + 0.3*hist），且需同时满足帧匹配比例 ≥ 0.7 与
+    融合相似度 ≥ 0.70，reason 为 "phash_histogram_fusion"。
     """
 
     @pytest.fixture
@@ -220,14 +226,18 @@ class TestVideoDeduplicatorCheckDuplicate:
     def mock_session(self):
         return MagicMock()
 
-    def _make_existing_video(self, video_id, md5, phashes=None):
-        """创建模拟已有视频的 mock 对象。"""
+    def _make_existing_video(self, video_id, md5, phashes=None, histograms=None):
+        """创建模拟已有视频的 mock 对象。
+
+        histograms 默认为 None（无直方图，融合时 hist_similarity=0.0）；
+        传入 [] 同样表示无直方图。
+        """
         video = MagicMock()
         video.id = video_id
         video.video_fingerprint = {
             "md5": md5,
             "keyframe_phashes": phashes or [],
-            "color_histograms": [],
+            "color_histograms": histograms if histograms is not None else [],
         }
         return video
 
@@ -267,15 +277,20 @@ class TestVideoDeduplicatorCheckDuplicate:
             self._restore_repo(mod, orig)
 
     def test_phash_similar_match(self, deduplicator, mock_session):
-        """pHash 距离 < 阈值时应判定为重复。"""
-        existing = self._make_existing_video("vid-1", "different_md5", phashes=["abcdef01"])
+        """pHash 距离 < 阈值且融合相似度达标时应判定为重复（Issue #1658 融合逻辑）。"""
+        existing = self._make_existing_video(
+            "vid-1",
+            "different_md5",
+            phashes=["abcdef01"],
+            histograms=[_NORM_HISTOGRAM],
+        )
         mock_repo = MagicMock()
         mock_repo.list_by_project.return_value = [existing]
 
         fingerprint = VideoFingerprint(
             md5="different_md5_new",
-            keyframe_phashes=["abcdef01"],  # 完全相同，距离=0
-            color_histograms=[],
+            keyframe_phashes=["abcdef01"],  # 完全相同，中位距离=0
+            color_histograms=[_NORM_HISTOGRAM],  # 颜色也完全一致 → hist_sim=1.0
             duration=10.0,
             resolution=(1280, 720),
         )
@@ -285,14 +300,17 @@ class TestVideoDeduplicatorCheckDuplicate:
             result = deduplicator.check_duplicate(fingerprint, "proj-1", mock_session)
             assert result is not None
             assert result["duplicate"] is True
-            assert result["similarity"] == 1.0  # distance=0 → 1.0
-            assert result["reason"] == "phash_similar"
+            # 中位距离=0 → phash_sim=1.0；hist_sim≈1.0（相同归一化直方图的
+            # Bhattacharyya 系数受浮点累加影响为 0.9999…）→ 融合相似度≈1.0
+            assert result["similarity"] == pytest.approx(1.0, abs=1e-9)
+            assert result["reason"] == "phash_histogram_fusion"
         finally:
             self._restore_repo(mod, orig)
 
     def test_no_match_returns_none(self, deduplicator, mock_session):
-        """pHash 平均距离 >= PHASH_THRESHOLD(10) 时应返回 None。"""
-        # 使用 16 字符 phash（64 bit），全部不同 → 距离=64 >= 10
+        """pHash 距离过大、融合相似度不达标时应返回 None（Issue #1658）。"""
+        # 使用 16 字符 phash（64 bit），全部不同 → 距离=64，
+        # match_ratio=0 < 0.7 且融合相似度仅为直方图贡献 → 不判重复
         existing = self._make_existing_video("vid-1", "md5_a", phashes=["0000000000000000"])
         mock_repo = MagicMock()
         mock_repo.list_by_project.return_value = [existing]
@@ -358,10 +376,22 @@ class TestVideoDeduplicatorCheckDuplicate:
 
     def test_first_match_returned(self, deduplicator, mock_session):
         """返回第一个通过阈值的匹配（非最优匹配）。"""
-        # vid-1: 距离=2 bits（0x03 XOR 0x01 = 0x02 → 1 bit），通过阈值
-        vid1 = self._make_existing_video("vid-1", "md5_1", phashes=["0000000000000003"])
+        # Issue #1658: 补全颜色直方图，使 vid-1（1 bit 差异）融合相似度
+        # = 0.7*(1-1/64) + 0.3*1.0 ≈ 0.989 ≥ 0.70，能通过融合阈值。
+        # vid-1: 距离=1 bit（0x03 XOR 0x01 = 0x02 → 1 bit），通过阈值
+        vid1 = self._make_existing_video(
+            "vid-1",
+            "md5_1",
+            phashes=["0000000000000003"],
+            histograms=[_NORM_HISTOGRAM],
+        )
         # vid-2: 距离=0 bits（完全匹配）
-        vid2 = self._make_existing_video("vid-2", "md5_2", phashes=["0000000000000001"])
+        vid2 = self._make_existing_video(
+            "vid-2",
+            "md5_2",
+            phashes=["0000000000000001"],
+            histograms=[_NORM_HISTOGRAM],
+        )
 
         mock_repo = MagicMock()
         mock_repo.list_by_project.return_value = [vid1, vid2]
@@ -369,7 +399,7 @@ class TestVideoDeduplicatorCheckDuplicate:
         fingerprint = VideoFingerprint(
             md5="md5_new",
             keyframe_phashes=["0000000000000001"],
-            color_histograms=[],
+            color_histograms=[_NORM_HISTOGRAM],
             duration=10.0,
             resolution=(1280, 720),
         )
@@ -378,7 +408,8 @@ class TestVideoDeduplicatorCheckDuplicate:
         try:
             result = deduplicator.check_duplicate(fingerprint, "proj-1", mock_session)
             assert result is not None
-            # 返回第一个通过阈值的匹配（vid-1 距离=1 < 10）
+            # 返回第一个通过融合阈值的匹配（vid-1 距离=1 < PHASH_THRESHOLD=8，
+            # 融合相似度≈0.989 ≥ 0.70），而非更优的 vid-2
             assert result["duplicate_of"] == "vid-1"
         finally:
             self._restore_repo(mod, orig)
@@ -405,17 +436,22 @@ class TestVideoDeduplicatorCheckDuplicate:
             self._restore_repo(mod, orig)
 
     def test_phash_similarity_formula(self, deduplicator, mock_session):
-        """验证相似度公式：similarity = 1.0 - (avg_distance / 64)。"""
+        """验证 Issue #1658 融合相似度公式：0.7*phash_sim + 0.3*hist_sim。"""
         # 使用已知距离的 phash 对
         # "0000000000000000" vs "0000000000000001" → XOR = 1 → 1 bit → distance = 1
-        existing = self._make_existing_video("vid-1", "md5_a", phashes=["0000000000000000"])
+        existing = self._make_existing_video(
+            "vid-1",
+            "md5_a",
+            phashes=["0000000000000000"],
+            histograms=[_NORM_HISTOGRAM],
+        )
         mock_repo = MagicMock()
         mock_repo.list_by_project.return_value = [existing]
 
         fingerprint = VideoFingerprint(
             md5="md5_b",
             keyframe_phashes=["0000000000000001"],  # 1 bit different
-            color_histograms=[],
+            color_histograms=[_NORM_HISTOGRAM],  # 直方图完全一致 → hist_sim=1.0
             duration=10.0,
             resolution=(1280, 720),
         )
@@ -425,28 +461,32 @@ class TestVideoDeduplicatorCheckDuplicate:
             result = deduplicator.check_duplicate(fingerprint, "proj-1", mock_session)
             assert result is not None
             assert result["duplicate"] is True
-            # similarity = 1.0 - (1 / 64) = 0.984375
-            assert abs(result["similarity"] - (1.0 - 1.0 / 64)) < 1e-6
+            # 中位距离=1 → phash_sim = 1.0 - 1/64；hist_sim = 1.0
+            # 融合相似度 = 0.7*(1 - 1/64) + 0.3*1.0 = 0.9890625
+            expected = 0.7 * (1.0 - 1.0 / 64) + 0.3 * 1.0
+            assert abs(result["similarity"] - expected) < 1e-6
         finally:
             self._restore_repo(mod, orig)
 
     def test_multiple_phashes_avg_distance(self, deduplicator, mock_session):
-        """多帧 phash 使用平均最小距离。"""
+        """多帧 phash 取每帧最小距离，Issue #1658 使用中位距离参与融合计算。"""
         # 已有视频有 2 帧 phash
         existing = self._make_existing_video(
             "vid-1",
             "md5_a",
             phashes=["0000000000000000", "ffffffffffffffff"],
+            histograms=[_NORM_HISTOGRAM, _NORM_HISTOGRAM],
         )
         mock_repo = MagicMock()
         mock_repo.list_by_project.return_value = [existing]
 
         # 新视频有 1 帧 phash，与第一帧距离=0，与第二帧距离=64
-        # min_distance = 0, avg = 0 → 匹配
+        # min_distance = 0，中位距离 = 0 → phash_sim = 1.0；
+        # 直方图完全一致 → hist_sim = 1.0 → 融合相似度 = 1.0
         fingerprint = VideoFingerprint(
             md5="md5_b",
             keyframe_phashes=["0000000000000000"],
-            color_histograms=[],
+            color_histograms=[_NORM_HISTOGRAM],
             duration=10.0,
             resolution=(1280, 720),
         )
@@ -456,7 +496,8 @@ class TestVideoDeduplicatorCheckDuplicate:
             result = deduplicator.check_duplicate(fingerprint, "proj-1", mock_session)
             assert result is not None
             assert result["duplicate"] is True
-            assert result["similarity"] == 1.0  # avg_distance = 0
+            # 中位距离=0 且 hist_sim≈1.0 → 融合相似度≈1.0（浮点累加误差内）
+            assert result["similarity"] == pytest.approx(1.0, abs=1e-9)
         finally:
             self._restore_repo(mod, orig)
 
@@ -464,7 +505,8 @@ class TestVideoDeduplicatorCheckDuplicate:
 class TestVideoDeduplicatorCheckBatchDuplicate:
     """VideoDeduplicator.check_batch_duplicate() 测试。
 
-    批次内查重逻辑与历史查重一致（MD5 + pHash），但搜索范围限定为同 batch_id 的视频。
+    批次内查重逻辑与历史查重一致（MD5 + pHash/颜色直方图融合，Issue #1658），
+    但搜索范围限定为同 batch_id 的视频，融合命中 reason 带 "batch_" 前缀。
     """
 
     @pytest.fixture
@@ -475,13 +517,13 @@ class TestVideoDeduplicatorCheckBatchDuplicate:
     def mock_session(self):
         return MagicMock()
 
-    def _make_batch_video(self, video_id, md5, phashes=None):
+    def _make_batch_video(self, video_id, md5, phashes=None, histograms=None):
         video = MagicMock()
         video.id = video_id
         video.video_fingerprint = {
             "md5": md5,
             "keyframe_phashes": phashes or [],
-            "color_histograms": [],
+            "color_histograms": histograms if histograms is not None else [],
         }
         return video
 
@@ -521,15 +563,20 @@ class TestVideoDeduplicatorCheckBatchDuplicate:
             self._restore_repo(mod, orig)
 
     def test_batch_phash_similar(self, deduplicator, mock_session):
-        """批次内 pHash 距离 < 阈值应判定为重复。"""
-        other = self._make_batch_video("vid-other", "md5_diff", phashes=["abcdef01"])
+        """批次内 pHash + 颜色直方图融合命中应判定为重复（Issue #1658）。"""
+        other = self._make_batch_video(
+            "vid-other",
+            "md5_diff",
+            phashes=["abcdef01"],
+            histograms=[_NORM_HISTOGRAM],
+        )
         mock_repo = MagicMock()
         mock_repo.list_by_batch.return_value = [other]
 
         fingerprint = VideoFingerprint(
             md5="md5_new",
-            keyframe_phashes=["abcdef01"],
-            color_histograms=[],
+            keyframe_phashes=["abcdef01"],  # 完全相同，中位距离=0
+            color_histograms=[_NORM_HISTOGRAM],  # 颜色一致 → hist_sim=1.0
             duration=10.0,
             resolution=(1280, 720),
         )
@@ -539,7 +586,10 @@ class TestVideoDeduplicatorCheckBatchDuplicate:
             result = deduplicator.check_batch_duplicate(fingerprint, "batch-1", "vid-self", mock_session)
             assert result is not None
             assert result["duplicate"] is True
-            assert result["reason"] == "batch_phash_similar"
+            # 融合命中：reason 带 batch_ 前缀
+            assert result["reason"] == "batch_phash_histogram_fusion"
+            # 中位距离=0、hist_sim≈1.0 → 融合相似度≈1.0（浮点累加误差内）
+            assert result["similarity"] == pytest.approx(1.0, abs=1e-9)
         finally:
             self._restore_repo(mod, orig)
 

tests/unit/test_duplicate_rate.py

@@ -15,17 +15,25 @@ sys.path.insert(0, str(ROOT / "apps" / "api"))
 sys.path.insert(0, str(ROOT / "packages"))
 sys.path.insert(0, str(ROOT / "apps" / "worker"))
 
+# Issue #1658: 归一化颜色直方图（96 维 = 3 通道 × 32 bins，sum=1.0）。
+# 相同归一化直方图的 Bhattacharyya 系数 ≈ 1.0（颜色完全一致），
+# 用于验证 pHash + 颜色直方图融合后的 duplicate_rate。
+_NORM_HISTOGRAM = [1.0 / 96] * 96
+
 
 class TestComputeDuplicateRate:
     """Test VideoDeduplicator.compute_duplicate_rate."""
 
-    def _make_fingerprint(self, md5="abc123", phashes=None):
+    def _make_fingerprint(self, md5="abc123", phashes=None, histograms=None):
         from video_processing.dedup import VideoFingerprint
 
+        phashes = phashes or ["ff00ff00ff00ff00"]
+        if histograms is None:
+            histograms = [_NORM_HISTOGRAM] * len(phashes)
         return VideoFingerprint(
             md5=md5,
-            keyframe_phashes=phashes or ["ff00ff00ff00ff00"],
-            color_histograms=[],
+            keyframe_phashes=phashes,
+            color_histograms=histograms,
             duration=10.0,
             resolution=(1920, 1080),
         )
@@ -104,7 +112,12 @@ class TestComputeDuplicateRate:
 
         existing = self._make_existing_video(
             "existing1",
-            {"md5": "other_md5", "keyframe_phashes": ["ff00ff00ff00ff03"]},
+            {
+                "md5": "other_md5",
+                "keyframe_phashes": ["ff00ff00ff00ff03"],
+                # Issue #1658: 提供归一化直方图（颜色一致 → hist_sim≈1.0）
+                "color_histograms": [_NORM_HISTOGRAM],
+            },
         )
         mock_model = MagicMock(spec=GeneratedVideoModel)
         mock_model.id = existing.id
@@ -121,8 +134,9 @@ class TestComputeDuplicateRate:
             session.query.return_value = query_mock
             rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
 
-        # hamming distance = 2, similarity = (1 - 2/64) * 100 = 96.875
-        assert rate == pytest.approx(96.88, abs=0.1)
+        # Issue #1658 融合公式：hamming distance = 2 → phash_sim = 1 - 2/64；
+        # hist_sim ≈ 1.0 → fusion = 0.7*(1-2/64) + 0.3*1.0 = 0.978125 → 97.81
+        assert rate == pytest.approx(97.81, abs=0.1)
 
     def test_excludes_self_video(self):
         from video_processing.dedup import VideoDeduplicator
@@ -160,8 +174,23 @@ class TestComputeDuplicateRate:
         fingerprint = self._make_fingerprint(md5="new_md5", phashes=["ff00ff00ff00ff00"])
         session = MagicMock()
 
-        existing1 = self._make_existing_video("e1", {"md5": "md5_1", "keyframe_phashes": ["ff00ff00ff00ff0f"]})
-        existing2 = self._make_existing_video("e2", {"md5": "md5_2", "keyframe_phashes": ["ff00ff00ff00ff01"]})
+        existing1 = self._make_existing_video(
+            "e1",
+            {
+                "md5": "md5_1",
+                "keyframe_phashes": ["ff00ff00ff00ff0f"],
+                # Issue #1658: 颜色直方图一致 → hist_sim≈1.0
+                "color_histograms": [_NORM_HISTOGRAM],
+            },
+        )
+        existing2 = self._make_existing_video(
+            "e2",
+            {
+                "md5": "md5_2",
+                "keyframe_phashes": ["ff00ff00ff00ff01"],
+                "color_histograms": [_NORM_HISTOGRAM],
+            },
+        )
         mock_model1 = MagicMock(spec=GeneratedVideoModel)
         mock_model1.id = existing1.id
         mock_model1.project_id = existing1.project_id
@@ -185,8 +214,9 @@ class TestComputeDuplicateRate:
             session.query.return_value = query_mock
             rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
 
-        # max similarity: e2 distance=1, (1-1/64)*100 = 98.4375
-        assert rate == pytest.approx(98.44, abs=0.1)
+        # Issue #1658 融合公式：e1 distance=4 → 0.7*(1-4/64)+0.3 ≈ 0.95625 → 95.62；
+        # e2 distance=1 → 0.7*(1-1/64)+0.3 = 0.9890625 → 98.91。取最大值 e2。
+        assert rate == pytest.approx(98.91, abs=0.1)
 
     def test_user_id_scope_cross_project(self):
         """传 user_id 时应跨项目查询，而非仅当前项目."""