feat: pHash阈值校准+颜色直方图融合 #1658

Issue #1658: 提高查重检测精度

改动摘要:
- PHASH_THRESHOLD 从 10 收紧到 8
- 均值 → 中位数抵抗黑帧/转场干扰 (statistics.median)
- 新增帧匹配比例条件 MATCH_RATIO_THRESHOLD=0.7
- 新增 Bhattacharyya 系数融合颜色直方图 (0.7 pHash + 0.3 hist)
- 删除旧 _average_histogram_similarity()
- 提取 _check_fusion_duplicate() 共用方法
- 改造 compute_duplicate_rate() 使用融合逻辑
- reason 标记: phash_histogram_fusion / batch_phash_histogram_fusion
- 向后兼容: 无直方图数据时 hist_similarity 回退到 0.5

新增常量:
- PHASH_THRESHOLD = 8
- MATCH_RATIO_THRESHOLD = 0.7
- DUPLICATE_THRESHOLD = 0.70
- PHASH_WEIGHT = 0.7
- HISTOGRAM_WEIGHT = 0.3

测试:
- test_dedup_v2.py: 34 个新测试全绿
- test_dedup_pure.py: 删除10个引用已删除方法的旧测试, 剩余10个全绿
- 总计 44 个测试全部通过

apps/worker/video_processing/dedup.py

@@ -1,8 +1,17 @@
-"""Video deduplication module - compute fingerprints and detect duplicates."""
+"""Video deduplication module - compute fingerprints and detect duplicates.
+
+Issue #1658: pHash 阈值校准 + 颜色直方图融合
+- PHASH_THRESHOLD 从 10 收紧到 8
+- 均值 → 中位数抵抗黑帧/转场干扰
+- 新增帧匹配比例条件 (MATCH_RATIO_THRESHOLD=0.7)
+- Bhattacharyya 系数融合颜色直方图 (PHASH_WEIGHT=0.7, HISTOGRAM_WEIGHT=0.3)
+- 删除旧 _average_histogram_similarity()
+"""
 
 import hashlib
 import logging
 import os
+import statistics
 import tempfile
 from dataclasses import dataclass, field
 from typing import Optional
@@ -164,10 +173,17 @@ class VideoFingerprint:
 
 
 class VideoDeduplicator:
-    """Video deduplication using multiple fingerprint methods."""
+    """Video deduplication using multiple fingerprint methods.
 
-    PHASH_THRESHOLD = 10
-    HISTOGRAM_THRESHOLD = 0.85
+    Issue #1658: pHash 阈值校准 + 颜色直方图融合
+    """
+
+    # ── Issue #1658: 校准后的常量 ──
+    PHASH_THRESHOLD = 8                # 从 10 收紧到 8
+    MATCH_RATIO_THRESHOLD = 0.7        # 至少 70% 帧匹配
+    DUPLICATE_THRESHOLD = 0.70         # 融合后相似度阈值
+    PHASH_WEIGHT = 0.7                 # pHash 权重
+    HISTOGRAM_WEIGHT = 0.3             # 直方图权重
 
     def compute_fingerprint(self, video_path: str) -> VideoFingerprint:
         """Compute video fingerprint using MD5, pHash, and color histogram.
@@ -255,14 +271,147 @@ class VideoDeduplicator:
             for r in rows
         ]
 
+    # ── Issue #1658: 新增 Bhattacharyya 系数方法 ──
+
+    @staticmethod
+    def _bhattacharyya_coefficient(hist_a: list[float], hist_b: list[float]) -> float:
+        """Bhattacharyya 系数：Σ √(a[i] * b[i])，范围 [0, 1]，1=完全相同。
+
+        Args:
+            hist_a: 第一组直方图数据
+            hist_b: 第二组直方图数据
+
+        Returns:
+            Bhattacharyya 系数，范围 [0, 1]
+        """
+        min_len = min(len(hist_a), len(hist_b))
+        a = hist_a[:min_len]
+        b = hist_b[:min_len]
+        return float(sum(np.sqrt(ai * bi) for ai, bi in zip(a, b)))
+
+    @staticmethod
+    def _compute_histogram_similarity(
+        histograms_a: list[list[float]],
+        histograms_b: list[list[float]],
+    ) -> float:
+        """对每组直方图，找到最佳匹配的 Bhattacharyya 系数，取平均。
+
+        Args:
+            histograms_a: 第一组直方图（每帧一个 list）
+            histograms_b: 第二组直方图
+
+        Returns:
+            平均最佳匹配 Bhattacharyya 系数，范围 [0, 1]
+        """
+        if not histograms_a or not histograms_b:
+            return 0.0
+        similarities = []
+        for ha in histograms_a:
+            best = 0.0
+            for hb in histograms_b:
+                bc = VideoDeduplicator._bhattacharyya_coefficient(ha, hb)
+                best = max(best, bc)
+            similarities.append(best)
+        return sum(similarities) / len(similarities) if similarities else 0.0
+
+    # ── Issue #1658: 内部辅助方法 ──
+
+    def _compute_min_distances(
+        self,
+        new_phashes: list[str],
+        existing_phashes: list[str],
+    ) -> list[int]:
+        """计算每个新关键帧到已有关键帧的最小汉明距离。
+
+        Args:
+            new_phashes: 新视频的 pHash 列表
+            existing_phashes: 已有视频的 pHash 列表
+
+        Returns:
+            每帧的最小距离列表
+        """
+        min_distances = []
+        for phash in new_phashes:
+            distances = [hamming_distance(phash, ep) for ep in existing_phashes]
+            min_distances.append(min(distances))
+        return min_distances
+
+    def _check_fusion_duplicate(
+        self,
+        fingerprint: VideoFingerprint,
+        existing_fingerprint: dict,
+        existing_phashes: list[str],
+        existing_histograms: list[list[float]],
+    ) -> Optional[dict]:
+        """Issue #1658: pHash + 直方图融合判定逻辑（check_duplicate / check_batch_duplicate / compute_duplicate_rate 共用）。
+
+        判定流程：
+        1. 计算每帧最小汉明距离
+        2. 检查帧匹配比例（≥70%）
+        3. 计算中位距离
+        4. 融合 pHash 相似度 + Bhattacharyya 直方图相似度
+        5. 综合得分 ≥ DUPLICATE_THRESHOLD 则判重复
+
+        Args:
+            fingerprint: 新视频指纹
+            existing_fingerprint: 已有视频的 fingerprint dict
+            existing_phashes: 已有视频的 pHash 列表（已解析）
+            existing_histograms: 已有视频的直方图列表（已解析）
+
+        Returns:
+            融合判定结果 dict（含 similarity, reason, _debug），或 None
+        """
+        if not existing_phashes:
+            return None
+
+        # Step 1: 计算每帧最小汉明距离
+        min_distances = self._compute_min_distances(fingerprint.keyframe_phashes, existing_phashes)
+
+        # Step 2: 帧匹配比例检查
+        matching_frames = sum(1 for d in min_distances if d < self.PHASH_THRESHOLD)
+        match_ratio = matching_frames / len(min_distances) if min_distances else 0
+        if match_ratio < self.MATCH_RATIO_THRESHOLD:
+            return None
+
+        # Step 3: 中位距离（替代均值，抵抗黑帧/转场异常值）
+        median_distance = statistics.median(min_distances)
+
+        # Step 4: 加权融合
+        phash_similarity = 1.0 - (median_distance / 64)
+        hist_similarity = (
+            self._compute_histogram_similarity(fingerprint.color_histograms, existing_histograms)
+            if existing_histograms
+            else 0.5  # 无直方图数据时给中间值（向后兼容）
+        )
+        combined_score = self.PHASH_WEIGHT * phash_similarity + self.HISTOGRAM_WEIGHT * hist_similarity
+
+        # Step 5: 判定
+        if combined_score < self.DUPLICATE_THRESHOLD:
+            return None
+
+        return {
+            "reason": "phash_histogram_fusion",
+            "similarity": combined_score,
+            "_debug": {
+                "median_distance": median_distance,
+                "match_ratio": match_ratio,
+                "phash_similarity": phash_similarity,
+                "hist_similarity": hist_similarity,
+                "combined_score": combined_score,
+            },
+        }
+
     def check_duplicate(self, fingerprint: VideoFingerprint, project_id: str, session: Session) -> Optional[dict]:
         """检查视频是否与项目中已有视频重复。
 
-        查重逻辑：
-        1. MD5 精确匹配 → similarity=1.0
-        2. pHash 相似度（优先从分片表读取，回退到 JSON 字段）
-
-        判定阈值：avg_distance < PHASH_THRESHOLD(10)
+        Issue #1658 改造后判定逻辑（按优先级）：
+        1. MD5 精确匹配：完全一致则 similarity=1.0，立即返回
+        2. pHash + 直方图融合：
+           a. 计算每帧最小汉明距离
+           b. 帧匹配比例 ≥ 70% 才继续
+           c. 中位距离替代均值（抵抗黑帧/转场干扰）
+           d. 加权融合 pHash 相似度 + Bhattacharyya 直方图相似度
+           e. combined_score ≥ 0.70 则判重复
 
         Args:
             fingerprint: 待检测视频的指纹
@@ -286,36 +435,28 @@ class VideoDeduplicator:
             if fingerprint.md5 == ef.get("md5"):
                 return {"duplicate": True, "duplicate_of": existing.id, "reason": "exact_md5_match", "similarity": 1.0}
 
-            # 优先从分片表读取已有视频的分片 phash
-            existing_phashes = []
+            # 优先从分片表读取已有视频的分片数据
             chunk_data = self._get_existing_chunks(existing.id, session)
             if chunk_data:
                 existing_phashes = [c["phash_binary"] for c in chunk_data]
+                existing_histograms = [c["color_histogram"] for c in chunk_data]
             else:
                 # 回退：从 JSON 字段读取（存量旧视频）
                 existing_phashes = ef.get("keyframe_phashes", [])
+                existing_histograms = ef.get("color_histograms", [])
 
             if not existing_phashes:
                 continue
 
-            # 计算每个新关键帧到已有关键帧的最小汉明距离，取平均
-            min_distances = []
-            for phash in fingerprint.keyframe_phashes:
-                distances = [hamming_distance(phash, ep) for ep in existing_phashes]
-                min_distances.append(min(distances))
-            avg_distance = sum(min_distances) / len(min_distances) if min_distances else 100
-
-            if avg_distance >= self.PHASH_THRESHOLD:
-                continue
-
-            phash_similarity = 1.0 - (avg_distance / 64)
-
-            return {
-                "duplicate": True,
-                "duplicate_of": existing.id,
-                "reason": "phash_similar",
-                "similarity": phash_similarity,
-            }
+            # Issue #1658: pHash + 直方图融合判定
+            fusion_result = self._check_fusion_duplicate(fingerprint, ef, existing_phashes, existing_histograms)
+            if fusion_result:
+                return {
+                    "duplicate": True,
+                    "duplicate_of": existing.id,
+                    "reason": fusion_result["reason"],
+                    "similarity": fusion_result["similarity"],
+                }
 
         return None
 
@@ -328,7 +469,8 @@ class VideoDeduplicator:
     ) -> Optional[dict]:
         """检查视频是否与同批次内其他视频重复。
 
-        逻辑与 check_duplicate 一致（MD5 + pHash），但搜索范围限定为同 batch_id 的视频。
+        Issue #1658: 与 check_duplicate 使用完全一致的融合逻辑（MD5 + pHash/直方图融合），
+        但搜索范围限定为同 batch_id 的视频。
 
         Args:
             fingerprint: 待检测视频的指纹
@@ -359,73 +501,29 @@ class VideoDeduplicator:
                 }
 
             # 优先从分片表读取
-            existing_phashes = []
             chunk_data = self._get_existing_chunks(existing.id, session)
             if chunk_data:
                 existing_phashes = [c["phash_binary"] for c in chunk_data]
+                existing_histograms = [c["color_histogram"] for c in chunk_data]
             else:
                 existing_phashes = ef.get("keyframe_phashes", [])
+                existing_histograms = ef.get("color_histograms", [])
 
             if not existing_phashes:
                 continue
 
-            min_distances = []
-            for phash in fingerprint.keyframe_phashes:
-                distances = [hamming_distance(phash, ep) for ep in existing_phashes]
-                min_distances.append(min(distances))
-            avg_distance = sum(min_distances) / len(min_distances) if min_distances else 100
-
-            if avg_distance >= self.PHASH_THRESHOLD:
-                continue
-
-            phash_similarity = 1.0 - (avg_distance / 64)
-            return {
-                "duplicate": True,
-                "duplicate_of": existing.id,
-                "reason": "batch_phash_similar",
-                "similarity": phash_similarity,
-            }
+            # Issue #1658: pHash + 直方图融合判定
+            fusion_result = self._check_fusion_duplicate(fingerprint, ef, existing_phashes, existing_histograms)
+            if fusion_result:
+                return {
+                    "duplicate": True,
+                    "duplicate_of": existing.id,
+                    "reason": "batch_" + fusion_result["reason"],
+                    "similarity": fusion_result["similarity"],
+                }
 
         return None
 
-    @staticmethod
-    def _average_histogram_similarity(histograms_a: list[list[float]], histograms_b: list[list[float]]) -> float:
-        """
-        计算两组颜色直方图之间的平均余弦相似度。
-
-        对每组直方图对取最小长度对齐，计算余弦相似度后取平均。
-
-        Args:
-            histograms_a: 第一组直方图（每帧一个 list）
-            histograms_b: 第二组直方图
-
-        Returns:
-            平均余弦相似度，范围 [0, 1]
-        """
-        if not histograms_a or not histograms_b:
-            return 0.0
-
-        similarities = []
-        for ha in histograms_a:
-            best = 0.0
-            vec_a = np.array(ha, dtype=np.float64)
-            norm_a = np.linalg.norm(vec_a)
-            if norm_a == 0:
-                continue
-            for hb in histograms_b:
-                vec_b = np.array(hb, dtype=np.float64)
-                # 对齐长度
-                min_len = min(len(vec_a), len(vec_b))
-                va, vb = vec_a[:min_len], vec_b[:min_len]
-                norm_b = np.linalg.norm(vb)
-                if norm_b == 0:
-                    continue
-                sim = float(np.dot(va, vb) / (norm_a * norm_b))
-                best = max(best, sim)
-            similarities.append(best)
-
-        return sum(similarities) / len(similarities) if similarities else 0.0
-
     def compute_duplicate_rate(
         self,
         fingerprint: VideoFingerprint,
@@ -437,12 +535,14 @@ class VideoDeduplicator:
     ) -> float:
         """计算当前视频与用户库内已有视频的最高相似度百分比。
 
+        Issue #1658 改造：使用与 check_duplicate 完全一致的融合逻辑。
+        - 帧匹配比例 ≥ 70% 才计入
+        - 中位距离替代均值
+        - 加权融合 pHash + 直方图
+        - 最终 duplicate_rate = fusion_score * 100
+
         优先按 user_id 全局比较（跨项目），user_id 为空时回退到项目级比较。
-        遍历最近 200 个其他有指纹的视频，对每个计算相似度：
-        - MD5 精确匹配 → 100%
-        - pHash 相似度 → (1.0 - avg_distance / 64) * 100
-        取最高值作为 duplicate_rate（0~100）。
-        如果没有其他视频可比较，返回 0.0。
+        遍历最近 200 个其他有指纹的视频，取最高值作为 duplicate_rate（0~100）。
 
         Args:
             fingerprint: 当前视频的指纹
@@ -469,7 +569,7 @@ class VideoDeduplicator:
             )
             logger.debug("compute_duplicate_rate: project-level fallback project_id=%s", project_id)
 
-        # 排除当前视频自身（记录可能已写入 DB，必须在查询层排除）
+        # 排除当前视频自身
         if current_video_id:
             query = query.filter(GeneratedVideoModel.id != current_video_id)
 
@@ -491,23 +591,23 @@ class VideoDeduplicator:
                 return 100.0
 
             # 优先从分片表读取
-            existing_phashes = []
             chunk_data = self._get_existing_chunks(existing.id, session)
             if chunk_data:
                 existing_phashes = [c["phash_binary"] for c in chunk_data]
+                existing_histograms = [c["color_histogram"] for c in chunk_data]
             else:
                 existing_phashes = ef.get("keyframe_phashes", [])
+                existing_histograms = ef.get("color_histograms", [])
 
             if not existing_phashes or not fingerprint.keyframe_phashes:
                 continue
 
-            min_distances = []
-            for phash in fingerprint.keyframe_phashes:
-                distances = [hamming_distance(phash, ep) for ep in existing_phashes]
-                min_distances.append(min(distances))
-            avg_distance = sum(min_distances) / len(min_distances) if min_distances else 64
-            similarity = (1.0 - avg_distance / 64) * 100
-            max_similarity = max(max_similarity, similarity)
+            # Issue #1658: 使用融合判定逻辑
+            fusion_result = self._check_fusion_duplicate(fingerprint, ef, existing_phashes, existing_histograms)
+            if fusion_result:
+                # fusion_result["similarity"] 是 0~1 的分数，转为 0~100 百分比
+                combined_score_pct = fusion_result["similarity"] * 100
+                max_similarity = max(max_similarity, combined_score_pct)
 
         return round(max(max_similarity, 0.0), 2)
 

tests/unit/test_dedup_pure.py

@@ -1,4 +1,7 @@
-"""去重纯算法测试 — hamming_distance + histogram_similarity + VideoFingerprint."""
+"""去重纯算法测试 — hamming_distance + VideoFingerprint.
+
+Issue #1658: 删除 TestAverageHistogramSimilarity（_average_histogram_similarity 已删除）
+"""
 
 from __future__ import annotations
 
@@ -19,10 +22,6 @@ def _mock_module(**attrs):
 
 
 # ── Module-level setup: mock deps, import dedup, then restore sys.modules ──
-# This pattern ensures:
-# 1. dedup is imported with mocks active (no db/celery/cv2 side effects)
-# 2. sys.modules is restored immediately so other test files are not polluted
-# 3. dedup objects are kept in module namespace for tests to use
 
 _SAVED_MODULES_KEYS = set(sys.modules.keys())
 _SAVED_MODULES_VALUES = {
@@ -40,6 +39,7 @@ _SAVED_MODULES_VALUES = {
         "worker_app.core.config",
         "packages.adapters.sqlalchemy_impl.session",
         "packages.adapters.sqlalchemy_impl.generated_video_repository",
+        "packages.adapters.sqlalchemy_impl.models",
         "packages.shared.config",
         "packages.shared.storage",
     ]
@@ -80,6 +80,10 @@ sys.modules["packages.adapters.sqlalchemy_impl.session"] = _mock_module(
     initialize_database=MagicMock(),
 )
 sys.modules["packages.adapters.sqlalchemy_impl.generated_video_repository"] = _mock_module()
+sys.modules["packages.adapters.sqlalchemy_impl.models"] = _mock_module(
+    VideoFingerprintChunkModel=MagicMock(),
+    GeneratedVideoModel=MagicMock(),
+)
 sys.modules["packages.shared.config"] = _mock_module(get_shared_settings=MagicMock(return_value=MagicMock()))
 sys.modules["packages.shared.storage"] = _mock_module()
 
@@ -91,8 +95,6 @@ from video_processing.dedup import (  # noqa: E402
 )
 
 # ── Restore sys.modules immediately after import ──
-# dedup is now cached in this module's namespace; other test files will get
-# their own fresh imports without our mock pollution
 for _key in list(sys.modules.keys()):
     if _key not in _SAVED_MODULES_KEYS:
         del sys.modules[_key]
@@ -118,22 +120,18 @@ class TestHammingDistance:
 
     def test_single_bit_diff(self):
         """1个bit不同."""
-        # 0x01 = 00000001, 0x00 = 00000000 → 1 bit不同
         assert hamming_distance("01", "00") == 1
 
     def test_four_bits_diff(self):
         """4个bit不同."""
-        # 0x0F = 00001111, 0xF0 = 11110000 → 8 bits都不同
         assert hamming_distance("0f", "f0") == 8
 
     def test_longer_hashes(self):
         """更长的哈希（如64-bit pHash）."""
-        # 两个完全不同的64-bit哈希
         assert hamming_distance("0000000000000000", "ffffffffffffffff") == 64
 
     def test_partial_difference(self):
         """部分bit不同."""
-        # a = 1010, 5 = 0101 → 4 bits不同（每个hex digit）
         assert hamming_distance("aa", "55") == 8
 
     def test_case_insensitive(self):
@@ -143,8 +141,6 @@ class TestHammingDistance:
 
     def test_different_length_hashes(self):
         """不同长度的哈希（短的前补零）."""
-        # "ff" = 0xff = 255, "0ff" = 0x0ff = 255
-        # int("ff", 16) = 255, int("0ff", 16) = 255
         assert hamming_distance("ff", "0ff") == 0
 
 
@@ -179,72 +175,3 @@ class TestVideoFingerprint:
         d = fp.to_dict()
         assert d["keyframe_phashes"] == []
         assert d["color_histograms"] == []
-
-
-class TestAverageHistogramSimilarity:
-    """_average_histogram_similarity 直方图相似度测试."""
-
-    def test_identical_histograms(self):
-        """完全相同的直方图相似度为1.0."""
-        hist = [[0.5, 0.5, 0.0], [0.3, 0.4, 0.3]]
-        sim = VideoDeduplicator._average_histogram_similarity(hist, hist)
-        assert sim == pytest.approx(1.0)
-
-    def test_empty_first_list(self):
-        """第一组为空返回0."""
-        sim = VideoDeduplicator._average_histogram_similarity([], [[0.5, 0.5]])
-        assert sim == 0.0
-
-    def test_empty_second_list(self):
-        """第二组为空返回0."""
-        sim = VideoDeduplicator._average_histogram_similarity([[0.5, 0.5]], [])
-        assert sim == 0.0
-
-    def test_both_empty(self):
-        """两组都为空返回0."""
-        sim = VideoDeduplicator._average_histogram_similarity([], [])
-        assert sim == 0.0
-
-    def test_orthogonal_histograms(self):
-        """正交直方图相似度为0."""
-        # [1, 0] 和 [0, 1] 正交
-        sim = VideoDeduplicator._average_histogram_similarity([[1.0, 0.0]], [[0.0, 1.0]])
-        assert sim == pytest.approx(0.0)
-
-    def test_partial_similarity(self):
-        """部分相似."""
-        # [1, 1] 和 [1, 0] 的余弦相似度 = 1/√2 ≈ 0.707
-        sim = VideoDeduplicator._average_histogram_similarity([[1.0, 1.0]], [[1.0, 0.0]])
-        assert sim == pytest.approx(1.0 / (2**0.5), rel=0.01)
-
-    def test_multiple_frames_best_match(self):
-        """多帧时取最佳匹配."""
-        # 第一帧完全不同，第二帧完全相同 → 平均 best = (0 + 1) / 2 = 0.5
-        sim = VideoDeduplicator._average_histogram_similarity(
-            [[1.0, 0.0], [0.0, 1.0]],
-            [[0.0, 1.0]],  # 只有一帧，和第一帧0相似，和第二帧1相似
-        )
-        # 第一帧最佳匹配=0，第二帧最佳匹配=1，平均=0.5
-        assert sim == pytest.approx(0.5)
-
-    def test_zero_norm_histogram_skipped(self):
-        """零范数直方图被跳过."""
-        sim = VideoDeduplicator._average_histogram_similarity([[0.0, 0.0]], [[1.0, 1.0]])
-        # 第一组的零范数被跳过，similarities为空，返回0
-        assert sim == 0.0
-
-    def test_different_length_histograms(self):
-        """不同长度的直方图取最小长度对齐."""
-        sim = VideoDeduplicator._average_histogram_similarity(
-            [[1.0, 1.0, 0.0, 0.0]],  # 4维
-            [[1.0, 1.0]],  # 2维
-        )
-        # 对齐到前2维，都是[1,1]，相似度1.0
-        assert sim == pytest.approx(1.0)
-
-    def test_similarity_in_zero_one_range(self):
-        """相似度在[0, 1]范围内."""
-        hist_a = [np.random.rand(96).tolist() for _ in range(5)]
-        hist_b = [np.random.rand(96).tolist() for _ in range(5)]
-        sim = VideoDeduplicator._average_histogram_similarity(hist_a, hist_b)
-        assert 0.0 <= sim <= 1.0

tests/unit/test_dedup_v2.py

@@ -0,0 +1,648 @@
+"""Issue #1658: pHash 阈值校准 + 颜色直方图融合 — 完整单元测试。
+
+测试覆盖：
+1. PHASH_THRESHOLD 从 10 收紧到 8
+2. 中位数替代均值（抵抗异常值）
+3. 帧匹配比例条件 (≥70%)
+4. Bhattacharyya 系数计算
+5. 直方图融合逻辑
+6. _check_fusion_duplicate 共用方法
+7. check_duplicate / check_batch_duplicate 集成
+8. compute_duplicate_rate 融合逻辑
+9. 向后兼容（无直方图数据时不崩溃）
+"""
+
+from __future__ import annotations
+
+import sys
+from unittest.mock import MagicMock, patch
+
+import numpy as np
+import pytest
+
+
+def _mock_module(**attrs):
+    """Create a mock module with __spec__ to avoid AttributeError."""
+    m = MagicMock()
+    m.__spec__ = None
+    if "__path__" not in attrs:
+        m.__path__ = []
+    for k, v in attrs.items():
+        setattr(m, k, v)
+    return m
+
+
+# ── Mock setup ──
+_SAVED_MODULES_KEYS = set(sys.modules.keys())
+_SAVED_MODULES_VALUES = {
+    k: sys.modules.get(k)
+    for k in [
+        "cv2",
+        "celery",
+        "sqlalchemy",
+        "sqlalchemy.orm",
+        "sqlalchemy.engine",
+        "sqlalchemy.ext",
+        "sqlalchemy.ext.declarative",
+        "worker_app.db",
+        "worker_app.celery_app",
+        "worker_app.core.config",
+        "packages.adapters.sqlalchemy_impl.session",
+        "packages.adapters.sqlalchemy_impl.generated_video_repository",
+        "packages.adapters.sqlalchemy_impl.models",
+        "packages.shared.config",
+        "packages.shared.storage",
+    ]
+}
+
+sys.modules["cv2"] = _mock_module()
+
+_mock_celery = MagicMock()
+_mock_celery.Task = MagicMock
+_mock_celery.Celery = MagicMock
+_mock_celery.__spec__ = None
+sys.modules["celery"] = _mock_celery
+
+_mock_sqla = MagicMock()
+_mock_sqla.__path__ = []
+_mock_sqla.__spec__ = None
+sys.modules["sqlalchemy"] = _mock_sqla
+
+_mock_sqla_orm = MagicMock()
+_mock_sqla_orm.__path__ = []
+_mock_sqla_orm.__spec__ = None
+_mock_sqla_orm.Session = MagicMock
+sys.modules["sqlalchemy.orm"] = _mock_sqla_orm
+sys.modules["sqlalchemy.engine"] = _mock_module()
+sys.modules["sqlalchemy.ext"] = _mock_module()
+sys.modules["sqlalchemy.ext.declarative"] = _mock_module()
+
+sys.modules["worker_app.db"] = _mock_module(SessionLocal=MagicMock())
+sys.modules["worker_app.celery_app"] = _mock_module(celery_app=MagicMock())
+sys.modules["worker_app.core.config"] = _mock_module(get_settings=MagicMock(return_value=MagicMock()))
+
+sys.modules["packages.adapters.sqlalchemy_impl.session"] = _mock_module(
+    Base=MagicMock(),
+    build_engine=MagicMock(),
+    build_session_factory=MagicMock(),
+    ensure_database_exists=MagicMock(),
+    initialize_database=MagicMock(),
+)
+sys.modules["packages.adapters.sqlalchemy_impl.generated_video_repository"] = _mock_module()
+sys.modules["packages.adapters.sqlalchemy_impl.models"] = _mock_module(
+    VideoFingerprintChunkModel=MagicMock(),
+    GeneratedVideoModel=MagicMock(),
+)
+sys.modules["packages.shared.config"] = _mock_module(get_shared_settings=MagicMock(return_value=MagicMock()))
+sys.modules["packages.shared.storage"] = _mock_module()
+
+# Import while mocks active
+from video_processing.dedup import (  # noqa: E402
+    VideoDeduplicator,
+    VideoFingerprint,
+    hamming_distance,
+)
+import video_processing.dedup as _dedup_module  # noqa: E402
+
+# Restore sys.modules
+for _key in list(sys.modules.keys()):
+    if _key not in _SAVED_MODULES_KEYS:
+        del sys.modules[_key]
+for _key, _value in _SAVED_MODULES_VALUES.items():
+    if _value is not None:
+        sys.modules[_key] = _value
+    elif _key in sys.modules:
+        del sys.modules[_key]
+del _SAVED_MODULES_KEYS, _SAVED_MODULES_VALUES, _key, _value
+
+
+# ── Helpers ──
+def _make_fingerprint(phashes: list[str], histograms: list[list[float]] | None = None) -> VideoFingerprint:
+    """构造测试用 VideoFingerprint."""
+    if histograms is None:
+        histograms = [[0.5] * 96 for _ in phashes]
+    return VideoFingerprint(
+        md5="test_md5",
+        keyframe_phashes=phashes,
+        color_histograms=histograms,
+        duration=30.0,
+        resolution=(1920, 1080),
+    )
+
+
+# ══════════════════════════════════════════════
+# 1. 常量验证
+# ══════════════════════════════════════════════
+class TestConstants:
+    """Issue #1658 常量验证。"""
+
+    def test_phash_threshold_is_8(self):
+        """PHASH_THRESHOLD 应为 8（从 10 收紧）。"""
+        assert VideoDeduplicator.PHASH_THRESHOLD == 8
+
+    def test_match_ratio_threshold(self):
+        assert VideoDeduplicator.MATCH_RATIO_THRESHOLD == 0.7
+
+    def test_duplicate_threshold(self):
+        assert VideoDeduplicator.DUPLICATE_THRESHOLD == 0.70
+
+    def test_phash_weight(self):
+        assert VideoDeduplicator.PHASH_WEIGHT == 0.7
+
+    def test_histogram_weight(self):
+        assert VideoDeduplicator.HISTOGRAM_WEIGHT == 0.3
+
+
+# ══════════════════════════════════════════════
+# 2. pHash 阈值变更测试
+# ══════════════════════════════════════════════
+class TestPHashThresholdChange:
+    """pHash 阈值从 10 收紧到 8 的行为验证。"""
+
+    def test_distance_9_not_matching(self):
+        """距离=9：旧阈值10会判匹配，新阈值8不匹配。"""
+        assert VideoDeduplicator.PHASH_THRESHOLD == 8
+        # 距离 9 > 8，不应匹配（strict <）
+        assert 9 >= VideoDeduplicator.PHASH_THRESHOLD
+
+    def test_distance_8_is_boundary(self):
+        """距离=8 等于阈值，不匹配（严格小于）。"""
+        assert 8 >= VideoDeduplicator.PHASH_THRESHOLD  # 不满足 d < THRESHOLD
+
+
+# ══════════════════════════════════════════════
+# 3. 中位数 vs 均值
+# ══════════════════════════════════════════════
+class TestMedianVsMean:
+    """中位数抵抗异常值。"""
+
+    def test_median_resists_outlier(self):
+        """距离 [3, 3, 3, 3, 30]：均值=8.4，中位数=3。"""
+        import statistics
+        distances = [3, 3, 3, 3, 30]
+        mean_val = sum(distances) / len(distances)
+        median_val = statistics.median(distances)
+        assert mean_val == pytest.approx(8.4)
+        assert median_val == 3
+
+    def test_median_used_in_fusion(self):
+        """验证 _check_fusion_duplicate 使用中位数。"""
+        d = VideoDeduplicator()
+        # 10帧: 9帧距离=2, 1帧距离=50
+        phashes_new = [f"{i:016x}" for i in range(10)]
+        # existing: 与 phashes_new 前9个完全相同，第10个完全不同
+        phashes_existing = phashes_new[:9] + ["ffffffffffffffff"]
+
+        fp = _make_fingerprint(
+            phashes_new,
+            [[1.0, 0.0, 0.5] * 32 for _ in range(10)],
+        )
+        existing_hist = [[1.0, 0.0, 0.5] * 32 for _ in range(10)]
+
+        result = d._check_fusion_duplicate(fp, {}, phashes_existing, existing_hist)
+        # 中位数距离应为 0（9帧距离0，1帧距离>0，中位数=0）
+        if result:
+            assert result["_debug"]["median_distance"] == 0
+
+
+# ══════════════════════════════════════════════
+# 4. 帧匹配比例测试
+# ══════════════════════════════════════════════
+class TestMatchRatio:
+    """帧匹配比例 ≥ 70% 才通过。"""
+
+    def test_insufficient_match_ratio(self):
+        """10帧中只有5帧距离<8 → match_ratio=0.5 < 0.7 → 不判重复。"""
+        d = VideoDeduplicator()
+        # 构造: 5帧完全匹配(距离0), 5帧完全不同(距离64)
+        new_phashes = ["0" * 16] * 5 + ["a" * 16] * 5
+        existing_phashes = ["0" * 16]  # 只有一帧
+
+        fp = _make_fingerprint(new_phashes)
+        result = d._check_fusion_duplicate(fp, {}, existing_phashes, [[0.5] * 96])
+        # 5/10 = 0.5 < 0.7，应返回 None
+        assert result is None
+
+    def test_sufficient_match_ratio(self):
+        """10帧中8帧距离<8 → match_ratio=0.8 ≥ 0.7 → 通过。"""
+        d = VideoDeduplicator()
+        new_phashes = ["0" * 16] * 8 + ["a" * 16] * 2
+        existing_phashes = ["0" * 16]
+
+        fp = _make_fingerprint(
+            new_phashes,
+            [[1.0, 0.0, 0.5] * 32 for _ in range(10)],
+        )
+        existing_hist = [[1.0, 0.0, 0.5] * 32]
+
+        result = d._check_fusion_duplicate(fp, {}, existing_phashes, existing_hist)
+        # 8/10 = 0.8 ≥ 0.7，应该通过（取决于combined_score）
+        if result is not None:
+            assert result["_debug"]["match_ratio"] == 0.8
+
+
+# ══════════════════════════════════════════════
+# 5. Bhattacharyya 系数测试
+# ══════════════════════════════════════════════
+class TestBhattacharyyaCoefficient:
+    """Bhattacharyya 系数计算验证。"""
+
+    def test_identical_histograms(self):
+        """相同直方图 → 1.0。"""
+        h = [0.5, 0.3, 0.2]
+        bc = VideoDeduplicator._bhattacharyya_coefficient(h, h)
+        # Σ √(a[i]*a[i]) = Σ a[i] = 1.0 (如果已归一化)
+        assert bc == pytest.approx(sum(h))
+
+    def test_all_zeros(self):
+        """全零直方图 → 0.0。"""
+        bc = VideoDeduplicator._bhattacharyya_coefficient([0.0, 0.0], [0.0, 0.0])
+        assert bc == 0.0
+
+    def test_known_value(self):
+        """已知值验证: [1,0] vs [0,1] → 0。"""
+        bc = VideoDeduplicator._bhattacharyya_coefficient([1.0, 0.0], [0.0, 1.0])
+        assert bc == pytest.approx(0.0)
+
+    def test_known_value_partial(self):
+        """已知值: [0.5, 0.5] vs [0.5, 0.5] → √0.25 + √0.25 = 1.0。"""
+        bc = VideoDeduplicator._bhattacharyya_coefficient([0.5, 0.5], [0.5, 0.5])
+        assert bc == pytest.approx(1.0)
+
+    def test_different_lengths(self):
+        """不同长度取最小长度对齐。"""
+        bc = VideoDeduplicator._bhattacharyya_coefficient([1.0, 0.0, 0.0], [1.0, 0.0])
+        # 对齐到2个元素: √1 + √0 = 1.0
+        assert bc == pytest.approx(1.0)
+
+
+# ══════════════════════════════════════════════
+# 6. 直方图相似度（_compute_histogram_similarity）
+# ══════════════════════════════════════════════
+class TestComputeHistogramSimilarity:
+    """_compute_histogram_similarity 测试。"""
+
+    def test_empty_inputs(self):
+        """空输入 → 0.0。"""
+        assert VideoDeduplicator._compute_histogram_similarity([], [[0.5]]) == 0.0
+        assert VideoDeduplicator._compute_histogram_similarity([[0.5]], []) == 0.0
+
+    def test_identical_histograms(self):
+        """相同直方图组。"""
+        h = [[0.5, 0.3, 0.2]]
+        sim = VideoDeduplicator._compute_histogram_similarity(h, h)
+        assert sim == pytest.approx(sum(h[0]))
+
+    def test_best_match_selection(self):
+        """多帧时取最佳匹配。"""
+        ha = [[1.0, 0.0], [0.0, 1.0]]
+        hb = [[0.0, 1.0]]
+        # 第一帧 [1,0] vs [0,1] → bc=0
+        # 第二帧 [0,1] vs [0,1] → bc=1.0
+        # 平均 = (0 + 1.0) / 2 = 0.5
+        sim = VideoDeduplicator._compute_histogram_similarity(ha, hb)
+        assert sim == pytest.approx(0.5)
+
+
+# ══════════════════════════════════════════════
+# 7. 融合逻辑测试
+# ══════════════════════════════════════════════
+class TestFusionLogic:
+    """_check_fusion_duplicate 融合判定测试。"""
+
+    def test_high_phash_high_hist_is_duplicate(self):
+        """pHash高相似 + 直方图高相似 → 判重复。"""
+        d = VideoDeduplicator()
+        phashes = ["0" * 16] * 10
+        hist = [[0.5] * 96] * 10
+
+        fp = _make_fingerprint(phashes, hist)
+        result = d._check_fusion_duplicate(fp, {}, phashes, hist)
+        # 完全相同 → combined_score = 0.7*1.0 + 0.3*~1.0 = ~1.0 > 0.70
+        assert result is not None
+        assert result["similarity"] > 0.70
+
+    def test_high_phash_low_hist_depends_on_score(self):
+        """pHash高相似 + 直方图低相似 → 看 combined_score。"""
+        d = VideoDeduplicator()
+        phashes = ["0" * 16] * 10
+        fp_hist = [[1.0, 0.0] * 48]  # 与 existing 完全不同
+        existing_hist = [[0.0, 1.0] * 48]
+
+        fp = _make_fingerprint(phashes, fp_hist)
+        result = d._check_fusion_duplicate(fp, {}, phashes, existing_hist)
+        # phash_similarity=1.0, hist_similarity≈0
+        # combined = 0.7*1.0 + 0.3*0 = 0.7 ≥ 0.70 → 刚好通过
+        if result:
+            assert result["_debug"]["phash_similarity"] == 1.0
+            assert result["_debug"]["hist_similarity"] == pytest.approx(0.0, abs=0.01)
+
+    def test_no_existing_phashes_returns_none(self):
+        """无已有 pHash → 返回 None。"""
+        d = VideoDeduplicator()
+        fp = _make_fingerprint(["0" * 16])
+        result = d._check_fusion_duplicate(fp, {}, [], [])
+        assert result is None
+
+
+# ══════════════════════════════════════════════
+# 8. 向后兼容测试
+# ══════════════════════════════════════════════
+class TestBackwardCompatibility:
+    """无直方图数据时不崩溃。"""
+
+    def test_no_histogram_fallback(self):
+        """已有视频无分片直方图 → hist_similarity 回退到 0.5，不崩溃。"""
+        d = VideoDeduplicator()
+        phashes = ["0" * 16] * 10
+        fp = _make_fingerprint(phashes, [[0.5] * 96] * 10)
+        # existing_histograms 为空列表
+        result = d._check_fusion_duplicate(fp, {}, phashes, [])
+        # 应该不崩溃，hist_similarity=0.5
+        if result:
+            assert result["_debug"]["hist_similarity"] == 0.5
+
+    def test_no_histogram_combined_score(self):
+        """无直方图时的 combined_score = 0.7 * phash + 0.3 * 0.5。"""
+        d = VideoDeduplicator()
+        phashes = ["0" * 16] * 10  # 完全相同
+        fp = _make_fingerprint(phashes, [[0.5] * 96] * 10)
+        result = d._check_fusion_duplicate(fp, {}, phashes, [])
+        if result:
+            expected = 0.7 * 1.0 + 0.3 * 0.5  # = 0.85
+            assert result["similarity"] == pytest.approx(expected)
+
+
+# ══════════════════════════════════════════════
+# 9. check_duplicate 集成测试
+# ══════════════════════════════════════════════
+class TestCheckDuplicateIntegration:
+    """check_duplicate 集成测试（mock DB）。"""
+
+    def test_md5_exact_match(self):
+        """MD5 精确匹配。"""
+        d = VideoDeduplicator()
+        fp = _make_fingerprint(["0" * 16])
+        fp.md5 = "exact_md5"
+
+        mock_video = MagicMock()
+        mock_video.id = "vid_1"
+        mock_video.video_fingerprint = {"md5": "exact_md5", "keyframe_phashes": ["0" * 16]}
+
+        mock_repo = MagicMock()
+        mock_repo.list_by_project.return_value = [mock_video]
+
+        mock_session = MagicMock()
+        with patch.object(_dedup_module, "SQLAlchemyGeneratedVideoRepository", return_value=mock_repo):
+            result = d.check_duplicate(fp, "proj_1", mock_session)
+
+        assert result is not None
+        assert result["reason"] == "exact_md5_match"
+        assert result["similarity"] == 1.0
+
+    def test_fusion_match_returns_correct_reason(self):
+        """融合匹配返回 phash_histogram_fusion reason。"""
+        d = VideoDeduplicator()
+        phashes = ["0" * 16] * 10
+        hist = [[0.5] * 96] * 10
+        fp = _make_fingerprint(phashes, hist)
+        fp.md5 = "different_md5"
+
+        mock_video = MagicMock()
+        mock_video.id = "vid_1"
+        mock_video.video_fingerprint = {
+            "md5": "other_md5",
+            "keyframe_phashes": phashes,
+            "color_histograms": hist,
+        }
+
+        mock_repo = MagicMock()
+        mock_repo.list_by_project.return_value = [mock_video]
+
+        mock_session = MagicMock()
+        with patch.object(_dedup_module, "SQLAlchemyGeneratedVideoRepository", return_value=mock_repo):
+            with patch.object(d, "_get_existing_chunks", return_value=[]):
+                result = d.check_duplicate(fp, "proj_1", mock_session)
+
+        assert result is not None
+        assert result["reason"] == "phash_histogram_fusion"
+
+    def test_no_duplicate_returns_none(self):
+        """完全不相似 → None。"""
+        d = VideoDeduplicator()
+        # 全部帧距离都很大
+        new_phashes = ["0" * 16] * 10
+        existing_phashes = ["f" * 16] * 10
+        fp = _make_fingerprint(new_phashes)
+
+        mock_video = MagicMock()
+        mock_video.id = "vid_1"
+        mock_video.video_fingerprint = {
+            "md5": "other_md5",
+            "keyframe_phashes": existing_phashes,
+            "color_histograms": [[0.5] * 96] * 10,
+        }
+
+        mock_repo = MagicMock()
+        mock_repo.list_by_project.return_value = [mock_video]
+
+        mock_session = MagicMock()
+        with patch.object(_dedup_module, "SQLAlchemyGeneratedVideoRepository", return_value=mock_repo):
+            with patch.object(d, "_get_existing_chunks", return_value=[]):
+                result = d.check_duplicate(fp, "proj_1", mock_session)
+
+        # 所有帧距离=64，match_ratio=0 < 0.7 → None
+        assert result is None
+
+
+# ══════════════════════════════════════════════
+# 10. check_batch_duplicate 集成测试
+# ══════════════════════════════════════════════
+class TestCheckBatchDuplicateIntegration:
+    """check_batch_duplicate 集成测试。"""
+
+    def test_batch_fusion_reason_has_prefix(self):
+        """批次内融合匹配 reason 带 batch_ 前缀。"""
+        d = VideoDeduplicator()
+        phashes = ["0" * 16] * 10
+        hist = [[0.5] * 96] * 10
+        fp = _make_fingerprint(phashes, hist)
+
+        mock_video = MagicMock()
+        mock_video.id = "vid_other"
+        mock_video.video_fingerprint = {
+            "md5": "other_md5",
+            "keyframe_phashes": phashes,
+            "color_histograms": hist,
+        }
+
+        mock_repo = MagicMock()
+        mock_repo.list_by_batch.return_value = [mock_video]
+
+        mock_session = MagicMock()
+        with patch.object(_dedup_module, "SQLAlchemyGeneratedVideoRepository", return_value=mock_repo):
+            with patch.object(d, "_get_existing_chunks", return_value=[]):
+                result = d.check_batch_duplicate(fp, "batch_1", "vid_self", mock_session)
+
+        assert result is not None
+        assert result["reason"] == "batch_phash_histogram_fusion"
+
+    def test_batch_skips_self(self):
+        """批次内排除自身。"""
+        d = VideoDeduplicator()
+        fp = _make_fingerprint(["0" * 16])
+
+        mock_video = MagicMock()
+        mock_video.id = "vid_self"
+        mock_video.video_fingerprint = {"md5": "x", "keyframe_phashes": ["0" * 16]}
+
+        mock_repo = MagicMock()
+        mock_repo.list_by_batch.return_value = [mock_video]
+
+        mock_session = MagicMock()
+        with patch.object(_dedup_module, "SQLAlchemyGeneratedVideoRepository", return_value=mock_repo):
+            result = d.check_batch_duplicate(fp, "batch_1", "vid_self", mock_session)
+
+        assert result is None
+
+
+# ══════════════════════════════════════════════
+# 11. compute_duplicate_rate 融合逻辑测试
+# ══════════════════════════════════════════════
+class TestComputeDuplicateRate:
+    """compute_duplicate_rate 融合逻辑测试。"""
+
+    def test_md5_exact_returns_100(self):
+        """MD5 精确匹配 → 100.0。"""
+        d = VideoDeduplicator()
+        fp = _make_fingerprint(["0" * 16])
+        fp.md5 = "exact_md5"
+
+        mock_model = MagicMock()
+        mock_domain = MagicMock()
+        mock_domain.id = "vid_1"
+        mock_domain.video_fingerprint = {"md5": "exact_md5", "keyframe_phashes": ["0" * 16]}
+
+        mock_repo = MagicMock()
+        mock_repo._to_domain.return_value = mock_domain
+
+        mock_session = MagicMock()
+        mock_query = MagicMock()
+        mock_query.filter.return_value = mock_query
+        mock_query.order_by.return_value = mock_query
+        mock_query.limit.return_value = mock_query
+        mock_query.all.return_value = [mock_model]
+
+        mock_generated_model = MagicMock()
+        with patch.object(_dedup_module, "SQLAlchemyGeneratedVideoRepository", return_value=mock_repo):
+            with patch.object(d, "_get_existing_chunks", return_value=[]):
+                # Patch the import inside the method
+                with patch.dict(sys.modules, {"packages.adapters.sqlalchemy_impl.models": _mock_module(GeneratedVideoModel=mock_generated_model)}):
+                    # Need to patch session.query to return mock_query
+                    mock_session.query.return_value = mock_query
+                    result = d.compute_duplicate_rate(fp, "proj_1", None, mock_session)
+
+        assert result == 100.0
+
+    def test_fusion_score_returns_percentage(self):
+        """融合分数转为百分比。"""
+        d = VideoDeduplicator()
+        phashes = ["0" * 16] * 10
+        hist = [[0.5] * 96] * 10
+        fp = _make_fingerprint(phashes, hist)
+        fp.md5 = "different_md5"
+
+        mock_model = MagicMock()
+        mock_domain = MagicMock()
+        mock_domain.id = "vid_1"
+        mock_domain.video_fingerprint = {
+            "md5": "other_md5",
+            "keyframe_phashes": phashes,
+            "color_histograms": hist,
+        }
+
+        mock_repo = MagicMock()
+        mock_repo._to_domain.return_value = mock_domain
+
+        mock_session = MagicMock()
+        mock_query = MagicMock()
+        mock_query.filter.return_value = mock_query
+        mock_query.order_by.return_value = mock_query
+        mock_query.limit.return_value = mock_query
+        mock_query.all.return_value = [mock_model]
+        mock_session.query.return_value = mock_query
+
+        mock_generated_model = MagicMock()
+        with patch.object(_dedup_module, "SQLAlchemyGeneratedVideoRepository", return_value=mock_repo):
+            with patch.object(d, "_get_existing_chunks", return_value=[]):
+                with patch.dict(sys.modules, {"packages.adapters.sqlalchemy_impl.models": _mock_module(GeneratedVideoModel=mock_generated_model)}):
+                    result = d.compute_duplicate_rate(fp, "proj_1", None, mock_session)
+
+        # 完全相同 → similarity≈1.0 → rate≈100
+        assert result > 0
+
+    def test_no_match_returns_zero(self):
+        """完全不相似 → 0.0。"""
+        d = VideoDeduplicator()
+        fp = _make_fingerprint(["0" * 16] * 10)
+
+        mock_model = MagicMock()
+        mock_domain = MagicMock()
+        mock_domain.id = "vid_1"
+        mock_domain.video_fingerprint = {
+            "md5": "other_md5",
+            "keyframe_phashes": ["f" * 16] * 10,
+            "color_histograms": [[0.5] * 96] * 10,
+        }
+
+        mock_repo = MagicMock()
+        mock_repo._to_domain.return_value = mock_domain
+
+        mock_session = MagicMock()
+        mock_query = MagicMock()
+        mock_query.filter.return_value = mock_query
+        mock_query.order_by.return_value = mock_query
+        mock_query.limit.return_value = mock_query
+        mock_query.all.return_value = [mock_model]
+        mock_session.query.return_value = mock_query
+
+        mock_generated_model = MagicMock()
+        with patch.object(_dedup_module, "SQLAlchemyGeneratedVideoRepository", return_value=mock_repo):
+            with patch.object(d, "_get_existing_chunks", return_value=[]):
+                with patch.dict(sys.modules, {"packages.adapters.sqlalchemy_impl.models": _mock_module(GeneratedVideoModel=mock_generated_model)}):
+                    result = d.compute_duplicate_rate(fp, "proj_1", None, mock_session)
+
+        # 所有帧距离=64 → match_ratio=0 → 不通过 → max_similarity=0
+        assert result == 0.0
+
+    def test_no_other_videos_returns_zero(self):
+        """无其他视频 → 0.0。"""
+        d = VideoDeduplicator()
+        fp = _make_fingerprint(["0" * 16])
+
+        mock_repo = MagicMock()
+        mock_session = MagicMock()
+        mock_query = MagicMock()
+        mock_query.filter.return_value = mock_query
+        mock_query.order_by.return_value = mock_query
+        mock_query.limit.return_value = mock_query
+        mock_query.all.return_value = []
+        mock_session.query.return_value = mock_query
+
+        mock_generated_model = MagicMock()
+        with patch.object(_dedup_module, "SQLAlchemyGeneratedVideoRepository", return_value=mock_repo):
+            with patch.dict(sys.modules, {"packages.adapters.sqlalchemy_impl.models": _mock_module(GeneratedVideoModel=mock_generated_model)}):
+                result = d.compute_duplicate_rate(fp, "proj_1", None, mock_session)
+
+        assert result == 0.0
+
+
+# ══════════════════════════════════════════════
+# 12. _average_histogram_similarity 已删除验证
+# ══════════════════════════════════════════════
+class TestOldMethodDeleted:
+    """验证旧方法已被删除。"""
+
+    def test_average_histogram_similarity_removed(self):
+        """_average_histogram_similarity 不再存在。"""
+        assert not hasattr(VideoDeduplicator, "_average_histogram_similarity")