apps/worker/video_processing/dedup.py

@@ -5,6 +5,7 @@ Dynamic keyframe detection + sliding window temporal matching (Issue #1659).
 
 import hashlib
 import logging
+import math
 import os
 import statistics
 import tempfile
@@ -422,7 +423,7 @@ def find_duplicate_segments(
 class VideoDeduplicator:
     """Video deduplication using multiple fingerprint methods."""
 
-    PHASH_THRESHOLD = 10
+    PHASH_THRESHOLD = 8  # Issue #1658: pHash 汉明距离阈值由 10 收紧到 8，降低不同视频误判率
     HISTOGRAM_THRESHOLD = 0.85
 
     def compute_fingerprint(self, video_path: str) -> VideoFingerprint:
@@ -531,7 +532,8 @@ class VideoDeduplicator:
         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, strict=False)))
+        # 纯标准库计算（不依赖 numpy）；max(0.0, ...) 防御上游异常负值导致 sqrt domain error
+        return float(sum(math.sqrt(max(0.0, ai * bi)) for ai, bi in zip(a, b, strict=False)))
 
     @staticmethod
     def _compute_histogram_similarity(
@@ -550,6 +552,29 @@ class VideoDeduplicator:
             similarities.append(best)
         return sum(similarities) / len(similarities) if similarities else 0.0
 
+    @staticmethod
+    def _compute_fusion_score(
+        median_distance: float,
+        histograms_a: list[list[float]],
+        histograms_b: list[list[float]],
+    ) -> float:
+        """pHash 相似度与颜色直方图相似度的加权融合得分（Issue #1658）。
+
+        - phash_similarity = 1.0 - median_distance / 64（64 为 64bit pHash 最大汉明距离）
+        - hist_similarity = Bhattacharyya 系数均值；无直方图数据时回退中性值 0.5
+        - 融合得分 = PHASH_WEIGHT * phash_similarity + HISTOGRAM_WEIGHT * hist_similarity
+
+        返回 0~1 的原始得分，是否判重由调用方与 DUPLICATE_THRESHOLD 比较决定。
+        """
+        # DB 中 color_histograms 可能为 NULL（None），显式回退空列表而非 `or []`，
+        # 以保留全黑视频的全零直方图（[0,0,...] 为有效数据，空列表才走 0.5 中性回退）。
+        hist_a = histograms_a if histograms_a is not None else []
+        hist_b = histograms_b if histograms_b is not None else []
+
+        phash_similarity = 1.0 - (median_distance / 64)
+        hist_similarity = VideoDeduplicator._compute_histogram_similarity(hist_a, hist_b) if hist_b else 0.5
+        return PHASH_WEIGHT * phash_similarity + HISTOGRAM_WEIGHT * hist_similarity
+
     def check_duplicate(
         self,
         fingerprint: VideoFingerprint,
@@ -619,7 +644,7 @@ class VideoDeduplicator:
             # 帧匹配比例检查
             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 < 0.7:
+            if match_ratio < MATCH_RATIO_THRESHOLD:
                 continue
 
             # 中位数距离
@@ -627,22 +652,16 @@ class VideoDeduplicator:
             if median_distance >= self.PHASH_THRESHOLD:
                 continue
 
-            # 直方图融合
-            existing_histograms = []
+            # 直方图融合（chunk 表优先，回退 JSON 字段；JSON NULL 显式回退空列表）
             if chunk_data:
                 existing_histograms = [c["color_histogram"] for c in chunk_data if c.get("color_histogram")]
             else:
-                existing_histograms = ef.get("color_histograms", [])
+                existing_histograms = ef.get("color_histograms") or []
 
-            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._compute_fusion_score(
+                median_distance, fingerprint.color_histograms, existing_histograms
             )
-            combined_score = 0.7 * phash_similarity + 0.3 * hist_similarity
 
-            # DUPLICATE_THRESHOLD from module level
             if combined_score < DUPLICATE_THRESHOLD:
                 continue
 
@@ -737,29 +756,23 @@ class VideoDeduplicator:
             # 帧匹配比例检查
             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 < 0.7:
+            if match_ratio < MATCH_RATIO_THRESHOLD:
                 continue
 
             median_distance = statistics.median(min_distances) if min_distances else 64
             if median_distance >= self.PHASH_THRESHOLD:
                 continue
 
-            # 直方图融合
-            existing_histograms = []
+            # 直方图融合（chunk 表优先，回退 JSON 字段；JSON NULL 显式回退空列表）
             if chunk_data:
                 existing_histograms = [c["color_histogram"] for c in chunk_data if c.get("color_histogram")]
             else:
-                existing_histograms = ef.get("color_histograms", [])
+                existing_histograms = ef.get("color_histograms") or []
 
-            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._compute_fusion_score(
+                median_distance, fingerprint.color_histograms, existing_histograms
             )
-            combined_score = 0.7 * phash_similarity + 0.3 * hist_similarity
 
-            # DUPLICATE_THRESHOLD from module level
             if combined_score < DUPLICATE_THRESHOLD:
                 continue
 
@@ -901,19 +914,13 @@ class VideoDeduplicator:
 
             # visual_similarity (融合相似度，归一化 0~1)
             median_distance = statistics.median(min_distances) if min_distances else 64
-            existing_histograms = []
             if chunk_data:
                 existing_histograms = [c["color_histogram"] for c in chunk_data if c.get("color_histogram")]
             else:
-                existing_histograms = ef.get("color_histograms", [])
+                # JSON NULL 显式回退空列表
+                existing_histograms = ef.get("color_histograms") or []
 
-            phash_sim = 1.0 - median_distance / 64
-            hist_sim = (
-                self._compute_histogram_similarity(fingerprint.color_histograms, existing_histograms)
-                if existing_histograms
-                else 0.5
-            )
-            visual_sim = 0.7 * phash_sim + 0.3 * hist_sim
+            visual_sim = self._compute_fusion_score(median_distance, fingerprint.color_histograms, existing_histograms)
 
             # 判定是否为重复（融合分数超过阈值）
             if visual_sim >= DUPLICATE_THRESHOLD:

tests/unit/test_phash_threshold_calibration_1658.py

@@ -0,0 +1,273 @@
+"""Issue #1658: pHash 阈值校准 + 颜色直方图融合 — 单元测试.
+
+在 #1659（动态抽帧+滑动窗口）与 #1660（查重率）已合入 develop 的基础上，
+本测试覆盖 #1658 的最小增量改动：
+
+1. PHASH_THRESHOLD 由 10 收紧到 8（核心校准）
+2. 融合权重常量 MATCH_RATIO_THRESHOLD / PHASH_WEIGHT / HISTOGRAM_WEIGHT 实际生效
+   （不再是硬编码魔法数字）
+3. VideoDeduplicator._compute_fusion_score 统一融合得分方法：
+   - 无直方图数据时回退中性值 0.5
+   - DB NULL（None）显式回退空列表，不崩溃
+   - 全零直方图（全黑视频）为有效数据，参与 Bhattacharyya 计算
+   - 返回 0~1 原始得分，判重由调用方与 DUPLICATE_THRESHOLD 比较
+4. Bhattacharyya 系数对上游异常负值有 sqrt domain 防御
+"""
+
+from __future__ import annotations
+
+import sys
+from unittest.mock import MagicMock
+
+import pytest
+
+
+def _mock_module(**attrs):
+    """Create a mock module with __spec__ to avoid AttributeError."""
+    m = MagicMock()
+    m.__spec__ = None
+    for k, v in attrs.items():
+        setattr(m, k, v)
+    return m
+
+
+# ── Module-level setup: mock deps, import dedup, then restore sys.modules ──
+_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(
+    SQLAlchemyGeneratedVideoRepository=MagicMock
+)
+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 video_processing.dedup as _dedup_mod  # noqa: E402
+from video_processing.dedup import (  # noqa: E402
+    DUPLICATE_THRESHOLD,
+    HISTOGRAM_WEIGHT,
+    MATCH_RATIO_THRESHOLD,
+    PHASH_WEIGHT,
+    VideoDeduplicator,
+)
+
+# ── Restore sys.modules immediately after import ──
+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
+
+
+# ── 测试夹具 ─────────────────────────────────────────────────────
+
+_UNIFORM_HIST = [1.0 / 96] * 96  # 归一化均匀直方图，sum=1.0，自相似度≈1.0
+_ZERO_HIST = [0.0] * 96  # 全黑视频的全零直方图（有效数据）
+
+
+# ── TestThresholdCalibration：#1658 核心校准 ────────────────────
+
+
+class TestThresholdCalibration:
+    """pHash 阈值由 10 收紧到 8（Issue #1658）。"""
+
+    def test_phash_threshold_is_8(self):
+        """PHASH_THRESHOLD 必须为 8（旧值 10 会放过 8~9 汉明距离的不同视频）。"""
+        assert VideoDeduplicator.PHASH_THRESHOLD == 8
+
+    def test_match_ratio_threshold_constant(self):
+        assert MATCH_RATIO_THRESHOLD == 0.7
+
+    def test_duplicate_threshold_constant(self):
+        assert DUPLICATE_THRESHOLD == 0.70
+
+    def test_fusion_weights(self):
+        assert PHASH_WEIGHT == 0.7
+        assert HISTOGRAM_WEIGHT == 0.3
+
+    def test_threshold_tightening_excludes_distance_8_and_9(self):
+        """距离 8、9 的帧：旧阈值 10 下算匹配，新阈值 8 下不算匹配。
+
+        场景：5 个关键帧距离为 [7, 7, 7, 9, 9]。
+        - 旧阈值 10：5 帧全部 < 10 → match_ratio = 1.0（误放过）
+        - 新阈值 8：仅 3 帧 < 8 → match_ratio = 0.6 < 0.7（正确跳过）
+        """
+        distances = [7, 7, 7, 9, 9]
+
+        matched_old = sum(1 for d in distances if d < 10)
+        assert matched_old == 5  # 旧行为：全匹配 → 误判风险
+
+        matched_new = sum(1 for d in distances if d < VideoDeduplicator.PHASH_THRESHOLD)
+        assert matched_new == 3
+        assert matched_new / len(distances) == 0.6
+        assert matched_new / len(distances) < MATCH_RATIO_THRESHOLD  # 被帧比例门槛拦截
+
+
+# ── TestComputeFusionScore：统一融合得分方法 ────────────────────
+
+
+class TestComputeFusionScore:
+    """_compute_fusion_score(median_distance, histograms_a, histograms_b)。"""
+
+    def test_no_histogram_falls_back_to_neutral_05(self):
+        """双方均无直方图 → hist_similarity 回退 0.5。
+
+        d=0: 0.7*1.0 + 0.3*0.5 = 0.85
+        """
+        score = VideoDeduplicator._compute_fusion_score(0, [], [])
+        assert score == pytest.approx(0.85, abs=1e-6)
+
+    def test_none_histograms_treated_as_empty(self):
+        """DB NULL（None）必须显式回退空列表，不得 len(None) 崩溃。"""
+        score_none = VideoDeduplicator._compute_fusion_score(0, [], None)
+        score_empty = VideoDeduplicator._compute_fusion_score(0, [], [])
+        assert score_none == pytest.approx(score_empty, abs=1e-9)
+        assert score_none == pytest.approx(0.85, abs=1e-6)
+
+    def test_none_histograms_on_query_side_no_crash(self):
+        """查询侧直方图为 None 时同样不崩溃。"""
+        score = VideoDeduplicator._compute_fusion_score(0, None, [_UNIFORM_HIST])
+        # 查询侧无直方图 → 平均相似度为 0（无 ha 可匹配）→ 0.7*1.0 + 0.3*0 = 0.7
+        assert score == pytest.approx(0.7, abs=1e-6)
+
+    def test_identical_uniform_histograms_score_near_1(self):
+        """完全相同的归一化直方图：Bhattacharyya≈1.0 → 融合分≈1.0。"""
+        score = VideoDeduplicator._compute_fusion_score(0, [_UNIFORM_HIST], [_UNIFORM_HIST])
+        assert score == pytest.approx(1.0, abs=1e-6)
+
+    def test_all_zero_histogram_is_valid_data(self):
+        """全零直方图（全黑视频）是有效数据，Bhattacharyya=0，不得走 0.5 回退。
+
+        若错误地用 `if histograms_b` 之外的 `or []` 把全零列表清空，
+        会错误回退到 0.5，把全黑视频的相似度抬高 0.15。
+        d=0 时：正确行为 hist_sim=0 → 0.7*1.0 + 0.3*0 = 0.7；
+        若全零直方图被错误清空回退 0.5 → 0.85。
+        """
+        score = VideoDeduplicator._compute_fusion_score(0, [_ZERO_HIST], [_ZERO_HIST])
+        assert score == pytest.approx(0.7, abs=1e-6)
+        # 与错误回退值 0.85 明确区分开
+        assert abs(score - 0.85) > 0.1
+        # 注：d=0 时 phash 满分 0.7 恰达 DUPLICATE_THRESHOLD，全黑+完全相同 phash 仍判重，符合预期
+        assert score >= DUPLICATE_THRESHOLD - 1e-9
+
+    def test_score_range_within_0_1(self):
+        for d in (0, 8, 16, 32, 64):
+            score = VideoDeduplicator._compute_fusion_score(d, [_UNIFORM_HIST], [_UNIFORM_HIST])
+            assert 0.0 <= score <= 1.0
+
+    def test_formula_matches_weights(self):
+        """得分 = PHASH_WEIGHT * (1 - d/64) + HISTOGRAM_WEIGHT * hist_sim。"""
+        d = 6  # phash_sim = 1 - 6/64 = 0.90625
+        score = VideoDeduplicator._compute_fusion_score(d, [], [])  # hist 回退 0.5
+        expected = PHASH_WEIGHT * (1 - d / 64) + HISTOGRAM_WEIGHT * 0.5
+        assert score == pytest.approx(expected, abs=1e-9)
+        # 0.7*0.90625 + 0.15 = 0.634375 + 0.15 = 0.784375
+        assert score == pytest.approx(0.784375, abs=1e-6)
+
+
+# ── TestBhattacharyyaDefense：负值/异常输入防御 ─────────────────
+
+
+class TestBhattacharyyaDefense:
+    """Bhattacharyya 系数对异常输入的防御。"""
+
+    def test_negative_values_do_not_raise(self):
+        """上游异常负值不得触发 sqrt domain error（max(0.0, ai*bi) 保护）。"""
+        bad_hist = [-0.01] * 96  # 异常负值
+        coeff = VideoDeduplicator._bhattacharyya_coefficient(bad_hist, _UNIFORM_HIST)
+        # 负值乘积被钳为 0，系数为 0 而不是抛 ValueError
+        assert coeff == pytest.approx(0.0, abs=1e-9)
+
+    def test_normal_histograms_coefficient_near_1(self):
+        coeff = VideoDeduplicator._bhattacharyya_coefficient(_UNIFORM_HIST, _UNIFORM_HIST)
+        assert coeff == pytest.approx(1.0, abs=1e-6)
+
+    def test_disjoint_histograms_coefficient_0(self):
+        """完全不重叠的直方图（前半 vs 后半非零）系数为 0。"""
+        hist_a = [0.0] * 96
+        hist_b = [0.0] * 96
+        for i in range(48):
+            hist_a[i] = 1.0 / 48
+        for i in range(48, 96):
+            hist_b[i] = 1.0 / 48
+        coeff = VideoDeduplicator._bhattacharyya_coefficient(hist_a, hist_b)
+        assert coeff == pytest.approx(0.0, abs=1e-9)
+
+
+# ── TestHistogramSimilarityEdgeCases ────────────────────────────
+
+
+class TestHistogramSimilarityEdgeCases:
+    """_compute_histogram_similarity 的边界行为。"""
+
+    def test_empty_either_side_returns_0(self):
+        assert VideoDeduplicator._compute_histogram_similarity([], [_UNIFORM_HIST]) == 0.0
+        assert VideoDeduplicator._compute_histogram_similarity([_UNIFORM_HIST], []) == 0.0
+
+    def test_best_match_per_histogram(self):
+        """每个查询直方图取与目标集合的最佳匹配，再取平均。"""
+        h1 = _UNIFORM_HIST
+        h2 = [0.0] * 96
+        h2[0] = 1.0  # 与均匀直方图完全不重叠
+        # 查询侧两张直方图：h1 最佳匹配≈1.0，h2 最佳匹配≈sqrt(1/96)≈0.102
+        sim = VideoDeduplicator._compute_histogram_similarity([h1, h2], [h1])
+        assert sim == pytest.approx((1.0 + (1.0 / 96) ** 0.5) / 2, abs=1e-3)