350 lines
13 KiB
Python
Executable File
350 lines
13 KiB
Python
Executable File
"""画中画（PiP）引擎 — 基于 FFmpeg overlay 滤镜实现多图层叠加.
|
||
|
||
支持能力：
|
||
- 多图层叠加：主画面 + 多个副画面
|
||
- 位置：9宫格 + 自由坐标（像素或百分比）
|
||
- 大小：宽高缩放（像素或百分比）
|
||
- 圆角裁剪：支持圆角矩形裁剪
|
||
- 透明度：0-100%
|
||
- 入场出场动画：淡入淡出、滑入滑出
|
||
- 时间同步：每个副画面独立开始时间和持续时长
|
||
- 降级策略：素材不存在时跳过，不阻断渲染
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from pathlib import Path
|
||
|
||
# 向后兼容：POSITION_BOTTOM_CENTER 也从 pip_config 再导出
|
||
from packages.domain.pip_config import POSITION_BOTTOM_CENTER # noqa: E402, F401
|
||
from packages.domain.pip_config import PiPConfig # noqa: F401
|
||
from packages.domain.pip_config import (
|
||
ANIMATION_FADE,
|
||
ANIMATION_SLIDE_BOTTOM,
|
||
ANIMATION_SLIDE_LEFT,
|
||
ANIMATION_SLIDE_RIGHT,
|
||
ANIMATION_SLIDE_TOP,
|
||
PiPLayerConfig,
|
||
)
|
||
from packages.domain.pip_config import ( # noqa: F401 — 向后兼容：保留模块级导出
|
||
calculate_pip_position as _calculate_pip_position_base,
|
||
)
|
||
from packages.domain.pip_config import parse_size_value as _parse_size_value_base
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ── PiP 引擎 ──────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class PiPEngine:
|
||
"""画中画引擎 — 生成 FFmpeg 滤镜链实现多图层叠加."""
|
||
|
||
def __init__(
|
||
self,
|
||
output_width: int,
|
||
output_height: int,
|
||
output_fps: int = 30,
|
||
):
|
||
self.output_width = output_width
|
||
self.output_height = output_height
|
||
self.output_fps = output_fps
|
||
|
||
def _parse_size(self, value: int | str, base: int) -> int:
|
||
"""解析尺寸值（像素或百分比）.
|
||
|
||
委托给 packages.domain.pip_config.parse_size_value 纯逻辑函数，
|
||
薄包装保留在类内以维持向后兼容。
|
||
"""
|
||
return _parse_size_value_base(value, base)
|
||
|
||
def _parse_position(
|
||
self,
|
||
layer: PiPLayerConfig,
|
||
pip_width: int,
|
||
pip_height: int,
|
||
) -> tuple[int, int]:
|
||
"""计算画中画的实际位置 (x, y).
|
||
|
||
委托给 packages.domain.pip_config.calculate_pip_position 纯逻辑函数，
|
||
薄包装保留在类内以维持向后兼容。
|
||
"""
|
||
return _calculate_pip_position_base(
|
||
position=layer.position,
|
||
output_width=self.output_width,
|
||
output_height=self.output_height,
|
||
pip_width=pip_width,
|
||
pip_height=pip_height,
|
||
margin=layer.margin,
|
||
custom_x=layer.x,
|
||
custom_y=layer.y,
|
||
)
|
||
|
||
def _build_pip_pre_filter(
|
||
self,
|
||
input_label: str,
|
||
layer: PiPLayerConfig,
|
||
pip_width: int,
|
||
pip_height: int,
|
||
output_label: str,
|
||
) -> str:
|
||
"""构建单个PiP图层的预处理滤镜链.
|
||
|
||
处理顺序：scale → 圆角裁剪（可选）→ 边框（可选）→ 透明度 → 动画（可选）
|
||
"""
|
||
filters: list[str] = []
|
||
|
||
# Step 1: scale
|
||
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 + 圆形遮罩实现圆角
|
||
# 更简单的方式：用 rounded 滤镜（FFmpeg 5.0+）或 format + alpha
|
||
# 这里用更通用的方式：创建圆角遮罩 + overlay 到透明背景
|
||
filters.append(
|
||
f"format=yuva420p,"
|
||
f"geq="
|
||
f"lum='lum(X,Y)':"
|
||
f"cb='cb(X,Y)':"
|
||
f"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 = layer.opacity
|
||
filters.append(f"format=yuva420p,colorchannelmixer=aa={alpha}")
|
||
|
||
# Step 5: 入场出场动画
|
||
if layer.animation_in or layer.animation_out:
|
||
filters.extend(self._build_animation_filters(layer, pip_width, pip_height))
|
||
|
||
filter_str = f"[{input_label}]{','.join(filters)}[{output_label}]"
|
||
return filter_str
|
||
|
||
def _build_animation_filters(
|
||
self,
|
||
layer: PiPLayerConfig,
|
||
pip_width: int,
|
||
pip_height: int,
|
||
) -> list[str]:
|
||
"""构建入场出场动画滤镜."""
|
||
filters: list[str] = []
|
||
anim_dur = layer.animation_duration
|
||
|
||
if layer.animation_in == ANIMATION_FADE:
|
||
# 淡入
|
||
filters.append(f"fade=t=in:st=0:d={anim_dur}:alpha=1")
|
||
elif layer.animation_in == ANIMATION_SLIDE_LEFT:
|
||
# 从左滑入 — 用 overlay 动态x实现，这里先标记位置表达式
|
||
pass # slide 动画在 overlay 表达式中处理
|
||
elif layer.animation_in == ANIMATION_SLIDE_RIGHT:
|
||
pass
|
||
elif layer.animation_in == ANIMATION_SLIDE_TOP:
|
||
pass
|
||
elif layer.animation_in == ANIMATION_SLIDE_BOTTOM:
|
||
pass
|
||
|
||
if layer.animation_out == ANIMATION_FADE:
|
||
# 淡出需要知道总时长，这里用表达式
|
||
if layer.duration > 0:
|
||
start_fade = layer.duration - anim_dur
|
||
filters.append(f"fade=t=out:st={max(0, start_fade)}:d={anim_dur}:alpha=1")
|
||
|
||
return filters
|
||
|
||
def _build_overlay_expr(
|
||
self,
|
||
layer: PiPLayerConfig,
|
||
base_x: int,
|
||
base_y: int,
|
||
pip_width: int,
|
||
pip_height: int,
|
||
) -> tuple[str, str]:
|
||
"""构建 overlay 滤镜的 x/y 表达式（支持滑动动画）.
|
||
|
||
Returns:
|
||
(x_expr, y_expr) — FFmpeg表达式字符串
|
||
"""
|
||
W = self.output_width
|
||
H = self.output_height
|
||
anim_dur = layer.animation_duration
|
||
|
||
x_expr = str(base_x)
|
||
y_expr = str(base_y)
|
||
|
||
# 入场滑入动画
|
||
if layer.animation_in == ANIMATION_SLIDE_LEFT:
|
||
# 从左侧滑入：x 从 -pip_width 变化到 base_x
|
||
x_expr = f"'{base_x}+(X)*0+if(lt(t,{anim_dur}),{-pip_width}+t/{anim_dur}*({base_x}+{pip_width}),{base_x})'"
|
||
elif layer.animation_in == ANIMATION_SLIDE_RIGHT:
|
||
# 从右侧滑入：x 从 W 变化到 base_x
|
||
x_expr = f"'{base_x}+if(lt(t,{anim_dur}),{W}-t/{anim_dur}*({W}-{base_x}),{base_x})'"
|
||
elif layer.animation_in == ANIMATION_SLIDE_TOP:
|
||
y_expr = f"'{base_y}+if(lt(t,{anim_dur}),{-pip_height}+t/{anim_dur}*({base_y}+{pip_height}),{base_y})'"
|
||
elif layer.animation_in == ANIMATION_SLIDE_BOTTOM:
|
||
y_expr = f"'{base_y}+if(lt(t,{anim_dur}),{H}-t/{anim_dur}*({H}-{base_y}),{base_y})'"
|
||
|
||
# 出场滑出动画（需要总时长）
|
||
if layer.duration > 0 and anim_dur > 0:
|
||
out_start = layer.duration - anim_dur
|
||
if layer.animation_out == ANIMATION_SLIDE_LEFT:
|
||
x_expr = f"'{base_x}+if(gt(t,{out_start}),{base_x}-(t-{out_start})/{anim_dur}*({base_x}+{pip_width}),{base_x})'"
|
||
elif layer.animation_out == ANIMATION_SLIDE_RIGHT:
|
||
x_expr = f"'{base_x}+if(gt(t,{out_start}),{base_x}+(t-{out_start})/{anim_dur}*({W}-{base_x}+{pip_width}),{base_x})'"
|
||
elif layer.animation_out == ANIMATION_SLIDE_TOP:
|
||
y_expr = f"'{base_y}+if(gt(t,{out_start}),{base_y}-(t-{out_start})/{anim_dur}*({base_y}+{pip_height}),{base_y})'"
|
||
elif layer.animation_out == ANIMATION_SLIDE_BOTTOM:
|
||
y_expr = f"'{base_y}+if(gt(t,{out_start}),{base_y}+(t-{out_start})/{anim_dur}*({H}-{base_y}+{pip_height}),{base_y})'"
|
||
|
||
return (x_expr, y_expr)
|
||
|
||
def build_pip_filters(
|
||
self,
|
||
base_label: str,
|
||
pip_sources: list[tuple[str, PiPLayerConfig, Path]],
|
||
*,
|
||
base_input_idx: int = 0,
|
||
) -> tuple[str, list[str], str]:
|
||
"""构建完整的画中画滤镜链和输入参数.
|
||
|
||
Args:
|
||
base_label: 底层视频的滤镜标签（如 "final_video" 或 "v0"，不带方括号）
|
||
pip_sources: [(input_label, layer_config, source_path), ...]
|
||
base_input_idx: PiP 素材在整个 FFmpeg 输入中的起始索引
|
||
|
||
Returns:
|
||
(filter_parts, input_args, final_label)
|
||
- filter_parts: 滤镜字符串列表（用 ; 连接后成为 filter_complex）
|
||
- input_args: 额外的输入参数列表 ["-i", path, "-i", path, ...]
|
||
- final_label: 最终合成后的输出标签（不带方括号）
|
||
"""
|
||
if not pip_sources:
|
||
return [], [], base_label
|
||
|
||
filter_parts: list[str] = []
|
||
input_args: list[str] = []
|
||
current_label = base_label
|
||
|
||
for i, (_input_label, layer, path) in enumerate(pip_sources):
|
||
# 添加输入
|
||
input_args.extend(["-i", str(path)])
|
||
|
||
# 计算实际大小
|
||
pip_w = self._parse_size(layer.width, self.output_width)
|
||
if layer.height:
|
||
pip_h = self._parse_size(layer.height, self.output_height)
|
||
else:
|
||
# 按宽度等比例（假设16:9，实际会scale时保持比例）
|
||
pip_h = int(pip_w * 9 / 16)
|
||
|
||
# 实际输入索引 = 起始索引 + 当前偏移
|
||
actual_input_idx = base_input_idx + i
|
||
|
||
# 预处理标签
|
||
pre_label = f"pip_pre_{i}"
|
||
|
||
# 构建预处理滤镜
|
||
pre_filter = self._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 = self._parse_position(layer, pip_w, pip_h)
|
||
|
||
# 构建overlay表达式（支持滑动动画）
|
||
x_expr, y_expr = self._build_overlay_expr(layer, base_x, base_y, pip_w, pip_h)
|
||
|
||
# 时间控制（enable表达式）
|
||
enable_expr = ""
|
||
if layer.start_time > 0 or layer.duration > 0:
|
||
start = layer.start_time
|
||
if layer.duration > 0:
|
||
end = start + layer.duration
|
||
enable_expr = f":enable='between(t,{start},{end})'"
|
||
else:
|
||
enable_expr = f":enable='gte(t,{start})'"
|
||
|
||
# 合成标签
|
||
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_layer_source(
|
||
self,
|
||
layer: PiPLayerConfig,
|
||
asset_path_map: dict[str, Path],
|
||
) -> Path | None:
|
||
"""验证图层素材是否可用，返回本地路径或None（降级跳过）.
|
||
|
||
安全：
|
||
- local_path 类型：必须在允许的目录内，防止路径遍历
|
||
- url 类型：必须通过 SSRF 安全校验
|
||
"""
|
||
from video_processing.path_security import is_in_allowed_dirs
|
||
from video_processing.url_security import UrlSecurityError, validate_url_safety
|
||
|
||
try:
|
||
if layer.source_type == "local_path":
|
||
if not layer.source:
|
||
return None
|
||
# 路径安全校验：必须在允许目录内
|
||
src_path = Path(layer.source)
|
||
if not src_path.exists():
|
||
return None
|
||
if not is_in_allowed_dirs(src_path):
|
||
logger.warning(
|
||
"PiP local_path 不在允许目录内，拒绝: %s",
|
||
layer.source[:80],
|
||
)
|
||
return None
|
||
return src_path.resolve()
|
||
elif layer.source_type == "asset_id":
|
||
if layer.source in asset_path_map:
|
||
return asset_path_map[layer.source]
|
||
return None
|
||
elif layer.source_type == "url":
|
||
# URL类型：先做SSRF安全校验，由调用者负责实际下载
|
||
try:
|
||
validate_url_safety(layer.source, purpose="pip_source")
|
||
logger.info("PiP URL 安全校验通过: %s", layer.source[:80])
|
||
except UrlSecurityError as e:
|
||
logger.warning("PiP URL 安全校验失败: %s (error=%s)", layer.source[:80], e)
|
||
return None
|
||
# 暂时不支持直接URL下载，返回None表示降级跳过
|
||
return None
|
||
except Exception as e:
|
||
logger.warning("PiP素材验证失败: %s", e)
|
||
|
||
return None
|