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)

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