"""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):
        """>=8 帧且所有 phash 完全相同 → 黑屏/纯色视频（#1702：短帧不误杀）。"""
        phashes = ["aaaaaaaaaaaaaaaa"] * 10
        assert VideoDeduplicator._is_bad_fingerprint(phashes) is True

    def test_short_identical_phashes_not_bad(self):
        """<8 帧完全相同不判坏——短视频内容连续时相邻采样帧 phash 天然相同（#1702）。"""
        assert VideoDeduplicator._is_bad_fingerprint(["bbbbbbbbbbbbbbbb"] * 5) is False
        assert VideoDeduplicator._is_bad_fingerprint(["bbbbbbbbbbbbbbbb", "bbbbbbbbbbbbbbbb"]) is False

    def test_all_very_similar_phashes_is_bad(self):
        """>=8 帧 phash 之间的汉明距离都 < 3 且高占比 → 近似黑屏。"""
        phashes = ["0000000000000000"] * 8 + ["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"] * 8) is True
        # <8 帧不判坏（#1702 短视频保护）
        assert VideoDeduplicator._is_bad_fingerprint(["9999999999999966"] * 5) is False


# ── 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"] * 10)
        mock_repo = MagicMock()
        mock_repo.list_by_user.return_value = [black_screen]

        fingerprint = VideoFingerprint(
            md5="md5_normal",
            keyframe_phashes=["aaaaaaaaaaaaaaaa"] * 10,
            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"] * 10)
        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"] * 10),
            _make_existing_video("vid-b2", "md5_b2", ["cccccccccccccccc"] * 5),  # hamming(a,c)=32 > PHASH_THRESHOLD
        ]
        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
