From 503d6ccface112e30fad56c7afb6613a85705181 Mon Sep 17 00:00:00 2001 From: saas-backend-agent Date: Fri, 4 Sep 2026 14:49:26 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=9F=A5=E9=87=8D=E7=B3=BB=E7=BB=9F?= =?UTF-8?q?=E9=BB=91=E5=B1=8F=E8=A7=86=E9=A2=91=E8=BF=87=E6=BB=A4=20?= =?UTF-8?q?=E2=80=94=20=E8=B7=B3=E8=BF=87=E5=9D=8F=E6=8C=87=E7=BA=B9?= =?UTF-8?q?=E9=98=B2=E6=AD=A2=E8=99=9A=E5=81=87=E5=8C=B9=E9=85=8D=20#1664?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 问题：1秒黑屏/纯色视频的所有关键帧phash几乎相同，与任何视频 比较都会得到虚假的'匹配'结果，污染查重池。 修复： - 新增 _is_bad_fingerprint() 静态方法检测低质量指纹： - 空 phash 列表 → 坏 - 多帧但所有 phash 完全相同 → 黑屏/纯色视频 - 多帧但所有 phash 汉明距离都 < 3 → 近似黑屏 - 单帧不视为坏指纹（兼容短视频/抽帧不足） - check_duplicate / compute_duplicate_rate / check_batch_duplicate 三个方法在 MD5 精确匹配之后、pHash 比较之前加过滤 - MD5 精确匹配优先级高于坏指纹过滤 测试：13 个新单测覆盖 - _is_bad_fingerprint 8 个用例（空/单帧/全同/近似/多样/混合/已知黑屏） - check_duplicate 3 个集成测试（黑屏跳过/正常不过滤/MD5优先） - compute_duplicate_rate 2 个集成测试（黑屏排除/全黑屏零查重率） 注：Bug #1（PHASH_THRESHOLD=40）和 Bug #2（visual_similarity覆盖） 已在 PR #1674/#1684 中修复：阈值已收紧到 8，visual_similarity 只在 dup_rate > max 时更新。 --- apps/worker/video_processing/dedup.py | 54 ++++ tests/unit/test_bad_fingerprint_filter.py | 314 ++++++++++++++++++++++ 2 files changed, 368 insertions(+) create mode 100644 tests/unit/test_bad_fingerprint_filter.py diff --git a/apps/worker/video_processing/dedup.py b/apps/worker/video_processing/dedup.py index 129e35ef9..b4c6aaba3 100755 --- a/apps/worker/video_processing/dedup.py +++ b/apps/worker/video_processing/dedup.py @@ -426,6 +426,42 @@ class VideoDeduplicator: PHASH_THRESHOLD = 8 # Issue #1658: pHash 汉明距离阈值由 10 收紧到 8，降低不同视频误判率 HISTOGRAM_THRESHOLD = 0.85 + @staticmethod + def _is_bad_fingerprint(phashes: list[str]) -> bool: + """检测指纹质量差的视频（黑屏/纯色视频）。 + + 当视频有多个关键帧但所有 phash 完全相同或极其相似时， + 说明视频内容无变化（如黑屏、纯色画面），这类指纹与任何视频 + 比较都会得到虚假的"匹配"结果，应跳过。 + + 注意：单帧视频（只有 1 个 phash）不视为坏指纹，可能是短视频或抽帧不足。 + + Args: + phashes: 关键帧 phash 列表 + + Returns: + True 表示指纹无效，应跳过 + """ + if not phashes: + return True + # 单帧不视为坏指纹（短视频或抽帧不足） + if len(phashes) == 1: + return False + # 多帧但所有 phash 完全相同 → 黑屏/纯色视频 + unique = set(phashes) + if len(unique) == 1: + return True + # 多帧但所有 phash 之间的汉明距离都极小（<3）→ 近似黑屏 + phash_list = list(unique) + if len(phash_list) >= 2: + all_distances = [] + for i in range(len(phash_list)): + for j in range(i + 1, len(phash_list)): + all_distances.append(hamming_distance(phash_list[i], phash_list[j])) + if all_distances and max(all_distances) < 3: + return True + return False + def compute_fingerprint(self, video_path: str) -> VideoFingerprint: """Compute video fingerprint using dynamic keyframe detection. @@ -623,6 +659,12 @@ class VideoDeduplicator: if fingerprint.md5 == ef.get("md5"): return {"duplicate": True, "duplicate_of": existing.id, "reason": "exact_md5_match", "similarity": 1.0} + # 跳过指纹质量差的视频（黑屏/纯色视频） + existing_phashes_for_check = ef.get("keyframe_phashes", []) + if self._is_bad_fingerprint(existing_phashes_for_check): + logger.debug("Skipping bad fingerprint video %s in check_duplicate", existing.id) + continue + # 优先从分片表读取已有视频的分片 phash existing_phashes = [] chunk_data = self._get_existing_chunks(existing.id, session) @@ -737,6 +779,12 @@ class VideoDeduplicator: "similarity": 1.0, } + # 跳过指纹质量差的视频（黑屏/纯色视频） + existing_phashes_batch = ef.get("keyframe_phashes", []) + if self._is_bad_fingerprint(existing_phashes_batch): + logger.debug("Skipping bad fingerprint video %s in check_batch_duplicate", existing.id) + continue + # 优先从分片表读取 existing_phashes = [] chunk_data = self._get_existing_chunks(existing.id, session) @@ -868,6 +916,12 @@ class VideoDeduplicator: "match_count": 1, } + # 跳过指纹质量差的视频（黑屏/纯色视频） + existing_phashes_check = ef.get("keyframe_phashes", []) + if self._is_bad_fingerprint(existing_phashes_check): + logger.debug("Skipping bad fingerprint video %s in compute_duplicate_rate", existing.id) + continue + # 优先从分片表读取 existing_phashes = [] chunk_data = self._get_existing_chunks(existing.id, session) diff --git a/tests/unit/test_bad_fingerprint_filter.py b/tests/unit/test_bad_fingerprint_filter.py new file mode 100644 index 000000000..05a18cb08 --- /dev/null +++ b/tests/unit/test_bad_fingerprint_filter.py @@ -0,0 +1,314 @@ +"""Tests for bad fingerprint (black screen / uniform color) filtering. + +Issue: 1秒黑屏视频（所有帧phash几乎相同）与任何视频的距离都~30，造成虚假匹配。 +Fix: _is_bad_fingerprint() 检测并跳过这类低质量指纹。 +""" + +from __future__ import annotations + +import sys +from unittest.mock import MagicMock, patch + +import pytest + +# --------------------------------------------------------------------------- +# Mock heavy deps before importing dedup module (same pattern as test_dedup_engine.py) +# --------------------------------------------------------------------------- +_ORIGINAL_MODULES = dict(sys.modules) +_MOCKED_MODULE_NAMES: list[str] = [] + + +def _mock_if_absent(name: str, mock_obj=None): + if name not in sys.modules: + sys.modules[name] = mock_obj if mock_obj is not None else MagicMock() + _MOCKED_MODULE_NAMES.append(name) + + +_mock_if_absent("ffmpeg") +for mod_name in ["worker_app", "worker_app.celery_app", "worker_app.db"]: + _mock_if_absent(mod_name) +if "worker_app.celery_app" in sys.modules and isinstance(sys.modules["worker_app.celery_app"], MagicMock): + sys.modules["worker_app.celery_app"].celery_app = MagicMock() +if "worker_app.db" in sys.modules and isinstance(sys.modules["worker_app.db"], MagicMock): + sys.modules["worker_app.db"].SessionLocal = MagicMock() +_mock_if_absent("celery", MagicMock()) +if "celery" in sys.modules and isinstance(sys.modules["celery"], MagicMock): + sys.modules["celery"].Task = object +_mock_if_absent("packages.shared.storage") +_mock_if_absent("packages.adapters.sqlalchemy_impl.generated_video_repository") + +_HAS_CV2 = False +try: + import cv2 as _cv2 + + if not isinstance(_cv2, MagicMock): + _HAS_CV2 = True +except (ImportError, ModuleNotFoundError): + pass + +if not _HAS_CV2: + _mock_if_absent("cv2") + +import numpy as np # noqa: E402 + +from apps.worker.video_processing.dedup import ( # noqa: E402 + VideoDeduplicator, + VideoFingerprint, +) + +# Restore mocked modules +for _name in ["worker_app", "worker_app.celery_app", "worker_app.db", "celery"]: + if _name in _MOCKED_MODULE_NAMES: + sys.modules.pop(_name, None) + _MOCKED_MODULE_NAMES.remove(_name) + + +@pytest.fixture(autouse=True, scope="session") +def _cleanup_mocks(): + yield + for name in _MOCKED_MODULE_NAMES: + sys.modules.pop(name, None) + + +# ── _is_bad_fingerprint 单元测试 ───────────────────────────────── + + +class TestIsBadFingerprint: + """VideoDeduplicator._is_bad_fingerprint() 静态方法测试。""" + + def test_empty_phashes_is_bad(self): + """空 phash 列表视为坏指纹。""" + assert VideoDeduplicator._is_bad_fingerprint([]) is True + + def test_single_phash_is_not_bad(self): + """单帧视频不视为坏指纹（短视频或抽帧不足）。""" + assert VideoDeduplicator._is_bad_fingerprint(["abcdef0123456789"]) is False + + def test_all_identical_phashes_is_bad(self): + """多帧但所有 phash 完全相同 → 黑屏/纯色视频。""" + phashes = ["aaaaaaaaaaaaaaaa"] * 5 + assert VideoDeduplicator._is_bad_fingerprint(phashes) is True + + def test_two_identical_phashes_is_bad(self): + """两帧完全相同也视为坏指纹。""" + assert VideoDeduplicator._is_bad_fingerprint(["bbbbbbbbbbbbbbbb", "bbbbbbbbbbbbbbbb"]) is True + + def test_all_very_similar_phashes_is_bad(self): + """多帧 phash 之间的汉明距离都 < 3 → 近似黑屏。""" + phashes = ["0000000000000000", "0000000000000001", "0000000000000002"] + assert VideoDeduplicator._is_bad_fingerprint(phashes) is True + + def test_diverse_phashes_is_good(self): + """多样化的 phash 列表是有效指纹。""" + phashes = [ + "abcdef0123456789", + "1234567890abcdef", + "fedcba9876543210", + "0123456789abcdef", + ] + assert VideoDeduplicator._is_bad_fingerprint(phashes) is False + + def test_mixed_similar_and_different_is_good(self): + """有些 phash 相似但有足够多样的 → 有效指纹。""" + phashes = [ + "0000000000000000", + "0000000000000001", + "0000000000000002", + "ffffffffffffffff", + ] + assert VideoDeduplicator._is_bad_fingerprint(phashes) is False + + def test_known_black_screen_phashes(self): + """已知黑屏视频的 phash 特征（全零或均匀分布）。""" + assert VideoDeduplicator._is_bad_fingerprint(["0000000000000000"] * 10) is True + assert VideoDeduplicator._is_bad_fingerprint(["ffffffffffffffff"] * 8) is True + assert VideoDeduplicator._is_bad_fingerprint(["9999999999999966"] * 6) is True + + +# ── Helper ────────────────────────────────────────────────────── + + +def _make_existing_video(video_id, md5, phashes): + """创建 mock 视频记录。""" + video = MagicMock() + video.id = video_id + video.video_fingerprint = { + "md5": md5, + "keyframe_phashes": phashes, + "color_histograms": [], + } + return video + + +# ── check_duplicate 集成测试 ──────────────────────────────────── + + +class TestCheckDuplicateBadFingerprint: + """check_duplicate 跳过坏指纹视频。""" + + def test_black_screen_existing_video_skipped(self): + """已有视频是黑屏指纹 → 被跳过，不匹配。""" + deduplicator = VideoDeduplicator() + mock_session = MagicMock() + + black_screen = _make_existing_video("vid-black", "md5_black", ["aaaaaaaaaaaaaaaa"] * 5) + mock_repo = MagicMock() + mock_repo.list_by_user.return_value = [black_screen] + + fingerprint = VideoFingerprint( + md5="md5_normal", + keyframe_phashes=["aaaaaaaaaaaaaaaa"] * 5, + color_histograms=[], + duration=10.0, + resolution=(1280, 720), + ) + + with patch( + "apps.worker.video_processing.dedup.SQLAlchemyGeneratedVideoRepository", + return_value=mock_repo, + ): + result = deduplicator.check_duplicate(fingerprint, "proj-1", mock_session, scope="user", user_id="user-1") + + assert result is None + + def test_normal_existing_video_not_skipped(self): + """正常视频不会被坏指纹过滤跳过。""" + deduplicator = VideoDeduplicator() + mock_session = MagicMock() + + normal = _make_existing_video( + "vid-normal", + "md5_normal_existing", + ["abcdef0123456789", "1234567890abcdef", "fedcba9876543210"], + ) + mock_repo = MagicMock() + mock_repo.list_by_user.return_value = [normal] + + fingerprint = VideoFingerprint( + md5="md5_normal_new", + keyframe_phashes=["abcdef0123456789", "1234567890abcdef", "fedcba9876543210"], + color_histograms=[], + duration=10.0, + resolution=(1280, 720), + ) + + with patch( + "apps.worker.video_processing.dedup.SQLAlchemyGeneratedVideoRepository", + return_value=mock_repo, + ): + result = deduplicator.check_duplicate(fingerprint, "proj-1", mock_session, scope="user", user_id="user-1") + + assert result is not None + assert result["duplicate"] is True + + def test_md5_match_overrides_bad_fingerprint(self): + """MD5 精确匹配优先于坏指纹过滤。""" + deduplicator = VideoDeduplicator() + mock_session = MagicMock() + + black_screen = _make_existing_video("vid-black", "same_md5", ["aaaaaaaaaaaaaaaa"] * 5) + mock_repo = MagicMock() + mock_repo.list_by_user.return_value = [black_screen] + + fingerprint = VideoFingerprint( + md5="same_md5", + keyframe_phashes=["bbbbbbbbbbbbbbbb"] * 3, + color_histograms=[], + duration=10.0, + resolution=(1280, 720), + ) + + with patch( + "apps.worker.video_processing.dedup.SQLAlchemyGeneratedVideoRepository", + return_value=mock_repo, + ): + result = deduplicator.check_duplicate(fingerprint, "proj-1", mock_session, scope="user", user_id="user-1") + + assert result is not None + assert result["reason"] == "exact_md5_match" + + +# ── compute_duplicate_rate 集成测试 ───────────────────────────── + + +class TestComputeDuplicateRateBadFingerprint: + """compute_duplicate_rate 跳过坏指纹视频。""" + + def test_black_screen_video_excluded_from_rate(self): + """黑屏视频不参与查重率计算。""" + deduplicator = VideoDeduplicator() + mock_session = MagicMock() + + videos = [ + _make_existing_video("vid-b1", "md5_b1", ["cccccccccccccccc"] * 5), + _make_existing_video("vid-b2", "md5_b2", ["dddddddddddddddd"] * 5), + _make_existing_video("vid-b3", "md5_b3", ["eeeeeeeeeeeeeeee"] * 5), + _make_existing_video( + "vid-normal", + "md5_n", + ["abcdef0123456789", "1234567890abcdef", "fedcba9876543210"], + ), + ] + mock_repo = MagicMock() + mock_repo.list_by_user.return_value = videos + + fingerprint = VideoFingerprint( + md5="md5_new", + keyframe_phashes=["abcdef0123456789", "1234567890abcdef", "fedcba9876543210"], + color_histograms=[], + duration=10.0, + resolution=(1280, 720), + ) + + with patch( + "apps.worker.video_processing.dedup.SQLAlchemyGeneratedVideoRepository", + return_value=mock_repo, + ): + result = deduplicator.compute_duplicate_rate( + fingerprint, + "proj-1", + "vid-new", + mock_session, + scope="user", + user_id="user-1", + ) + + assert result is not None + assert isinstance(result["duplicate_rate"], float) + assert isinstance(result["match_count"], int) + + def test_only_black_screen_videos_zero_rate(self): + """所有已有视频都是黑屏 → 查重率为 0。""" + deduplicator = VideoDeduplicator() + mock_session = MagicMock() + + videos = [ + _make_existing_video("vid-b1", "md5_b1", ["aaaaaaaaaaaaaaaa"] * 5), + _make_existing_video("vid-b2", "md5_b2", ["bbbbbbbbbbbbbbbb"] * 5), + ] + mock_repo = MagicMock() + mock_repo.list_by_user.return_value = videos + + fingerprint = VideoFingerprint( + md5="md5_new", + keyframe_phashes=["aaaaaaaaaaaaaaaa"] * 5, + color_histograms=[], + duration=10.0, + resolution=(1280, 720), + ) + + with patch( + "apps.worker.video_processing.dedup.SQLAlchemyGeneratedVideoRepository", + return_value=mock_repo, + ): + result = deduplicator.compute_duplicate_rate( + fingerprint, + "proj-1", + "vid-new", + mock_session, + scope="user", + user_id="user-1", + ) + + assert result["duplicate_rate"] == 0.0 + assert result["match_count"] == 0 -- 2.54.0