xiaoxia-saas/packages/domain/render_layer_utils.py

"""渲染图层工具函数 — 纯函数集合.

从 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 的有效时长（#1749：目标段长始终为准）。

    规则：
    - duration > 0: 直接返回 duration（配音分配的目标段长；素材短于段长的部分
      由渲染侧末帧冻结 tpad / 音频 apad 铺满，不在此处钳制）；
    - duration <= 0: actual_duration，actual=0 时返回 0。

    Args:
        duration: 配置的目标时长（0 表示使用完整素材）
        actual_duration: 素材实际时长（probe 后的结果）

    Returns:
        有效时长（秒）
    """
    if duration > 0:
        return float(duration)
    return float(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 = list(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
    )

    # #1749：逐处转场重叠扣减，与 voice_duration_planner 分配口径严格一致
    # （cut 零重叠；非 cut 按该 clip 自身 transition_duration；转场挂在后段）。
    for c in clips[1:]:
        effect = (getattr(c, "transition_effect", "cut") or "cut").strip().lower()
        if effect in ("cut", "", "none"):
            continue
        tdur = float(getattr(c, "transition_duration", 0.0) or 0.0)
        if tdur > 0:
            total -= tdur
        elif transition_duration > 0:
            total -= transition_duration

    return max(0.1, round(total, 3))


# ── 直通 / 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