diff --git a/tests/unit/test_dedup_engine.py b/tests/unit/test_dedup_engine.py index a716c7ff9..aa099a7a5 100644 --- a/tests/unit/test_dedup_engine.py +++ b/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) diff --git a/tests/unit/test_duplicate_rate.py b/tests/unit/test_duplicate_rate.py index f92a67a85..87888510c 100644 --- a/tests/unit/test_duplicate_rate.py +++ b/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 时应跨项目查询，而非仅当前项目."""