diff --git a/apps/worker/video_processing/dedup.py b/apps/worker/video_processing/dedup.py index 1cbd4e123..3ef210ffc 100755 --- a/apps/worker/video_processing/dedup.py +++ b/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) diff --git a/tests/unit/test_dedup_pure.py b/tests/unit/test_dedup_pure.py index c9d71375e..01ecb5b17 100755 --- a/tests/unit/test_dedup_pure.py +++ b/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 diff --git a/tests/unit/test_dedup_v2.py b/tests/unit/test_dedup_v2.py new file mode 100644 index 000000000..b0f812886 --- /dev/null +++ b/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")