test(wave105): extract render_layer_utils domain module (#983)

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),
+        )