"""画中画(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)