xiaoxia
df0cc0c1b5
实现 v1.3 爆款视频参考风格分析 6 步管线： ① FFmpeg 抽关键帧（每 2s 一帧 + 场景切换帧） ② PySceneDetect ContentDetector(threshold=27) 镜头分割 ③ OpenCV Farneback 光流运镜检测（推/拉/摇/移/zoom/static + 强度） ④ librosa BPM 分析（>110 fast_cut / 80-110 medium / <80 slow_cinematic） ⑤ OSS 上传关键帧 + 豆包 VLM 色调/构图/光线分析 ⑥ 豆包 LLM 整合 style_guide JSON（失败走规则合成降级） 新增渲染参数映射： - 运镜 → ken_burns 参数（push_in→zoom_in_slow、tilt_up→pan+zoom 等 11 种） - 转场 → TransitionEngine xfade 名称（hard_cut→cut/dissolve/wipeleft 等） - 色调 → FFmpeg eq + colorchannelmixer 预设（5 套：warm_vintage/cool_fresh/high_contrast/soft_pastel/dramatic_cinematic） - BPM → BGM 选曲 BPM±5 提示 - build_render_params_for_clip() 按镜头索引聚合可直接喂 URS 的参数 降级链： - FFmpeg 抽帧失败 → VLM 路径；OpenCV 不可用 → static/low；librosa 不可用 → VLM 估计； - LLM/VLM 不可用 → 规则合成 style_guide；任意子步骤异常不抛，best-effort 返回最小可用 style_guide。 资源约束：参考视频 ≤60s/≤100MB，总超时 60s，临时帧 try/finally 清理。 依赖新增：scenedetect==0.6.4、librosa==0.10.2.post1、soundfile==0.12.1（requirements-worker.txt）。 orchestrator 改为从 apps.worker.viral_video.video_analyzer 导入 analyze_video_style， 传入 style_strength 透传 light/medium/strict 三档。 单测 48 条覆盖：映射函数/光流运镜判定/BPM 档位/规则合成/聚合输出/降级/临时文件清理。
361 lines
13 KiB
Python
361 lines
13 KiB
Python
"""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"
|