"""统一渲染引擎 — 输入 EditPlan + EditPlanClips，按时间线+图层渲染视频. 核心原则（灵应）：渲染引擎是统一的，不判断模式，只按 clip_type/config.role 分组为图层再合成。 图层分组： main (无 config.role) → main (z=0) main + config.role=b_roll → broll (z=0，与 main 同层替换) overlay → overlay (z=1，画中画叠加) background → background (z=0，全屏底图) corner_voice → corner_voice (z=1，右上角小窗) b_roll → broll (z=0) intro / outro → main (z=0，按 order 排在首/尾) 合成流程： 1. 每个 clip 先 trim + scale + setpts 预处理 2. 同层 clips 按 order 用 xfade 串联 3. overlay/corner_voice 层 overlay 到主层 4. 如有独立音频轨，amix 混入 """ from __future__ import annotations import logging import os import subprocess from dataclasses import dataclass, field from pathlib import Path from typing import Any from video_processing.ffmpeg_utils import ( DEFAULT_FPS, DEFAULT_OUTPUT_HEIGHT, DEFAULT_OUTPUT_WIDTH, DEFAULT_TRANSITION_DURATION, FFMPEG_BIN, build_xfade_filter_chain, probe_duration, probe_video_info, run_ffmpeg, ) logger = logging.getLogger(__name__) # ── 常量 ────────────────────────────────────────────────────────────────────── # Title/Subtitle 默认边距（像素） TITLE_MARGIN_TOP = 60 TITLE_MARGIN_BOTTOM = 60 TITLE_MARGIN_SIDE = 40 # ── 数据结构 ────────────────────────────────────────────────────────────────── @dataclass class ResolvedClip: """已解析到本地路径的片段。""" clip_id: str asset_id: str local_path: Path clip_type: str order: int start_time: float = 0.0 duration: float = 0.0 # 0 表示使用素材完整时长 transition_effect: str = "cut" config: dict[str, Any] = field(default_factory=dict) # 运行时填充 actual_duration: float = 0.0 # 素材实际时长（probe 后填充） @dataclass class RenderLayer: """渲染图层。""" role: str # "main" | "overlay" | "pip" | "background" | "corner_voice" | "broll" | "audio" clips: list[ResolvedClip] = field(default_factory=list) z_index: int = 0 opacity: float = 1.0 position: tuple[int, int] | None = None # (x, y) 偏移，None 表示全屏 @dataclass class RenderResult: """渲染结果。""" output_path: Path duration: float file_size: int width: int height: int # ── clip_type → layer role 映射 ────────────────────────────────────────────── # ── ASS 字幕工具 ───────────────────────────────────────────────────────────── def _hex_to_ass_color(hex_color: str) -> str: """将 HEX 颜色（#RRGGBB）转换为 ASS &HBBGGRR 格式。""" hex_color = hex_color.lstrip("#") if len(hex_color) != 6: return "&H000000" r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6] return f"&H{b.upper()}{g.upper()}{r.upper()}" def _position_to_ass_alignment(position: str) -> int: """将文字位置映射为 ASS \an 对齐编号。 ASS 对齐编号（数字小键盘布局）： 7 8 9 4 5 6 1 2 3 """ mapping = { "top": 8, # 顶部居中 "center": 5, # 居中 "bottom": 2, # 底部居中 } return mapping.get(position, 8) def _build_ass_style( style_name: str, *, font_name: str = "思源黑体", font_size: int = 48, primary_color: str = "&H00FFFFFF", outline_color: str = "&H00000000", outline_width: float = 1.0, shadow_blur: float = 0.0, shadow_offset: tuple[int, int] = (0, 0), bold: bool = False, italic: bool = False, alignment: int = 8, margin_v: int = 60, margin_l: int = 40, margin_r: int = 40, ) -> str: """构建 ASS Style 行。 Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding """ bold_val = -1 if bold else 0 italic_val = -1 if italic else 0 # BackColour 用于阴影（BorderStyle=1 时 outline + shadow） back_color = primary_color # 阴影颜色默认同文字色（带透明度由阴影模糊控制） # Shadow 值：ASS 中 Shadow 字段是阴影偏移距离（像素）， # 我们用 shadow_offset[1] 作为纵向偏移，模糊由 BorderStyle=3 实现 # 简化：BorderStyle=1（outline + drop shadow），Shadow 字段表示阴影深度 shadow_depth = shadow_offset[1] if shadow_blur > 0 else 0 return ( f"Style: {style_name},{font_name},{font_size},{primary_color}," f"&H000000FF,{outline_color},{back_color}," f"{bold_val},{italic_val},0,0,100,100,0,0," f"1,{outline_width},{shadow_depth},{alignment}," f"{margin_l},{margin_r},{margin_v},1" ) def generate_ass_subtitles( output_path: Path, *, video_width: int, video_height: int, video_duration: float, title_text: str = "", title_config: dict[str, Any] | None = None, subtitle_text: str = "", subtitle_config: dict[str, Any] | None = None, ) -> Path: """生成 ASS 字幕文件。 支持 Title（标题）和 Subtitle（字幕）两种字幕类型， 各自可独立配置样式、位置和内容。 Args: output_path: 输出 ASS 文件路径 video_width: 视频宽度（用于 ASS PlayResX） video_height: 视频高度（用于 ASS PlayResY） video_duration: 视频总时长（秒），字幕显示整个时长 title_text: 标题文本 title_config: 标题样式配置（TitleConfig dict） subtitle_text: 字幕文本 subtitle_config: 字幕样式配置（SubtitleConfig dict） Returns: 生成的 ASS 文件路径 """ title_config = title_config or {} subtitle_config = subtitle_config or {} title_enabled = title_config.get("enabled", True) and bool(title_text.strip()) subtitle_enabled = subtitle_config.get("enabled", True) and bool(subtitle_text.strip()) if not title_enabled and not subtitle_enabled: # 没有字幕，生成空文件（仍返回路径，调用方自行判断是否使用） output_path.write_text("", encoding="utf-8") return output_path styles: list[str] = [] events: list[str] = [] # ── Title 样式与事件 ────────────────────────────────────────────────── if title_enabled: title_color = _hex_to_ass_color(title_config.get("color", "#ffffff")) title_stroke = title_config.get("stroke", {}) or {} title_shadow = title_config.get("shadow", {}) or {} stroke_color = _hex_to_ass_color(title_stroke.get("color", "#000000")) stroke_width = float(title_stroke.get("width", 1)) if title_stroke.get("enabled", False) else 0.0 shadow_blur = float(title_shadow.get("blur", 4)) if title_shadow.get("enabled", False) else 0.0 shadow_offset = ( title_shadow.get("offset_x", 2) if title_shadow.get("enabled", False) else 0, title_shadow.get("offset_y", 2) if title_shadow.get("enabled", False) else 0, ) title_alignment = _position_to_ass_alignment(title_config.get("position", "top")) styles.append( _build_ass_style( "TitleStyle", font_name=title_config.get("font", "思源黑体"), font_size=int(title_config.get("size", 48)), primary_color=title_color, outline_color=stroke_color, outline_width=stroke_width, shadow_blur=shadow_blur, shadow_offset=shadow_offset, bold=bool(title_config.get("bold", True)), italic=bool(title_config.get("italic", False)), alignment=title_alignment, margin_v=TITLE_MARGIN_TOP, margin_l=TITLE_MARGIN_SIDE, margin_r=TITLE_MARGIN_SIDE, ) ) # 转义 ASS 特殊字符 safe_title_text = _escape_ass_text(title_text) events.append( "Dialogue: 0,0:00:00.00," f"{_format_ass_time(video_duration)}," "TitleStyle,,0,0,0,," f"{safe_title_text}" ) # ── Subtitle 样式与事件 ─────────────────────────────────────────────── if subtitle_enabled: sub_color = _hex_to_ass_color(subtitle_config.get("color", "#ffffff")) sub_alignment = _position_to_ass_alignment(subtitle_config.get("position", "bottom")) styles.append( _build_ass_style( "SubtitleStyle", font_name=subtitle_config.get("font", "思源黑体"), font_size=int(subtitle_config.get("size", 24)), primary_color=sub_color, outline_color="&H00000000", outline_width=1.0, shadow_blur=0.0, shadow_offset=(0, 0), bold=False, italic=False, alignment=sub_alignment, margin_v=TITLE_MARGIN_BOTTOM, margin_l=TITLE_MARGIN_SIDE, margin_r=TITLE_MARGIN_SIDE, ) ) safe_subtitle_text = _escape_ass_text(subtitle_text) events.append( "Dialogue: 0,0:00:00.00," f"{_format_ass_time(video_duration)}," "SubtitleStyle,,0,0,0,," f"{safe_subtitle_text}" ) # ── 组装 ASS 文件 ───────────────────────────────────────────────────── ass_content = f"""[Script Info] ScriptType: v4.00+ PlayResX: {video_width} PlayResY: {video_height} ScaledBorderAndShadow: yes WrapStyle: 2 Encoding: UTF-8 [V4+ Styles] Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding {chr(10).join(styles)} [Events] Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text {chr(10).join(events)} """ output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text(ass_content, encoding="utf-8") return output_path def _escape_ass_text(text: str) -> str: r"""转义 ASS 文本中的特殊字符。 ASS 中换行用 \N（硬换行）或 \n（软换行）， 大括号 {} 用于覆盖样式，需要转义。 """ # 将实际换行转为 ASS 硬换行 text = text.replace("\r\n", "\\N").replace("\n", "\\N").replace("\r", "\\N") # 转义大括号（ASS 用它做样式覆盖标签） text = text.replace("{", "(").replace("}", ")") return text def _format_ass_time(seconds: float) -> str: """将秒数格式化为 ASS 时间格式 H:MM:SS.cc。""" hours = int(seconds // 3600) minutes = int((seconds % 3600) // 60) secs = seconds % 60 return f"{hours}:{minutes:02d}:{secs:05.2f}" def _resolve_layer_role(clip_type: str, config: dict[str, Any]) -> 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 (default) → "main" """ 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" return "main" # ── 图层默认 z_index ───────────────────────────────────────────────────────── _LAYER_Z_INDEX: dict[str, int] = { "background": -1, "broll": 0, "main": 0, "overlay": 1, "corner_voice": 1, "audio": 2, } # 图层默认 PiP 位置（相对输出画布的偏移） _PIP_SCALE = 0.25 # PiP 占主画面的比例 # ── 统一渲染引擎 ───────────────────────────────────────────────────────────── class UnifiedRenderService: """统一渲染引擎。 输入 EditPlan + EditPlanClips + 素材路径映射，按时间线+图层执行渲染。 """ def __init__( self, plan: Any, # EditPlan clips: list[Any], # list[EditPlanClip] asset_path_map: dict[str, Path], # asset_id → local_path work_dir: Path, *, output_width: int = DEFAULT_OUTPUT_WIDTH, output_height: int = DEFAULT_OUTPUT_HEIGHT, output_fps: int = DEFAULT_FPS, transition_duration: float = DEFAULT_TRANSITION_DURATION, ): self.plan = plan self.clips = clips self.asset_path_map = asset_path_map self.work_dir = work_dir self.output_width = output_width self.output_height = output_height self.output_fps = output_fps self.transition_duration = transition_duration def render(self) -> RenderResult: """执行渲染，返回 RenderResult. 优化路径： - 单图层单 clip → 直通模式（-vf），性能最优 - 其他情况 → 完整 filter_complex 渲染 字幕渲染流程： 1. 视频主渲染（直通或完整链路） 2. 如有 title/subtitle，叠加 ASS 字幕 Raises: ValueError: 没有可渲染的片段时抛出 """ # 1. 解析 clips → ResolvedClips（跳过无素材的 clip） resolved = self._resolve_clips() if not resolved: raise ValueError("没有可渲染的片段（所有片段素材缺失或下载失败）") # 2. 分组为 RenderLayers layers = self._group_clips_into_layers(resolved) # 3. 计算视频总时长（用于字幕显示时长） video_duration = self._estimate_total_duration(layers) # 4. 生成 ASS 字幕文件（如果有 title/subtitle 配置） ass_path = self._maybe_generate_ass(video_duration) output_path = self.work_dir / f"rendered_{self.plan.id}.mp4" # 5. 视频主渲染 if self._can_use_pass_through(layers): self._render_pass_through(layers, output_path, ass_path=ass_path) else: filter_complex, input_args = self._build_filter_complex(layers, ass_path=ass_path) self._execute_ffmpeg(filter_complex, input_args, output_path) # 6. 探测输出 duration, file_size, width, height = self._probe_output(output_path) return RenderResult( output_path=output_path, duration=duration, file_size=file_size, width=width, height=height, ) def _estimate_total_duration(self, layers: list[RenderLayer]) -> float: """估算视频总时长（用于字幕等需要）。 取主图层（main/broll/background）的总时长，转场重叠按 transition_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( ( min(c.duration, c.actual_duration) if c.duration > 0 and c.actual_duration > 0 else (c.duration if c.duration > 0 else c.actual_duration) ) 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) def _maybe_generate_ass(self, video_duration: float) -> Path | None: """根据 plan.config 生成 ASS 字幕文件。 Returns: ASS 文件路径，没有字幕时返回 None """ config = self.plan.config or {} title_cfg = config.get("title", {}) or {} subtitle_cfg = config.get("subtitle", {}) or {} title_enabled = title_cfg.get("enabled", True) subtitle_enabled = subtitle_cfg.get("enabled", True) title_text = title_cfg.get("text", "") or "" subtitle_text = subtitle_cfg.get("text", "") or "" has_title = title_enabled and bool(title_text.strip()) has_subtitle = subtitle_enabled and bool(subtitle_text.strip()) if not has_title and not has_subtitle: return None ass_path = self.work_dir / f"subtitles_{self.plan.id}.ass" generate_ass_subtitles( ass_path, video_width=self.output_width, video_height=self.output_height, video_duration=video_duration, title_text=title_text, title_config=title_cfg, subtitle_text=subtitle_text, subtitle_config=subtitle_cfg, ) logger.info( "生成字幕: plan_id=%s title=%s subtitle=%s ass=%s", self.plan.id, has_title, has_subtitle, ass_path, ) return ass_path def _can_use_pass_through(self, layers: list[RenderLayer]) -> bool: """判断是否可以走直通优化路径。 条件： 1. 只有 1 个图层 2. 该图层是视频图层（main/broll/background），不是 overlay/corner_voice/audio 3. 该图层只有 1 个 clip（无转场需求） """ if len(layers) != 1: return False layer = layers[0] if layer.role not in ("main", "broll", "background"): return False if len(layer.clips) != 1: return False return True def _render_pass_through( self, layers: list[RenderLayer], output_path: Path, *, ass_path: Path | None = None ) -> None: """单图层单 clip 直通渲染（使用 -vf 而非 -filter_complex）。 性能优化：避免 filter_complex 的解析和调度开销， 对于一镜到底场景性能提升 ~30%，接近链路A水平。 Args: layers: 图层列表（只有1个图层1个clip） output_path: 输出文件路径 ass_path: ASS 字幕文件路径，有则叠加字幕 """ clip = layers[0].clips[0] role = layers[0].role # 构建滤镜链（与 _build_filter_complex 中预处理逻辑一致） filters: list[str] = [] # trim effective_duration = 0.0 if clip.duration > 0: effective_duration = min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration elif clip.actual_duration > 0: effective_duration = clip.actual_duration if effective_duration > 0: filters.append(f"trim=duration={effective_duration}") filters.append("setpts=PTS-STARTPTS") # scale + crop（铺满裁剪） if role in ("overlay", "corner_voice"): pip_w = int(self.output_width * _PIP_SCALE) pip_h = int(self.output_height * _PIP_SCALE) filters.append(f"scale={pip_w}:{pip_h}") else: # main / broll / background: 铺满裁剪 filters.append(f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=increase") filters.append(f"crop={self.output_width}:{self.output_height}") filters.append("setpts=PTS-STARTPTS") filters.append(f"fps={self.output_fps}") filters.append("format=yuv420p") # 字幕叠加 if ass_path is not None: # ASS 文件路径需要转义：Windows 反斜杠转正斜杠，冒号转义 ass_filter_path = str(ass_path).replace("\\", "/").replace(":", "\\:") filters.append(f"subtitles='{ass_filter_path}'") vf_str = ",".join(filters) command = [ FFMPEG_BIN, "-y", "-i", str(clip.local_path), "-vf", vf_str, "-c:v", "libx264", "-crf", "23", "-preset", "medium", "-pix_fmt", "yuv420p", "-movflags", "+faststart", "-an", # 直通模式暂不处理音频，音频统一在后续混音阶段处理 str(output_path), ] logger.info( "直通渲染: plan_id=%s clip=%s role=%s duration=%.2fs", self.plan.id, clip.clip_id, role, effective_duration, ) try: run_ffmpeg(command) except subprocess.CalledProcessError as e: logger.error( "直通渲染失败: plan_id=%s clip=%s exit_code=%d\nvf=%s", self.plan.id, clip.clip_id, e.returncode, vf_str[:2000], ) raise # ── 内部方法 ────────────────────────────────────────────────────────────── def _resolve_clips(self) -> list[ResolvedClip]: """将 EditPlanClip 列表解析为 ResolvedClip 列表。 跳过 asset_id 为空或在 asset_path_map 中找不到的片段。 """ resolved: list[ResolvedClip] = [] for clip in self.clips: asset_id = clip.asset_id if not asset_id: logger.warning("片段无素材: clip_id=%s", clip.id) continue local_path = self.asset_path_map.get(asset_id) if local_path is None or not local_path.exists(): logger.warning("素材不存在: clip_id=%s asset_id=%s", clip.id, asset_id) continue # 探测实际时长 try: actual_duration = probe_duration(local_path) except Exception: actual_duration = clip.duration or 5.0 rc = ResolvedClip( clip_id=clip.id, asset_id=asset_id, local_path=local_path, clip_type=clip.clip_type, order=clip.order, start_time=clip.start_time, duration=clip.duration, transition_effect=clip.transition_effect or "cut", config=clip.config or {}, actual_duration=actual_duration, ) resolved.append(rc) # 按 order 排序 resolved.sort(key=lambda c: c.order) return resolved def _group_clips_into_layers(self, resolved_clips: list[ResolvedClip]) -> list[RenderLayer]: """将 ResolvedClips 分组为 RenderLayers。 分组规则见 _resolve_layer_role 函数文档。 """ layer_map: dict[str, RenderLayer] = {} for clip in resolved_clips: role = _resolve_layer_role(clip.clip_type, clip.config) if role not in layer_map: z = _LAYER_Z_INDEX.get(role, 0) layer_map[role] = RenderLayer(role=role, z_index=z) layer_map[role].clips.append(clip) # 每个 layer 内的 clips 按 order 排序 for layer in layer_map.values(): layer.clips.sort(key=lambda c: c.order) # 计算 PiP 位置 pip_width = int(self.output_width * _PIP_SCALE) pip_height = int(self.output_height * _PIP_SCALE) margin = 20 # 边距 if "overlay" in layer_map: layer_map["overlay"].position = ( self.output_width - pip_width - margin, margin, ) if "corner_voice" in layer_map: layer_map["corner_voice"].position = ( self.output_width - pip_width - margin, margin, ) # 按 z_index 排序返回 layers = sorted(layer_map.values(), key=lambda lyr: lyr.z_index) return layers def _build_filter_complex( self, layers: list[RenderLayer], *, ass_path: Path | None = None ) -> tuple[str, list[str]]: """构建 FFmpeg filter_complex 字符串和输入参数列表。 Args: layers: 图层列表 ass_path: ASS 字幕文件路径，有则在最后叠加字幕 Returns: (filter_complex_str, input_args_list) input_args_list 是 ["-i", path1, "-i", path2, ...] 格式 """ if not layers: raise ValueError("没有可渲染的图层") # 收集所有 clips（按图层顺序，同层按 order） all_clips: list[ResolvedClip] = [] for layer in layers: all_clips.extend(layer.clips) # 构建输入参数 input_args: list[str] = [] clip_to_input_idx: dict[str, int] = {} for i, clip in enumerate(all_clips): input_args.extend(["-i", str(clip.local_path)]) clip_to_input_idx[clip.clip_id] = i filter_parts: list[str] = [] # Step 1: 预处理每个 clip — scale + setpts # 为每个 clip 生成预处理后的标签 [v0], [v1], ... preprocessed_labels: list[str] = [] for i, clip in enumerate(all_clips): label = f"v{i}" role = _resolve_layer_role(clip.clip_type, clip.config) filters: list[str] = [] # trim — 始终将输出截断到有效时长，防止 xfade offset 与实际时长不匹配 # 有效时长 = min(指定时长, 实际时长)；若均未设置则跳过 effective_duration = 0.0 if clip.duration > 0: effective_duration = ( min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration ) elif clip.actual_duration > 0: effective_duration = clip.actual_duration if effective_duration > 0: filters.append(f"trim=duration={effective_duration}") filters.append("setpts=PTS-STARTPTS") # scale if role in ("overlay", "corner_voice"): pip_w = int(self.output_width * _PIP_SCALE) pip_h = int(self.output_height * _PIP_SCALE) filters.append(f"scale={pip_w}:{pip_h}") elif role == "background": filters.append( f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=increase" ) filters.append(f"crop={self.output_width}:{self.output_height}") else: # main / broll: 铺满裁剪（scale to cover + center crop） # 对齐链路A编辑器合成行为，与主流短视频平台一致 filters.append( f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=increase" ) filters.append(f"crop={self.output_width}:{self.output_height}") filters.append("setpts=PTS-STARTPTS") filters.append(f"fps={self.output_fps}") filter_str = f"[{i}:v]{','.join(filters)}[{label}]" filter_parts.append(filter_str) preprocessed_labels.append(label) # Step 2: 同层 clips 用 xfade 串联 layer_output_labels: dict[str, str] = {} for layer in layers: layer_clip_indices = [all_clips.index(c) for c in layer.clips] layer_labels = [preprocessed_labels[i] for i in layer_clip_indices] # 使用 trim 后的有效时长，与 Step 1 的 trim=duration 保持一致 layer_durations = [] for i in layer_clip_indices: c = all_clips[i] if c.duration > 0: eff = min(c.duration, c.actual_duration) if c.actual_duration > 0 else c.duration else: eff = c.actual_duration if c.actual_duration > 0 else 0.0 layer_durations.append(eff) layer_transitions = [all_clips[i].transition_effect for i in layer_clip_indices] if len(layer_labels) == 1: # 单 clip 层，直接使用预处理标签 layer_output_labels[layer.role] = layer_labels[0] else: # 多 clip 层，用 xfade 串联 out_label = f"{layer.role}_merged" xfade_filter, _ = build_xfade_filter_chain( clip_durations=layer_durations, clip_video_labels=layer_labels, transitions=layer_transitions, transition_duration=self.transition_duration, output_label=out_label, ) if xfade_filter: filter_parts.append(xfade_filter) layer_output_labels[layer.role] = out_label # Step 3: 合成各层 # 找到主层 — background 优先作为底图，其次 broll / main final_video_label = None if "background" in layer_output_labels: final_video_label = layer_output_labels["background"] # b_roll / main 叠加到 background 上 for role in ("broll", "main"): if role in layer_output_labels: base_label = layer_output_labels[role] combined_label = f"combined_{role}" filter_parts.append( f"[{final_video_label}][{base_label}]" f"overlay=(W-w)/2:(H-h)/2[{combined_label}]" ) final_video_label = combined_label else: # 无 background 时，取 broll 或 main 作为基础 for role in ("broll", "main"): if role in layer_output_labels: final_video_label = layer_output_labels[role] break if final_video_label is None: # 没有任何主层，使用第一个层 final_video_label = layer_output_labels[layers[0].role] # 叠加 overlay 层 for layer in layers: if layer.role in ("overlay", "corner_voice"): if layer.role not in layer_output_labels: continue overlay_label = layer_output_labels[layer.role] x, y = layer.position or ( self.output_width - int(self.output_width * _PIP_SCALE) - 20, 20, ) combined_label = f"combined_{layer.role}" filter_parts.append(f"[{final_video_label}][{overlay_label}]" f"overlay={x}:{y}[{combined_label}]") final_video_label = combined_label # 叠加字幕（如有）+ 最终像素格式 if ass_path is not None: ass_filter_path = str(ass_path).replace("\\", "/").replace(":", "\\:") filter_parts.append(f"[{final_video_label}]subtitles='{ass_filter_path}',format=yuv420p[final_video]") else: filter_parts.append(f"[{final_video_label}]format=yuv420p[final_video]") filter_complex = ";".join(filter_parts) return filter_complex, input_args def _execute_ffmpeg( self, filter_complex: str, input_args: list[str], output_path: Path, ) -> None: """执行 FFmpeg 渲染命令。 失败时记录完整 filter_complex 以便排查（如 exit code 183）。 """ command = [ FFMPEG_BIN, "-y", *input_args, "-filter_complex", filter_complex, "-map", "[final_video]", "-c:v", "libx264", "-crf", "23", "-preset", "medium", "-pix_fmt", "yuv420p", "-movflags", "+faststart", str(output_path), ] logger.info( "执行渲染: plan_id=%s inputs=%d output=%s", self.plan.id, input_args.count("-i"), output_path, ) try: run_ffmpeg(command) except subprocess.CalledProcessError as e: # 额外记录 filter_complex，方便排查滤镜链构建问题 logger.error( "渲染失败: plan_id=%s exit_code=%d\nfilter_complex:\n%s", self.plan.id, e.returncode, filter_complex[:5000], ) raise def _probe_output(self, output_path: Path) -> tuple[float, int, int, int]: """探测输出文件的时长、大小、宽高。 Returns: (duration, file_size, width, height) """ info = probe_video_info(str(output_path)) file_size = output_path.stat().st_size if output_path.exists() else 0 return ( info["duration"], file_size, info["width"], info["height"], )