xiaoxia-saas/tests/unit/test_video_analyzer.py

"""video_analyzer（#2051）单元测试。

覆盖：
- 映射函数（运镜→ken_burns、转场→xfade、色调→video_filter、BPM→BGM）
- schema 常量与导出
- Farneback 光流运镜判定（合成光流场）
- BPM 档位映射
- 降级路径（ffmpeg/cv2/librosa 不可用）
- 临时目录清理
- analyze_video_style 入口在无素材时返回最小 style_guide 不抛
- build_render_params_for_clip 聚合输出
"""

from __future__ import annotations

import os
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch

import numpy as np
import pytest

from apps.worker.viral_video.video_analyzer import (  # noqa: E402
    COLOR_FILTER_PRESETS,
    DEFAULT_ANALYSIS_TIMEOUT,
    MAX_REFERENCE_DURATION_SEC,
    MAX_REFERENCE_SIZE_MB,
    STYLE_GUIDE_SCHEMA,
    TRANSITION_TO_XFADE,
    ShotBoundary,
    _detect_camera_movement,
    _pace_from_bpm,
    _rule_based_style_guide,
    analyze_video_style,
    build_render_params_for_clip,
    map_bgm_bpm,
    map_camera_to_ken_burns,
    map_color_to_video_filter,
    map_transition_to_xfade,
)

# ── 映射函数 ──────────────────────────────────────────────────────────


def test_constants_exported():
    assert MAX_REFERENCE_DURATION_SEC == 60
    assert MAX_REFERENCE_SIZE_MB == 100
    assert DEFAULT_ANALYSIS_TIMEOUT == 60
    assert "style_name" in STYLE_GUIDE_SCHEMA
    assert "ken_burns_params" in STYLE_GUIDE_SCHEMA
    assert "video_filter_eq_params" in STYLE_GUIDE_SCHEMA


# ── 运镜→ken_burns ────────────────────────────────────────────────────


@pytest.mark.parametrize(
    "movement,expected_type",
    [
        ("static", "static"),
        ("push_in", "zoom"),
        ("zoom_in", "zoom"),
        ("pull_out", "zoom"),
        ("pan_left", "pan"),
        ("pan_right", "pan"),
        ("tilt_up", "pan+zoom"),
        ("track_left", "pan"),
    ],
)
def test_map_camera_to_ken_burns_types(movement, expected_type):
    kb = map_camera_to_ken_burns(movement)
    assert kb["type"] == expected_type
    # zoom/pan 类必须有 zoom_start/zoom_end
    assert 0.8 <= kb["zoom_start"] <= 1.3
    assert 0.8 <= kb["zoom_end"] <= 1.3


def test_map_camera_unknown_falls_back_to_static():
    kb = map_camera_to_ken_burns("unknown_movement_xyz")
    assert kb["type"] == "static"
    assert kb["zoom_start"] == kb["zoom_end"] == 1.0


def test_map_camera_isolation_no_mutation():
    a = map_camera_to_ken_burns("push_in")
    a["zoom_end"] = 9.99
    b = map_camera_to_ken_burns("push_in")
    assert b["zoom_end"] != 9.99


# ── 转场→xfade ────────────────────────────────────────────────────────


@pytest.mark.parametrize(
    "ttype,expected",
    [
        ("hard_cut", "cut"),
        ("cross_dissolve", "dissolve"),
        ("fade", "fade"),
        ("fade_black", "fadeblack"),
        ("zoom_whip", "zoom"),
        ("wipe_left", "wipeleft"),
        ("slide_right", "slideleft"),
    ],
)
def test_map_transition(ttype, expected):
    assert map_transition_to_xfade(ttype) == expected


def test_map_transition_unknown_falls_back_to_cut():
    assert map_transition_to_xfade("some_random_transition") == "cut"


# ── 色调→video_filter ─────────────────────────────────────────────────


@pytest.mark.parametrize(
    "name", ["none", "warm_vintage", "cool_fresh", "high_contrast", "soft_pastel", "dramatic_cinematic"]
)
def test_map_color_presets_available(name):
    p = map_color_to_video_filter(name)
    assert isinstance(p, dict)
    # 所有预设必须能被 FFmpeg eq/colorchannelmixer 消费：eq 是 dict，ccm 是 dict
    assert "eq" in p or p == {} or "colorchannelmixer" in p


def test_map_color_unknown_is_none_preset():
    p = map_color_to_video_filter("not_a_real_filter")
    assert p == {}


def test_map_color_isolation():
    a = map_color_to_video_filter("warm_vintage")
    a["eq"]["brightness"] = 9.99
    b = map_color_to_video_filter("warm_vintage")
    assert b["eq"]["brightness"] != 9.99


# ── BPM → BGM ─────────────────────────────────────────────────────────


@pytest.mark.parametrize(
    "bpm,expected",
    [
        (0, 90),
        (70, 70),
        (120, 120),
        (200, 180),
        (30, 60),
    ],
)
def test_map_bgm_bpm(bpm, expected):
    assert map_bgm_bpm(bpm) == expected


def test_pace_from_bpm_buckets():
    assert _pace_from_bpm(120) == "fast_cut"
    assert _pace_from_bpm(110) == "fast_cut"
    assert _pace_from_bpm(90) == "medium"
    assert _pace_from_bpm(80) == "medium"
    assert _pace_from_bpm(60) == "slow_cinematic"
    assert _pace_from_bpm(0) == "medium"


# ── Farneback 光流→运镜（合成光流） ───────────────────────────────────


def _make_flow(dx: float, dy: float, w: int = 60, h: int = 40, zoom: float = 0.0):
    """构造一个合成光流场：整体平移(dx,dy)+径向发散(zoom>0=zoom in,<0=out)。"""
    ys, xs = np.mgrid[0:h, 0:w].astype(np.float32)
    cx, cy = w / 2.0, h / 2.0
    fx = dx + (xs - cx) * zoom
    fy = dy + (ys - cy) * zoom
    return np.stack([fx, fy], axis=-1).astype(np.float32)


def test_detect_movement_static():
    flow = _make_flow(0.0, 0.0, zoom=0.0)
    m, i = _detect_camera_movement(flow, 60, 40)
    assert m == "static"
    assert i == "low"


def test_detect_movement_pan_right():
    flow = _make_flow(2.0, 0.0)
    m, i = _detect_camera_movement(flow, 60, 40)
    assert m == "pan_right"
    assert i in ("medium", "high")


def test_detect_movement_pan_left():
    flow = _make_flow(-2.0, 0.0)
    m, _ = _detect_camera_movement(flow, 60, 40)
    assert m == "pan_left"


def test_detect_movement_tilt_down():
    flow = _make_flow(0.0, 2.0)
    m, _ = _detect_camera_movement(flow, 60, 40)
    assert m == "tilt_down"


def test_detect_movement_zoom_in_radial():
    # 径向向外发散 = zoom in
    flow = _make_flow(0.0, 0.0, zoom=0.08)
    m, _ = _detect_camera_movement(flow, 60, 40)
    assert m == "zoom_in"


def test_detect_movement_zoom_out_radial():
    flow = _make_flow(0.0, 0.0, zoom=-0.08)
    m, _ = _detect_camera_movement(flow, 60, 40)
    assert m == "zoom_out"


# ── 规则合成 style_guide ──────────────────────────────────────────────


def _sample_shots(n=4):
    return [
        ShotBoundary(
            index=i,
            start_sec=float(i * 3),
            end_sec=float((i + 1) * 3),
            movement=["static", "push_in", "pan_left", "zoom_in"][i],
            intensity=["low", "medium", "low", "high"][i],
            transition="hard_cut",
        )
        for i in range(n)
    ]


CAMERA_TO_KEN_BURNS_DIRS = {
    "zoom_in_slow",
    "zoom_out_slow",
    "pan_left_slow",
    "pan_right_slow",
    "zoom_in_medium",
    "zoom_out_medium",
    "diagonal_push",
    "static",
}


def test_rule_based_style_guide_structure():
    shots = _sample_shots()
    sg = _rule_based_style_guide(shots, bpm=120, vlm={"color_filter": "warm_vintage"})
    # 关键字段存在且类型正确
    assert sg["shot_count"] == 4
    assert sg["pace"] == "fast_cut"
    assert sg["bpm"] == 120
    assert sg["avg_shot_duration"] == 3.0
    assert len(sg["camera_movements"]) == 4
    assert sg["color_filter"] == "warm_vintage"
    assert "eq" in sg["video_filter_eq_params"]
    assert "default" in sg["ken_burns_params"]
    assert isinstance(sg["transition_map"], dict)
    assert isinstance(sg["ken_burns_direction_hint"], str) and sg["ken_burns_direction_hint"]


# ── build_render_params_for_clip 聚合 ─────────────────────────────────


def test_build_render_params_for_clip_shape():
    sg = _rule_based_style_guide(_sample_shots(), bpm=95, vlm={"color_filter": "cool_fresh"})
    p0 = build_render_params_for_clip(0, sg, duration_sec=3.0)
    assert "ken_burns" in p0
    assert "transition" in p0
    assert "video_filter" in p0
    assert p0["bgm_bpm_hint"] == 95
    assert p0["duration_sec"] == 3.0
    # clip 1 是 push_in → zoom
    p1 = build_render_params_for_clip(1, sg)
    assert p1["ken_burns"]["type"] == "zoom"


def test_build_render_params_high_intensity_amplifies():
    shots = _sample_shots()  # shot 3 = zoom_in/high
    sg = _rule_based_style_guide(shots, bpm=120, vlm={})
    p3 = build_render_params_for_clip(3, sg)
    base = map_camera_to_ken_burns("zoom_in")
    assert p3["ken_burns"]["zoom_end"] > base["zoom_end"]


# ── 降级与容错 ────────────────────────────────────────────────────────


def test_analyze_with_nonexistent_file_returns_minimum_guide():
    sg = analyze_video_style("/nonexistent/path/fake_video.mp4")
    assert isinstance(sg, dict)
    assert "style_name" in sg
    assert sg["shot_count"] == 0
    # 不抛异常且字段完整


def test_analyze_invalid_style_strength_defaults_to_medium():
    # 即使视频不存在，也应被规范化为 medium 并写入返回值
    with patch("apps.worker.viral_video.video_analyzer._ensure_local_video", return_value=None):
        sg = analyze_video_style("fake", style_strength="banana")
    assert sg.get("style_strength", "medium") == "medium"


def test_temp_dir_cleaned_up_after_run():
    """用临时真实空文件模拟本地路径，确认 frames 临时目录被清理。"""
    with tempfile.TemporaryDirectory() as td:
        fake = Path(td) / "fake.mp4"
        fake.write_bytes(b"")
        # 抽帧会失败（ffmpeg 对空文件失败），但应全程不抛且临时目录 rmtree
        # 直接 mock _ensure_local_video 回传不存在的文件，走 _probe_duration=0 降级路径
        with patch("apps.worker.viral_video.video_analyzer._ensure_local_video", return_value=None):
            sg = analyze_video_style("proto://fake", style_strength="light")
        assert "style_name" in sg


def test_ffmpeg_failure_falls_back_to_vlm_only_path():
    """模拟 ffmpeg 抽帧失败，仍能返回 style_guide。"""
    with tempfile.TemporaryDirectory() as td:
        fake = Path(td) / "ref.mp4"
        fake.write_bytes(b"not a real video")
        with patch(
            "apps.worker.viral_video.video_analyzer._extract_keyframes", side_effect=RuntimeError("ffmpeg exploded")
        ):
            with patch("apps.worker.viral_video.video_analyzer._detect_shots") as mock_shots:
                mock_shots.return_value = [ShotBoundary(0, 0.0, 3.0)]
                with patch("apps.worker.viral_video.video_analyzer._analyze_movements"):
                    with patch("apps.worker.viral_video.video_analyzer._detect_bpm", return_value=90):
                        with patch(
                            "apps.worker.viral_video.video_analyzer._vlm_analyze_frames",
                            return_value={"color_filter": "none"},
                        ):
                            with patch(
                                "apps.worker.viral_video.video_analyzer._llm_synthesize",
                                side_effect=lambda shots, bpm, vlm, ss: _rule_based_style_guide(shots, bpm, vlm),
                            ):
                                sg = analyze_video_style(str(fake))
        assert sg["bpm"] == 90
        assert sg["shot_count"] == 1


# ── 转场映射完整性 ────────────────────────────────────────────────────


def test_transition_map_covers_observed_types():
    for t in ("hard_cut", "cross_dissolve", "fade_black", "zoom_whip"):
        assert t in TRANSITION_TO_XFADE


# ── 预设完整性 ────────────────────────────────────────────────────────


def test_color_filter_preset_keys_are_safe_for_ffmpeg():
    for name, preset in COLOR_FILTER_PRESETS.items():
        if preset == {}:
            continue
        # eq 所有值都是数字
        for k, v in preset.get("eq", {}).items():
            assert isinstance(v, (int, float)), f"{name}.eq.{k} not numeric"
        for k, v in preset.get("colorchannelmixer", {}).items():
            assert isinstance(v, (int, float)), f"{name}.ccm.{k} not numeric"