diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index 45a3e0c85..3a13199fa 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -52,6 +52,15 @@ from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_fr from video_processing.tts_engine import TtsEngine from video_processing.watermark_engine import WatermarkConfig, WatermarkEngine +from packages.domain.render_layer_utils import ( + LAYER_Z_INDEX as _IMPORTED_LAYER_Z_INDEX, + can_pass_through as _can_pass_through_pure, + clip_adjusted_duration as _clip_adjusted_duration_pure, + clip_effective_duration as _clip_effective_duration_pure, + clip_playback_speed as _clip_playback_speed_pure, + estimate_total_duration as _estimate_total_duration_pure, + resolve_layer_role as _resolve_layer_role_pure, +) from packages.domain.tts_config import TtsConfig logger = logging.getLogger(__name__) @@ -107,47 +116,16 @@ class RenderResult: def _resolve_layer_role(clip_type: str, config: dict[str, Any]) -> str: - """根据 clip_type 和 config.role 确定图层角色。 + """根据 clip_type 和 config.role 确定图层角色(向后兼容别名)。 - 映射规则: - intro / outro → "main"(按 order 排在首/尾) - overlay → "overlay"(画中画叠加,z=1) - corner_voice → "corner_voice"(右上角小窗,z=1) - background → "background"(全屏底图,z=0) - b_roll → "broll"(z=0) - main + config.role=b_roll → "broll" - main (default) → "main" + 实际实现移至 packages.domain.render_layer_utils.resolve_layer_role。 """ - role = config.get("role", "") - - if clip_type in ("intro", "outro"): - return "main" - if clip_type == "overlay": - return "overlay" - if clip_type == "corner_voice": - return "corner_voice" - if clip_type == "background": - return "background" - if clip_type == "b_roll": - return "broll" - # main type - if role == "b_roll": - return "broll" - if role == "audio": - return "audio" - return "main" + return _resolve_layer_role_pure(clip_type, config) # ── 图层默认 z_index ───────────────────────────────────────────────────────── -_LAYER_Z_INDEX: dict[str, int] = { - "background": -1, - "broll": 0, - "main": 0, - "overlay": 1, - "corner_voice": 1, - "audio": 2, -} +_LAYER_Z_INDEX: dict[str, int] = _IMPORTED_LAYER_Z_INDEX # 图层默认 PiP 位置(相对输出画布的偏移) _PIP_SCALE = 0.25 # PiP 占主画面的比例 @@ -489,29 +467,9 @@ class UnifiedRenderService: def _estimate_total_duration(self, layers: list[RenderLayer]) -> float: """估算视频总时长(用于字幕等需要)。 - 取主图层(main/broll/background)的总时长,转场重叠按 transition_duration 估算。 + 实际实现移至 packages.domain.render_layer_utils.estimate_total_duration。 """ - # 找主图层(第一个有视频内容的图层) - main_layer = None - for role in ("main", "broll", "background"): - for layer in layers: - if layer.role == role: - main_layer = layer - break - if main_layer: - break - - if not main_layer or not main_layer.clips: - return 0.0 - - total = sum(UnifiedRenderService._clip_adjusted_duration(c) for c in main_layer.clips) - - # 减去转场重叠时间(粗略估算) - n_clips = len(main_layer.clips) - if n_clips > 1: - total -= (n_clips - 1) * self.transition_duration - - return max(0.1, total) + return _estimate_total_duration_pure(layers, self.transition_duration) def _maybe_generate_ass(self, video_duration: float) -> Path | None: """根据 plan.config 生成 ASS 字幕文件。 @@ -1869,10 +1827,11 @@ class UnifiedRenderService: @staticmethod def _clip_effective_duration(clip: ResolvedClip) -> float: - """计算 clip 的有效时长(原速 trim 后时长).""" - if clip.duration > 0: - return min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration - return clip.actual_duration if clip.actual_duration > 0 else 0.0 + """计算 clip 的有效时长(原速 trim 后时长)。 + + 实际实现移至 packages.domain.render_layer_utils.clip_effective_duration。 + """ + return _clip_effective_duration_pure(clip.duration, clip.actual_duration) # ── 画中画(PiP)相关方法 ────────────────────────────────────────────────── @@ -1969,17 +1928,20 @@ class UnifiedRenderService: @staticmethod def _clip_speed(clip: ResolvedClip) -> float: - """获取 clip 的播放速度,无效值回退到 1.0.""" - speed = getattr(clip, "playback_speed", 1.0) - if not isinstance(speed, (int, float)) or speed <= 0: - return 1.0 - return float(speed) + """获取 clip 的播放速度,无效值回退到 1.0。 + + 实际实现移至 packages.domain.render_layer_utils.clip_playback_speed。 + """ + return _clip_playback_speed_pure(getattr(clip, "playback_speed", 1.0)) @staticmethod def _clip_adjusted_duration(clip: ResolvedClip) -> float: - """计算调速后的 clip 实际时长(用于拼接计算).""" - base = UnifiedRenderService._clip_effective_duration(clip) - speed = UnifiedRenderService._clip_speed(clip) - if abs(speed - 1.0) < 1e-6: - return base - return base / speed + """计算调速后的 clip 实际时长(用于拼接计算)。 + + 实际实现移至 packages.domain.render_layer_utils.clip_adjusted_duration。 + """ + return _clip_adjusted_duration_pure( + clip.duration, + clip.actual_duration, + getattr(clip, "playback_speed", 1.0), + ) diff --git a/packages/domain/render_layer_utils.py b/packages/domain/render_layer_utils.py new file mode 100755 index 000000000..003271f55 --- /dev/null +++ b/packages/domain/render_layer_utils.py @@ -0,0 +1,241 @@ +"""渲染图层工具函数 — 纯函数集合. + +从 unified_render_service.py 抽离的纯逻辑,负责: +- clip 时长计算(有效时长、调速后时长) +- clip_type → layer_role 映射 +- 总时长估算 +- 图层默认属性(z_index 等) + +所有函数均为纯函数,不依赖 FFmpeg、数据库或外部 IO。 +""" + +from __future__ import annotations + +from typing import Any + +# ── 图层角色定义 ───────────────────────────────────────────────────────────── + +# 图层默认 z_index 映射 +LAYER_Z_INDEX: dict[str, int] = { + "background": -1, + "broll": 0, + "main": 0, + "overlay": 1, + "corner_voice": 1, + "audio": 2, +} + +# 图层默认 PiP 缩放比例(相对于主画面) +PIP_DEFAULT_SCALE = 0.25 + +# 主视频图层角色(用于总时长计算、直通判断等) +MAIN_LAYER_ROLES = frozenset({"main", "broll", "background"}) + + +# ── clip_type → layer_role 映射 ────────────────────────────────────────────── + + +def resolve_layer_role(clip_type: str, config: dict[str, Any] | None = None) -> str: + """根据 clip_type 和 config.role 确定图层角色。 + + 映射规则: + intro / outro → "main"(按 order 排在首/尾) + overlay → "overlay"(画中画叠加,z=1) + corner_voice → "corner_voice"(右上角小窗,z=1) + background → "background"(全屏底图,z=0) + b_roll → "broll"(z=0) + main + config.role=b_roll → "broll" + main + config.role=audio → "audio" + main (default) → "main" + + Args: + clip_type: 片段类型字符串 + config: 片段配置字典(可选) + + Returns: + 图层角色字符串 + """ + role = (config or {}).get("role", "") if config else "" + + if clip_type in ("intro", "outro"): + return "main" + if clip_type == "overlay": + return "overlay" + if clip_type == "corner_voice": + return "corner_voice" + if clip_type == "background": + return "background" + if clip_type == "b_roll": + return "broll" + # main type + if role == "b_roll": + return "broll" + if role == "audio": + return "audio" + return "main" + + +def get_layer_z_index(role: str) -> int: + """获取图层角色的默认 z_index。 + + Args: + role: 图层角色 + + Returns: + z_index 值,未知角色返回 0 + """ + return LAYER_Z_INDEX.get(role, 0) + + +# ── clip 时长计算 ────────────────────────────────────────────────────────── + + +def clip_effective_duration( + duration: float, + actual_duration: float = 0.0, +) -> float: + """计算 clip 的有效时长(原速 trim 后时长)。 + + 规则: + - duration > 0: min(duration, actual_duration),actual=0 时用 duration + - duration <= 0: actual_duration,actual=0 时返回 0 + + Args: + duration: 配置的时长(0 表示使用完整素材) + actual_duration: 素材实际时长(probe 后的结果) + + Returns: + 有效时长(秒) + """ + if duration > 0: + return min(duration, actual_duration) if actual_duration > 0 else duration + return actual_duration if actual_duration > 0 else 0.0 + + +def clip_playback_speed(playback_speed: Any) -> float: + """获取 clip 的播放速度,无效值回退到 1.0。 + + Args: + playback_speed: 播放速度(可为任意类型 + + Returns: + 有效的播放速度(正数) + """ + if not isinstance(playback_speed, (int, float)): + return 1.0 + if playback_speed <= 0: + return 1.0 + return float(playback_speed) + + +def clip_adjusted_duration( + duration: float, + actual_duration: float = 0.0, + playback_speed: Any = 1.0, +) -> float: + """计算调速后的 clip 实际时长(用于拼接计算)。 + + Args: + duration: 配置的时长 + actual_duration: 素材实际时长 + playback_speed: 播放速度 + + Returns: + 调速后的时长 + """ + base = clip_effective_duration(duration, actual_duration) + speed = clip_playback_speed(playback_speed) + if abs(speed - 1.0) < 1e-6: + return base + return base / speed + + +# ── 总时长估算 ──────────────────────────────────────────────────────────── + + +def estimate_total_duration( + layers: list[Any], + transition_duration: float = 0.0, +) -> float: + """估算视频总时长。 + + 取主图层(main/broll/background)的总调整后时长,减去转场重叠时间。 + + Args: + layers: 图层列表(每个元素需有 role 和 clips 属性, + clips 中元素需有 duration/actual_duration/playback_speed 属性) + transition_duration: 转场时长(秒),用于估算重叠时间 + + Returns: + 估算的总时长(秒),最小 0.1 + """ + # 找主图层(第一个有视频内容的图层) + main_layer = None + for role in ("main", "broll", "background"): + for layer in layers: + if getattr(layer, "role", None) == role and getattr(layer, "clips", None): + main_layer = layer + break + if main_layer: + break + + if not main_layer or not getattr(main_layer, "clips", None): + return 0.0 + + clips = getattr(main_layer, "clips", []) + total = sum( + clip_adjusted_duration( + duration=getattr(c, "duration", 0), + actual_duration=getattr(c, "actual_duration", 0.0), + playback_speed=getattr(c, "playback_speed", 1.0), + ) + for c in clips + ) + + # 减去转场重叠时间(粗略估算) + n_clips = len(clips) + if n_clips > 1 and transition_duration > 0: + total -= (n_clips - 1) * transition_duration + + return max(0.1, total) + + +# ── 直通 / Stream Copy 判断辅助 ────────────────────────────────────────────── + + +def can_pass_through( + layers: list[Any], + has_stickers: bool = False, + has_watermark: bool = False, +) -> bool: + """判断是否可以走直通优化路径(单 clip 简单场景)。 + + 条件: + 1. 只有 1 个图层 + 2. 该图层是视频图层(main/broll/background) + 3. 该图层只有 1 个 clip(无转场需求) + 4. 没有贴纸 + 5. 没有水印 + + Args: + layers: 图层列表 + has_stickers: 是否有贴纸 + has_watermark: 是否有水印 + + Returns: + 是否可以走直通 + """ + if len(layers) != 1: + return False + layer = layers[0] + role = getattr(layer, "role", "") + if role not in MAIN_LAYER_ROLES: + return False + clips = getattr(layer, "clips", []) + if len(clips) != 1: + return False + if has_stickers: + return False + if has_watermark: + return False + return True diff --git a/tests/unit/test_render_layer_utils.py b/tests/unit/test_render_layer_utils.py new file mode 100755 index 000000000..65f8d2eb0 --- /dev/null +++ b/tests/unit/test_render_layer_utils.py @@ -0,0 +1,361 @@ +"""render_layer_utils 模块单元测试.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +import pytest + +from packages.domain.render_layer_utils import ( + LAYER_Z_INDEX, + MAIN_LAYER_ROLES, + PIP_DEFAULT_SCALE, + can_pass_through, + clip_adjusted_duration, + clip_effective_duration, + clip_playback_speed, + estimate_total_duration, + get_layer_z_index, + resolve_layer_role, +) + +# ── 辅助数据类 ────────────────────────────────────────────────────────────── + + +@dataclass +class FakeClip: + duration: float = 0.0 + actual_duration: float = 0.0 + playback_speed: Any = 1.0 + + +@dataclass +class FakeLayer: + role: str = "main" + clips: list[FakeClip] = field(default_factory=list) + + +# ── 常量验证 ──────────────────────────────────────────────────────────────── + + +class TestConstants: + def test_layer_z_index_has_expected_keys(self): + assert set(LAYER_Z_INDEX.keys()) == { + "background", + "broll", + "main", + "overlay", + "corner_voice", + "audio", + } + + def test_layer_z_index_ordering(self): + assert LAYER_Z_INDEX["background"] < LAYER_Z_INDEX["main"] + assert LAYER_Z_INDEX["main"] == LAYER_Z_INDEX["broll"] + assert LAYER_Z_INDEX["overlay"] > LAYER_Z_INDEX["main"] + assert LAYER_Z_INDEX["corner_voice"] > LAYER_Z_INDEX["main"] + assert LAYER_Z_INDEX["audio"] > LAYER_Z_INDEX["overlay"] + + def test_pip_default_scale_positive(self): + assert 0 < PIP_DEFAULT_SCALE < 1 + + def test_main_layer_roles(self): + assert "main" in MAIN_LAYER_ROLES + assert "broll" in MAIN_LAYER_ROLES + assert "background" in MAIN_LAYER_ROLES + assert "overlay" not in MAIN_LAYER_ROLES + + +# ── resolve_layer_role ────────────────────────────────────────────────────── + + +class TestResolveLayerRole: + def test_intro_maps_to_main(self): + assert resolve_layer_role("intro") == "main" + + def test_outro_maps_to_main(self): + assert resolve_layer_role("outro") == "main" + + def test_overlay_maps_to_overlay(self): + assert resolve_layer_role("overlay") == "overlay" + + def test_corner_voice_maps_to_corner_voice(self): + assert resolve_layer_role("corner_voice") == "corner_voice" + + def test_background_maps_to_background(self): + assert resolve_layer_role("background") == "background" + + def test_b_roll_maps_to_broll(self): + assert resolve_layer_role("b_roll") == "broll" + + def test_main_defaults_to_main(self): + assert resolve_layer_role("main") == "main" + + def test_main_with_b_roll_role(self): + assert resolve_layer_role("main", {"role": "b_roll"}) == "broll" + + def test_main_with_audio_role(self): + assert resolve_layer_role("main", {"role": "audio"}) == "audio" + + def test_main_with_other_role_stays_main(self): + assert resolve_layer_role("main", {"role": "overlay"}) == "main" + + def test_none_config(self): + assert resolve_layer_role("main", None) == "main" + + def test_empty_config(self): + assert resolve_layer_role("main", {}) == "main" + + def test_unknown_type_defaults_to_main(self): + assert resolve_layer_role("unknown_type") == "main" + + +# ── get_layer_z_index ────────────────────────────────────────────────────── + + +class TestGetLayerZIndex: + def test_known_roles(self): + for role, expected in LAYER_Z_INDEX.items(): + assert get_layer_z_index(role) == expected + + def test_unknown_role_returns_zero(self): + assert get_layer_z_index("nonexistent") == 0 + + def test_empty_string_returns_zero(self): + assert get_layer_z_index("") == 0 + + +# ── clip_effective_duration ──────────────────────────────────────────────── + + +class TestClipEffectiveDuration: + def test_explicit_duration_no_actual(self): + assert clip_effective_duration(5.0) == 5.0 + + def test_explicit_duration_with_shorter_actual(self): + assert clip_effective_duration(5.0, 3.0) == 3.0 + + def test_explicit_duration_with_longer_actual(self): + assert clip_effective_duration(5.0, 10.0) == 5.0 + + def test_zero_duration_uses_actual(self): + assert clip_effective_duration(0, 8.0) == 8.0 + + def test_negative_duration_uses_actual(self): + assert clip_effective_duration(-1.0, 8.0) == 8.0 + + def test_zero_duration_zero_actual(self): + assert clip_effective_duration(0, 0) == 0.0 + + def test_no_args_returns_zero(self): + assert clip_effective_duration(0) == 0.0 + + def test_equal_duration_and_actual(self): + assert clip_effective_duration(5.0, 5.0) == 5.0 + + +# ── clip_playback_speed ──────────────────────────────────────────────────── + + +class TestClipPlaybackSpeed: + def test_normal_speed(self): + assert clip_playback_speed(1.0) == 1.0 + + def test_fast_speed(self): + assert clip_playback_speed(2.0) == 2.0 + + def test_slow_speed(self): + assert clip_playback_speed(0.5) == 0.5 + + def test_zero_speed_defaults_to_one(self): + assert clip_playback_speed(0) == 1.0 + + def test_negative_speed_defaults_to_one(self): + assert clip_playback_speed(-1.0) == 1.0 + + def test_none_defaults_to_one(self): + assert clip_playback_speed(None) == 1.0 + + def test_string_defaults_to_one(self): + assert clip_playback_speed("fast") == 1.0 + + def test_int_speed(self): + assert clip_playback_speed(2) == 2.0 + + +# ── clip_adjusted_duration ───────────────────────────────────────────────── + + +class TestClipAdjustedDuration: + def test_normal_speed_same_as_effective(self): + assert clip_adjusted_duration(5.0, 10.0, 1.0) == 5.0 + + def test_double_speed_half_duration(self): + assert clip_adjusted_duration(10.0, 10.0, 2.0) == pytest.approx(5.0) + + def test_half_speed_double_duration(self): + assert clip_adjusted_duration(5.0, 10.0, 0.5) == pytest.approx(10.0) + + def test_invalid_speed_uses_default(self): + assert clip_adjusted_duration(5.0, 10.0, 0) == 5.0 + + def test_zero_duration(self): + assert clip_adjusted_duration(0, 0, 1.0) == 0.0 + + def test_actual_duration_only(self): + assert clip_adjusted_duration(0, 8.0, 1.0) == 8.0 + + def test_actual_duration_only_with_speed(self): + assert clip_adjusted_duration(0, 8.0, 2.0) == pytest.approx(4.0) + + def test_very_close_to_normal_speed(self): + # 1.0000001 应该被认为接近 1.0,不做除法 + result = clip_adjusted_duration(5.0, 10.0, 1.0 + 1e-10) + assert result == 5.0 + + +# ── estimate_total_duration ──────────────────────────────────────────────── + + +class TestEstimateTotalDuration: + def test_empty_layers(self): + assert estimate_total_duration([]) == 0.0 + + def test_no_main_layer(self): + layers = [FakeLayer(role="overlay", clips=[FakeClip(duration=5.0)])] + assert estimate_total_duration(layers) == 0.0 + + def test_single_clip_main_layer(self): + layers = [FakeLayer(role="main", clips=[FakeClip(duration=5.0)])] + assert estimate_total_duration(layers) == pytest.approx(5.0) + + def test_multiple_clips_no_transition(self): + layers = [ + FakeLayer( + role="main", + clips=[ + FakeClip(duration=3.0), + FakeClip(duration=2.0), + FakeClip(duration=5.0), + ], + ) + ] + assert estimate_total_duration(layers) == pytest.approx(10.0) + + def test_multiple_clips_with_transition(self): + layers = [ + FakeLayer( + role="main", + clips=[ + FakeClip(duration=3.0), + FakeClip(duration=2.0), + FakeClip(duration=5.0), + ], + ) + ] + # 3 + 2 + 5 - 2 * 0.5 = 9.0 + assert estimate_total_duration(layers, transition_duration=0.5) == pytest.approx(9.0) + + def test_prefers_main_over_broll(self): + layers = [ + FakeLayer(role="broll", clips=[FakeClip(duration=10.0)]), + FakeLayer(role="main", clips=[FakeClip(duration=5.0)]), + ] + assert estimate_total_duration(layers) == pytest.approx(5.0) + + def test_prefers_broll_over_background(self): + layers = [ + FakeLayer(role="background", clips=[FakeClip(duration=10.0)]), + FakeLayer(role="broll", clips=[FakeClip(duration=5.0)]), + ] + assert estimate_total_duration(layers) == pytest.approx(5.0) + + def test_main_layer_empty_clips(self): + layers = [FakeLayer(role="main", clips=[])] + assert estimate_total_duration(layers) == 0.0 + + def test_minimum_duration(self): + layers = [ + FakeLayer( + role="main", + clips=[ + FakeClip(duration=0.01), + FakeClip(duration=0.01), + ], + ) + ] + result = estimate_total_duration(layers, transition_duration=0.5) + assert result >= 0.1 + + def test_with_playback_speed(self): + layers = [ + FakeLayer( + role="main", + clips=[ + FakeClip(duration=10.0, playback_speed=2.0), + FakeClip(duration=10.0, playback_speed=0.5), + ], + ) + ] + # 5 + 20 = 25 + assert estimate_total_duration(layers) == pytest.approx(25.0) + + +# ── can_pass_through ────────────────────────────────────────────────────── + + +class TestCanPassThrough: + def test_single_main_clip_no_effects(self): + layers = [FakeLayer(role="main", clips=[FakeClip(duration=5.0)])] + assert can_pass_through(layers) is True + + def test_single_broll_clip(self): + layers = [FakeLayer(role="broll", clips=[FakeClip(duration=5.0)])] + assert can_pass_through(layers) is True + + def test_single_background_clip(self): + layers = [FakeLayer(role="background", clips=[FakeClip(duration=5.0)])] + assert can_pass_through(layers) is True + + def test_multiple_layers(self): + layers = [ + FakeLayer(role="main", clips=[FakeClip(duration=5.0)]), + FakeLayer(role="overlay", clips=[FakeClip(duration=3.0)]), + ] + assert can_pass_through(layers) is False + + def test_overlay_layer(self): + layers = [FakeLayer(role="overlay", clips=[FakeClip(duration=5.0)])] + assert can_pass_through(layers) is False + + def test_multiple_clips_in_layer(self): + layers = [ + FakeLayer( + role="main", + clips=[ + FakeClip(duration=3.0), + FakeClip(duration=2.0), + ], + ) + ] + assert can_pass_through(layers) is False + + def test_with_stickers(self): + layers = [FakeLayer(role="main", clips=[FakeClip(duration=5.0)])] + assert can_pass_through(layers, has_stickers=True) is False + + def test_with_watermark(self): + layers = [FakeLayer(role="main", clips=[FakeClip(duration=5.0)])] + assert can_pass_through(layers, has_watermark=True) is False + + def test_with_stickers_and_watermark(self): + layers = [FakeLayer(role="main", clips=[FakeClip(duration=5.0)])] + assert can_pass_through(layers, has_stickers=True, has_watermark=True) is False + + def test_empty_layer_list(self): + assert can_pass_through([]) is False + + def test_audio_layer_only(self): + layers = [FakeLayer(role="audio", clips=[FakeClip(duration=5.0)])] + assert can_pass_through(layers) is False