diff --git a/apps/worker/video_processing/dedup.py b/apps/worker/video_processing/dedup.py index 54e02e258..129e35ef9 100755 --- a/apps/worker/video_processing/dedup.py +++ b/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: diff --git a/tests/unit/test_phash_threshold_calibration_1658.py b/tests/unit/test_phash_threshold_calibration_1658.py new file mode 100644 index 000000000..6e01b4a6a --- /dev/null +++ b/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)