"""统一渲染引擎 — 输入 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 import time 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" if role == "audio": return "audio" 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 字幕 音频后处理： 1. 主图层音频 concat 拼接 2. 独立音频轨 amix 混入 3. 合并到输出视频 Raises: ValueError: 没有可渲染的片段时抛出 """ t_start = time.time() # 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) # 灰度埋点：开始渲染 layer_roles = [layer.role for layer in layers] clip_counts = {layer.role: len(layer.clips) for layer in layers} logger.info( "[unified-render] start render: plan_id=%s clip_count=%d layers=%s clip_counts=%s", self.plan.id, len(resolved), layer_roles, clip_counts, ) # 5. 视频主渲染 t_video_start = time.time() video_only_path = self.work_dir / f"rendered_{self.plan.id}_video.mp4" output_path = self.work_dir / f"rendered_{self.plan.id}.mp4" is_pass_through = self._can_use_pass_through(layers) pass_through_has_audio = False if is_pass_through: # 直通优化：单clip场景一次FFmpeg同时处理视频+音频，省去提取+合并两次调用 pass_through_has_audio = self._render_pass_through( layers, output_path, ass_path=ass_path, video_duration=video_duration ) else: filter_complex, input_args = self._build_filter_complex(layers, ass_path=ass_path) self._execute_ffmpeg(filter_complex, input_args, video_only_path) t_video_end = time.time() video_render_ms = int((t_video_end - t_video_start) * 1000) logger.info( "[unified-render] video render done: plan_id=%s duration_ms=%d pass_through=%s", self.plan.id, video_render_ms, is_pass_through, ) # 6. 音频后处理混音（直通场景已合并处理，跳过） t_audio_start = time.time() audio_mix_ms = 0 has_audio = False if is_pass_through: # 直通场景已在一次调用中完成视频+音频 has_audio = pass_through_has_audio else: audio_path = self._mix_audio(layers, video_duration) t_audio_end = time.time() audio_mix_ms = int((t_audio_end - t_audio_start) * 1000) has_audio = audio_path is not None if has_audio: logger.info( "[unified-render] audio mix done: plan_id=%s duration_ms=%d", self.plan.id, audio_mix_ms, ) # 7. 合并音视频 self._merge_audio_video(video_only_path, audio_path, output_path) else: # 无音频，直接用无声视频 import shutil shutil.copy2(video_only_path, output_path) # 8. 探测输出 duration, file_size, width, height = self._probe_output(output_path) t_total = int((time.time() - t_start) * 1000) logger.info( "[unified-render] render done: plan_id=%s total_ms=%d video_ms=%d audio_ms=%d " "output_duration=%.2fs output_size=%d resolution=%dx%d has_audio=%s", self.plan.id, t_total, video_render_ms, audio_mix_ms if has_audio else 0, duration, file_size, width, height, has_audio, ) 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(UnifiedRenderService._clip_effective_duration(c) 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, video_duration: float = 0.0, ) -> bool: """单图层单 clip 直通渲染（使用 -vf 而非 -filter_complex），一次性输出带音频的最终视频。 性能优化： - 避免 filter_complex 的解析和调度开销，单clip场景性能提升 ~30% - 视频+音频一次FFmpeg调用完成，省去后续音频提取+音视频合并两次调用 Args: layers: 图层列表（只有1个图层1个clip） output_path: 输出文件路径 ass_path: ASS 字幕文件路径，有则叠加字幕 video_duration: 视频总时长（用于截断音频，0表示不额外截断） Returns: True 表示输出包含音频（近似判断，实际以输出文件为准） """ clip = layers[0].clips[0] role = layers[0].role # 构建视频滤镜链（与 _build_filter_complex 中预处理逻辑一致） filters: list[str] = [] # trim effective_duration = UnifiedRenderService._clip_effective_duration(clip) 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_filter_path = str(ass_path).replace("\\", "/").replace(":", "\\:") filters.append(f"subtitles='{ass_filter_path}'") vf_str = ",".join(filters) # 最终输出时长：取 clip 有效时长和 video_duration 的较小值 final_duration = effective_duration if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration): final_duration = video_duration command = [ FFMPEG_BIN, "-y", "-i", str(clip.local_path), "-vf", vf_str, "-c:v", "libx264", "-crf", "23", "-preset", "medium", "-pix_fmt", "yuv420p", "-movflags", "+faststart", ] # 音频处理：background 通常是图片无音频，跳过；其他编码为 aac # background 以外的视频素材，默认带音频 has_audio = role != "background" if has_audio: command.extend(["-c:a", "aac", "-b:a", "128k"]) # 统一截断时长（同时作用于视频和音频） if final_duration > 0: command.extend(["-t", f"{final_duration:.3f}"]) command.append(str(output_path)) logger.info( "直通渲染: plan_id=%s clip=%s role=%s duration=%.2fs has_audio=%s", self.plan.id, clip.clip_id, role, effective_duration, has_audio, ) 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 return has_audio # ── 内部方法 ────────────────────────────────────────────────────────────── 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 与实际时长不匹配 effective_duration = UnifiedRenderService._clip_effective_duration(clip) 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 = [UnifiedRenderService._clip_effective_duration(all_clips[i]) for i in layer_clip_indices] 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"], ) # ── 音频后处理 ──────────────────────────────────────────────────────── def _mix_audio(self, layers: list[RenderLayer], video_duration: float) -> Path | None: """音频后处理混音. 处理逻辑： 1. 主音频源按优先级查找：main > broll（background 不参与主音频，通常是图片无音轨） 2. 主图层音频按顺序 concat 拼接 3. 独立音频轨（audio role）用 amix 混入 4. 输出时长截断到 video_duration 5. 无音频流的 clip 会被自动跳过，避免 FFmpeg 引用 [i:a] 失败 Args: layers: 图层列表 video_duration: 视频总时长（用于截断音频） Returns: 混音后的音频文件路径，无音频时返回 None """ # 按优先级精确查找主音频图层：main > broll # background 不参与主音频（通常是静态图片，无音轨） layer_map = {layer.role: layer for layer in layers} main_layer = None for role in ("main", "broll"): if role in layer_map and layer_map[role].clips: main_layer = layer_map[role] break main_clips: list[ResolvedClip] = main_layer.clips if main_layer else [] # 没有主视频图层时兜底：检查 overlay/corner_voice 层是否有带音频的素材 if not main_clips: for role in ("overlay", "corner_voice"): if role in layer_map and layer_map[role].clips: main_clips = layer_map[role].clips break # 收集独立音频轨 audio_clips: list[ResolvedClip] = [] if "audio" in layer_map: audio_clips = layer_map["audio"].clips # ── 防御：过滤掉无音频流的 clip ── # 源视频可能没有音频流（如静音视频、纯图片转的视频），直接引用 [i:a] 会导致 FFmpeg 失败 main_clips = [c for c in main_clips if self._clip_has_audio(c)] audio_clips = [c for c in audio_clips if self._clip_has_audio(c)] if not main_clips and not audio_clips: return None # 构建音频处理命令 output_path = self.work_dir / f"audio_{self.plan.id}.aac" # 简单场景：只有主图层 + 无独立音频 → 直接从视频提取音频并拼接 if main_clips and not audio_clips: self._concat_main_audio(main_clips, output_path, video_duration) return output_path # 有独立音频轨 → amix 混音 self._mix_with_independent_audio(main_clips, audio_clips, output_path, video_duration) return output_path def _concat_main_audio(self, clips: list[ResolvedClip], output_path: Path, video_duration: float) -> None: """主图层音频 concat 拼接（对齐链路A行为）. 每个 clip 提取音频 → trim → 按顺序 concat。 """ if len(clips) == 1: # 单 clip，直接提取音频，截断到 min(clip有效时长, 视频总时长) clip = clips[0] effective_duration = self._clip_effective_duration(clip) # 最终时长：取 clip 有效时长和视频总时长的较小值 # （视频总时长由主图层决定，但单 clip 场景下两者应该一致，仍做保护） final_duration = effective_duration if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration): final_duration = video_duration command = [ FFMPEG_BIN, "-y", "-i", str(clip.local_path), "-vn", "-acodec", "aac", "-b:a", "128k", ] if final_duration > 0: command.extend(["-t", f"{final_duration:.3f}"]) command.append(str(output_path)) run_ffmpeg(command) return # 多 clip，用 filter_complex concat input_args: list[str] = [] filter_parts: list[str] = [] for i, clip in enumerate(clips): input_args.extend(["-i", str(clip.local_path)]) effective_duration = self._clip_effective_duration(clip) if effective_duration > 0: filter_parts.append(f"[{i}:a]atrim=0:{effective_duration:.3f},asetpts=PTS-STARTPTS[a{i}]") else: filter_parts.append(f"[{i}:a]asetpts=PTS-STARTPTS[a{i}]") audio_labels = "".join(f"[a{i}]" for i in range(len(clips))) filter_parts.append(f"{audio_labels}concat=n={len(clips)}:v=0:a=1[outa]") # 截断到视频总时长 if video_duration > 0: filter_parts.append(f"[outa]atrim=0:{video_duration:.3f}[final_audio]") final_label = "final_audio" else: final_label = "outa" filter_complex = ";".join(filter_parts) command = [ FFMPEG_BIN, "-y", *input_args, "-filter_complex", filter_complex, "-map", f"[{final_label}]", "-acodec", "aac", "-b:a", "128k", str(output_path), ] run_ffmpeg(command) def _mix_with_independent_audio( self, main_clips: list[ResolvedClip], audio_clips: list[ResolvedClip], output_path: Path, video_duration: float, ) -> None: """主音频 + 独立音频轨 amix 混音. Args: main_clips: 主视频 clips（提取音频后 concat） audio_clips: 独立音频轨 clips output_path: 输出路径 video_duration: 视频总时长 """ input_args: list[str] = [] filter_parts: list[str] = [] mix_labels: list[str] = [] input_idx = 0 # 1. 主图层音频 concat if main_clips: for clip in main_clips: input_args.extend(["-i", str(clip.local_path)]) effective_duration = self._clip_effective_duration(clip) if effective_duration > 0: filter_parts.append( f"[{input_idx}:a]atrim=0:{effective_duration:.3f},asetpts=PTS-STARTPTS[ma{input_idx}]" ) else: filter_parts.append(f"[{input_idx}:a]asetpts=PTS-STARTPTS[ma{input_idx}]") input_idx += 1 if len(main_clips) == 1: mix_labels.append("ma0") else: main_labels = "".join(f"[ma{i}]" for i in range(len(main_clips))) filter_parts.append(f"{main_labels}concat=n={len(main_clips)}:v=0:a=1[main_audio]") mix_labels.append("main_audio") # 2. 独立音频轨 for j, clip in enumerate(audio_clips): input_args.extend(["-i", str(clip.local_path)]) effective_duration = self._clip_effective_duration(clip) volume = clip.config.get("volume", 1.0) if clip.config else 1.0 label = f"ia{j}" filters = [] if effective_duration > 0: filters.append(f"atrim=0:{effective_duration:.3f}") filters.append("asetpts=PTS-STARTPTS") if volume != 1.0: filters.append(f"volume={volume}") filter_parts.append(f"[{input_idx}:a]{','.join(filters)}[{label}]") mix_labels.append(label) input_idx += 1 # 3. amix 混音 mix_inputs = "".join(f"[{label}]" for label in mix_labels) n_inputs = len(mix_labels) # normalized=0 保持音量，duration=shortest 取最短 filter_parts.append(f"{mix_inputs}amix=inputs={n_inputs}:duration=longest:normalize=0[mixed_audio]") # 4. 截断到视频时长 if video_duration > 0: filter_parts.append(f"[mixed_audio]atrim=0:{video_duration:.3f}[final_audio]") final_label = "final_audio" else: final_label = "mixed_audio" filter_complex = ";".join(filter_parts) command = [ FFMPEG_BIN, "-y", *input_args, "-filter_complex", filter_complex, "-map", f"[{final_label}]", "-acodec", "aac", "-b:a", "128k", str(output_path), ] logger.info( "音频混音: plan_id=%s main_clips=%d audio_clips=%d", self.plan.id, len(main_clips), len(audio_clips), ) try: run_ffmpeg(command) except subprocess.CalledProcessError as e: logger.error( "音频混音失败: plan_id=%s exit_code=%d\nfilter_complex:\n%s", self.plan.id, e.returncode, filter_complex[:3000], ) raise def _merge_audio_video(self, video_path: Path, audio_path: Path, output_path: Path) -> None: """将音频合并到视频中（视频流拷贝，音频直接复用）. Args: video_path: 无声视频路径 audio_path: 音频文件路径 output_path: 输出文件路径 """ command = [ FFMPEG_BIN, "-y", "-i", str(video_path), "-i", str(audio_path), "-c:v", "copy", "-c:a", "aac", "-b:a", "128k", "-map", "0:v:0", "-map", "1:a:0", "-shortest", "-movflags", "+faststart", str(output_path), ] logger.info("合并音视频: plan_id=%s", self.plan.id) try: run_ffmpeg(command) except subprocess.CalledProcessError as e: logger.error( "合并音视频失败: plan_id=%s exit_code=%d", self.plan.id, e.returncode, ) raise @staticmethod def _clip_effective_duration(clip: ResolvedClip) -> float: """计算 clip 的有效时长.""" if clip.duration > 0: return min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration return clip.actual_duration if clip.actual_duration > 0 else 0.0 def _clip_has_audio(self, clip: ResolvedClip) -> bool: """探测 clip 是否有音频流（带缓存）. 避免同一个 clip 被多次 ffprobe 探测。 """ if not hasattr(self, "_audio_cache"): self._audio_cache: dict[str, bool] = {} key = str(clip.local_path) if key not in self._audio_cache: from .ffmpeg_utils import probe_has_audio self._audio_cache[key] = probe_has_audio(clip.local_path) return self._audio_cache[key]