xiaoxia-saas/apps/worker/video_processing/pip_engine_pure.py

"""画中画（PiP）引擎纯逻辑模块.

从 pip_engine.py 抽离的纯函数，0 FFmpeg 依赖，可完全单测。
原模块 pip_engine.py 保持不变，向后兼容。

抽离范围：
- 滤镜链构建（scale / 圆角 / 边框 / 透明度 / 动画 / overlay）
- 位置与尺寸计算辅助（封装 domain 层调用）
- 完整 PiP 滤镜链编排
- 配置验证与降级策略判断
"""

from __future__ import annotations

import logging
from pathlib import Path

from packages.domain.pip_config import (
    ANIMATION_FADE,
    ANIMATION_SLIDE_BOTTOM,
    ANIMATION_SLIDE_LEFT,
    ANIMATION_SLIDE_RIGHT,
    ANIMATION_SLIDE_TOP,
    PiPLayerConfig,
    calculate_pip_position,
    parse_size_value,
)

logger = logging.getLogger(__name__)


# ── 尺寸与位置 ────────────────────────────────────────────────────────────────


def compute_pip_size(
    layer: PiPLayerConfig,
    output_width: int,
    output_height: int,
) -> tuple[int, int]:
    """计算画中画图层的实际像素尺寸.

    Args:
        layer: 图层配置
        output_width: 输出视频宽度
        output_height: 输出视频高度

    Returns:
        (width, height) 像素值
    """
    pip_w = parse_size_value(layer.width, output_width)
    if layer.height:
        pip_h = parse_size_value(layer.height, output_height)
    else:
        # 按宽度等比例（默认 16:9）
        pip_h = int(pip_w * 9 / 16)

    # 钳制到输出尺寸内
    pip_w = max(1, min(pip_w, output_width))
    pip_h = max(1, min(pip_h, output_height))
    return pip_w, pip_h


def compute_pip_position(
    layer: PiPLayerConfig,
    pip_width: int,
    pip_height: int,
    output_width: int,
    output_height: int,
) -> tuple[int, int]:
    """计算画中画的实际位置 (x, y).

    封装 domain 层的 calculate_pip_position，
    提供默认值并做边界钳制。
    """
    x, y = calculate_pip_position(
        position=layer.position,
        output_width=output_width,
        output_height=output_height,
        pip_width=pip_width,
        pip_height=pip_height,
        margin=layer.margin,
        custom_x=layer.x,
        custom_y=layer.y,
    )

    # 边界钳制：确保不超出画面
    x = max(0, min(x, output_width - pip_width))
    y = max(0, min(y, output_height - pip_height))
    return x, y


# ── 预处理滤镜 ────────────────────────────────────────────────────────────────


def build_pip_pre_filter(
    input_label: str,
    layer: PiPLayerConfig,
    pip_width: int,
    pip_height: int,
    output_label: str,
) -> str:
    """构建单个 PiP 图层的预处理滤镜链.

    处理顺序：scale → 圆角裁剪（可选）→ 边框（可选）→ 透明度 → 动画（可选）

    Args:
        input_label: 输入标签（带方括号，如 "[1:v]"）
        layer: 图层配置
        pip_width: 缩放后的宽度（像素）
        pip_height: 缩放后的高度（像素）
        output_label: 输出标签（不带方括号）

    Returns:
        filter_complex 片段，如 "[1:v]scale=...,setsar=1[pip_pre_0]"
    """
    filters: list[str] = []

    # Step 1: scale + SAR
    filters.append(f"scale={pip_width}:{pip_height}")
    filters.append("setsar=1")

    # Step 2: 圆角裁剪
    if layer.corner_radius > 0:
        r = min(layer.corner_radius, pip_width // 2, pip_height // 2)
        # 用 geq + 圆形遮罩实现四角圆角
        filters.append(
            "format=yuva420p,"
            "geq="
            "lum='lum(X,Y)':"
            "cb='cb(X,Y)':"
            "cr='cr(X,Y)':"
            f"a='if(lt(X,{r})*lt(Y,{r}),"
            f"gt(hypot({r}-X,{r}-Y),{r})*0+1,"
            f"if(gt(X,W-{r})*lt(Y,{r}),"
            f"gt(hypot(X-(W-{r}),{r}-Y),{r})*0+1,"
            f"if(lt(X,{r})*gt(Y,H-{r}),"
            f"gt(hypot({r}-X,Y-(H-{r})),{r})*0+1,"
            f"if(gt(X,W-{r})*gt(Y,H-{r}),"
            f"gt(hypot(X-(W-{r}),Y-(H-{r})),{r})*0+1,1))))'"
        )

    # Step 3: 边框
    if layer.border_width > 0:
        bw = layer.border_width
        color = layer.border_color
        filters.append(f"pad={pip_width + 2 * bw}:{pip_height + 2 * bw}:{bw}:{bw}:{color}")

    # Step 4: 透明度
    if layer.opacity < 1.0:
        alpha = max(0.0, min(1.0, layer.opacity))
        filters.append(f"format=yuva420p,colorchannelmixer=aa={alpha}")

    # Step 5: 入场出场动画（fade 类直接在预处理中加）
    anim_filters = build_animation_filters(layer, pip_width, pip_height)
    if anim_filters:
        filters.extend(anim_filters)

    return f"{input_label}{','.join(filters)}[{output_label}]"


def build_animation_filters(
    layer: PiPLayerConfig,
    pip_width: int,
    pip_height: int,
) -> list[str]:
    """构建 fade 类入场出场动画滤镜.

    注意：slide 类动画由 overlay 表达式处理，不在此函数内。

    Returns:
        滤镜字符串列表（每项是一个完整 filter，可直接用逗号连接）
    """
    filters: list[str] = []
    anim_dur = max(0.0, layer.animation_duration)

    # 入场动画
    if layer.animation_in == ANIMATION_FADE and anim_dur > 0:
        filters.append(f"fade=t=in:st=0:d={anim_dur}:alpha=1")

    # 出场动画（需要总时长）
    if layer.animation_out == ANIMATION_FADE and anim_dur > 0 and layer.duration is not None and layer.duration > 0:
        start_fade = max(0.0, layer.duration - anim_dur)
        filters.append(f"fade=t=out:st={start_fade}:d={anim_dur}:alpha=1")

    return filters


# ── Overlay 表达式 ────────────────────────────────────────────────────────────


def build_overlay_expr(
    layer: PiPLayerConfig,
    base_x: int,
    base_y: int,
    pip_width: int,
    pip_height: int,
    output_width: int,
    output_height: int,
) -> tuple[str, str]:
    """构建 overlay 滤镜的 x/y 表达式（支持滑动动画）.

    Args:
        layer: 图层配置
        base_x: 基础 x 坐标（无动画时的最终位置）
        base_y: 基础 y 坐标
        pip_width: PiP 图层宽度
        pip_height: PiP 图层高度
        output_width: 输出视频宽度
        output_height: 输出视频高度

    Returns:
        (x_expr, y_expr) — 可直接传入 overlay= 的参数字符串
        无动画时返回纯数字字符串，有动画时返回带引号的表达式
    """
    anim_dur = max(0.0, layer.animation_duration)

    x_expr = str(base_x)
    y_expr = str(base_y)

    # ── 入场滑入动画 ──
    if anim_dur > 0:
        if layer.animation_in == ANIMATION_SLIDE_LEFT:
            # 从左侧滑入：x 从 -pip_width 变化到 base_x
            x_expr = (
                f"'{base_x}+if(lt(t,{anim_dur})," f"{-pip_width}+t/{anim_dur}*({base_x + pip_width})," f"{base_x})'"
            )
        elif layer.animation_in == ANIMATION_SLIDE_RIGHT:
            # 从右侧滑入：x 从 output_width 变化到 base_x
            x_expr = (
                f"'{base_x}+if(lt(t,{anim_dur}),"
                f"{output_width}-t/{anim_dur}*({output_width - base_x}),"
                f"{base_x})'"
            )
        elif layer.animation_in == ANIMATION_SLIDE_TOP:
            # 从顶部滑入
            y_expr = (
                f"'{base_y}+if(lt(t,{anim_dur})," f"{-pip_height}+t/{anim_dur}*({base_y + pip_height})," f"{base_y})'"
            )
        elif layer.animation_in == ANIMATION_SLIDE_BOTTOM:
            # 从底部滑入
            y_expr = (
                f"'{base_y}+if(lt(t,{anim_dur}),"
                f"{output_height}-t/{anim_dur}*({output_height - base_y}),"
                f"{base_y})'"
            )

    # ── 出场滑出动画（需要总时长） ──
    if layer.duration is not None and layer.duration > 0 and anim_dur > 0:
        out_start = layer.duration - anim_dur
        if out_start < 0:
            out_start = 0

        if layer.animation_out == ANIMATION_SLIDE_LEFT:
            # 向左滑出
            x_expr = (
                f"'{base_x}+if(gt(t,{out_start}),"
                f"{base_x}-(t-{out_start})/{anim_dur}*({base_x + pip_width}),"
                f"{base_x})'"
            )
        elif layer.animation_out == ANIMATION_SLIDE_RIGHT:
            # 向右滑出
            x_expr = (
                f"'{base_x}+if(gt(t,{out_start}),"
                f"{base_x}+(t-{out_start})/{anim_dur}*({output_width - base_x + pip_width}),"
                f"{base_x})'"
            )
        elif layer.animation_out == ANIMATION_SLIDE_TOP:
            # 向上滑出
            y_expr = (
                f"'{base_y}+if(gt(t,{out_start}),"
                f"{base_y}-(t-{out_start})/{anim_dur}*({base_y + pip_height}),"
                f"{base_y})'"
            )
        elif layer.animation_out == ANIMATION_SLIDE_BOTTOM:
            # 向下滑出
            y_expr = (
                f"'{base_y}+if(gt(t,{out_start}),"
                f"{base_y}+(t-{out_start})/{anim_dur}*({output_height - base_y + pip_height}),"
                f"{base_y})'"
            )

    return x_expr, y_expr


def build_enable_expr(
    layer: PiPLayerConfig,
) -> str:
    """构建 overlay 的 enable 时间控制表达式.

    Returns:
        enable 表达式片段，如 ":enable='between(t,1,5)'"
        无时间限制时返回空字符串
    """
    start = max(0.0, layer.start_time)
    duration = layer.duration

    if start <= 0 and (duration is None or duration <= 0):
        return ""

    if duration and duration > 0:
        end = start + duration
        return f":enable='between(t,{start},{end})'"
    else:
        return f":enable='gte(t,{start})'"


# ── 完整滤镜链 ────────────────────────────────────────────────────────────────


def build_pip_filters(
    base_label: str,
    layers: list[PiPLayerConfig],
    source_paths: list[Path | str],
    *,
    output_width: int,
    output_height: int,
    base_input_idx: int = 0,
) -> tuple[list[str], list[str], str]:
    """构建完整的画中画滤镜链和输入参数（纯函数版）.

    与 PiPEngine.build_pip_filters 对应，但不依赖类实例，
    所有参数显式传入，方便测试。

    Args:
        base_label: 底层视频标签（不带方括号）
        layers: 图层配置列表
        source_paths: 对应每个图层的源文件路径列表
        output_width: 输出视频宽度
        output_height: 输出视频高度
        base_input_idx: PiP 素材的起始输入索引

    Returns:
        (filter_parts, input_args, final_label)
        - filter_parts: 滤镜片段列表（用 ; 连接成 filter_complex）
        - input_args: 输入参数列表 ["-i", path, "-i", path, ...]
        - final_label: 最终输出标签（不带方括号）

    Raises:
        ValueError: layers 和 source_paths 长度不一致
    """
    if len(layers) != len(source_paths):
        raise ValueError(f"layers ({len(layers)}) 和 source_paths ({len(source_paths)}) 长度不一致")

    if not layers:
        return [], [], base_label

    filter_parts: list[str] = []
    input_args: list[str] = []
    current_label = base_label

    for i, (layer, path) in enumerate(zip(layers, source_paths, strict=False)):
        # 计算实际大小
        pip_w, pip_h = compute_pip_size(layer, output_width, output_height)

        # 添加输入
        input_args.extend(["-i", str(path)])

        # 实际输入索引
        actual_input_idx = base_input_idx + i

        # 预处理标签
        pre_label = f"pip_pre_{i}"

        # 构建预处理滤镜
        pre_filter = build_pip_pre_filter(
            input_label=f"[{actual_input_idx}:v]",
            layer=layer,
            pip_width=pip_w,
            pip_height=pip_h,
            output_label=pre_label,
        )
        filter_parts.append(pre_filter)

        # 计算位置
        base_x, base_y = compute_pip_position(layer, pip_w, pip_h, output_width, output_height)

        # 构建 overlay 表达式
        x_expr, y_expr = build_overlay_expr(layer, base_x, base_y, pip_w, pip_h, output_width, output_height)

        # 时间控制
        enable_expr = build_enable_expr(layer)

        # 合成标签
        combined_label = f"pip_combined_{i}"

        # overlay 滤镜
        overlay_filter = (
            f"[{current_label}][{pre_label}]" f"overlay={x_expr}:{y_expr}{enable_expr}" f"[{combined_label}]"
        )
        filter_parts.append(overlay_filter)

        current_label = combined_label

    return filter_parts, input_args, current_label


# ── 配置验证 ──────────────────────────────────────────────────────────────────


def validate_pip_layer(layer: PiPLayerConfig) -> tuple[bool, str]:
    """验证单个 PiP 图层配置是否合法.

    Returns:
        (is_valid, error_message) — 合法时 error_message 为空
    """
    errors: list[str] = []

    # 源类型检查
    if not layer.source_type:
        errors.append("source_type 不能为空")
    elif layer.source_type not in ("local_path", "asset_id", "url"):
        errors.append(f"不支持的 source_type: {layer.source_type}")

    if not layer.source:
        errors.append("source 不能为空")

    # 尺寸检查
    if layer.width is None or layer.width == "":
        errors.append("width 不能为空")

    # 位置检查
    valid_positions = {
        "top_left",
        "top_center",
        "top_right",
        "center_left",
        "center",
        "center_right",
        "bottom_left",
        "bottom_center",
        "bottom_right",
        "custom",
    }
    if layer.position not in valid_positions:
        errors.append(f"不支持的 position: {layer.position}")

    # 数值范围检查
    if layer.opacity < 0.0 or layer.opacity > 1.0:
        errors.append(f"opacity 必须在 0-1 之间: {layer.opacity}")

    if layer.corner_radius < 0:
        errors.append(f"corner_radius 不能为负: {layer.corner_radius}")

    if layer.border_width < 0:
        errors.append(f"border_width 不能为负: {layer.border_width}")

    if layer.animation_duration < 0:
        errors.append(f"animation_duration 不能为负: {layer.animation_duration}")

    if layer.start_time < 0:
        errors.append(f"start_time 不能为负: {layer.start_time}")

    if layer.duration is not None and layer.duration < 0:
        errors.append(f"duration 不能为负: {layer.duration}")

    # 动画类型检查
    valid_anims = {
        "",
        None,
        ANIMATION_FADE,
        ANIMATION_SLIDE_LEFT,
        ANIMATION_SLIDE_RIGHT,
        ANIMATION_SLIDE_TOP,
        ANIMATION_SLIDE_BOTTOM,
    }
    if layer.animation_in and layer.animation_in not in valid_anims:
        errors.append(f"不支持的 animation_in: {layer.animation_in}")
    if layer.animation_out and layer.animation_out not in valid_anims:
        errors.append(f"不支持的 animation_out: {layer.animation_out}")

    return (len(errors) == 0, "; ".join(errors))


def count_visible_layers(layers: list[PiPLayerConfig]) -> int:
    """统计可见图层数量（排除完全透明的）."""
    count = 0
    for layer in layers:
        if layer.opacity > 0:
            count += 1
    return count


def sort_layers_by_z_index(layers: list[PiPLayerConfig]) -> list[PiPLayerConfig]:
    """按 z_index 从小到大排序图层（z_index 小的先画，在底层）."""
    return sorted(layers, key=lambda layer: layer.z_index)