From fdcf48103e9b53c2460b59ace03bf0b2482a3d8a Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Sun, 12 Jul 2026 23:00:51 +0800 Subject: [PATCH] =?UTF-8?q?feat(unified-render):=20Phase=203=20-=20?= =?UTF-8?q?=E9=9F=B3=E9=A2=91=E7=BB=9F=E4=B8=80=E6=B7=B7=E9=9F=B3=20+=20?= =?UTF-8?q?=E7=81=B0=E5=BA=A6=E8=A7=82=E6=B5=8B=E5=9F=8B=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 3 unified render with audio mixing + gray scale observation metrics --- apps/worker/video_processing/ffmpeg_utils.py | 35 ++ .../worker/video_processing/render_adapter.py | 34 +- .../unified_render_service.py | 473 +++++++++++++-- tests/unit/test_unified_render_service.py | 546 ++++++++++++++++++ 4 files changed, 1031 insertions(+), 57 deletions(-) diff --git a/apps/worker/video_processing/ffmpeg_utils.py b/apps/worker/video_processing/ffmpeg_utils.py index 519e19832..6501dca76 100755 --- a/apps/worker/video_processing/ffmpeg_utils.py +++ b/apps/worker/video_processing/ffmpeg_utils.py @@ -85,6 +85,41 @@ def run_ffmpeg( raise +def probe_has_audio(local_path: str | Path) -> bool: + """探测文件是否包含音频流。 + + Args: + local_path: 本地文件路径 + + Returns: + True 表示有音频流（或探测失败保守返回），False 表示确认无音频流 + """ + try: + result = subprocess.run( # nosec B603 + [ + FFPROBE_BIN, + "-v", + "error", + "-select_streams", + "a:0", + "-show_entries", + "stream=codec_type", + "-of", + "default=noprint_wrappers=1:nokey=1", + str(local_path), + ], + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + timeout=10, + ) + return result.stdout.strip() == "audio" + except Exception: + # 探测失败保守返回 True，让 FFmpeg 自己处理（避免误删音频） + return True + + def probe_duration(local_path: str | Path) -> float: """用 ffprobe 获取视频时长（秒）。 diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py index 69d517ac3..2b09bc3db 100755 --- a/apps/worker/video_processing/render_adapter.py +++ b/apps/worker/video_processing/render_adapter.py @@ -21,17 +21,10 @@ from typing import Any, Callable from sqlalchemy.orm import Session from video_processing.oss_helpers import download_asset, upload_to_oss -from video_processing.unified_render_service import ( - RenderResult, - UnifiedRenderService, -) +from video_processing.unified_render_service import RenderResult, UnifiedRenderService -from packages.adapters.sqlalchemy_impl.edit_plan_clip_repository import ( - SQLAlchemyEditPlanClipRepository, -) -from packages.adapters.sqlalchemy_impl.edit_plan_repository import ( - SQLAlchemyEditPlanRepository, -) +from packages.adapters.sqlalchemy_impl.edit_plan_clip_repository import SQLAlchemyEditPlanClipRepository +from packages.adapters.sqlalchemy_impl.edit_plan_repository import SQLAlchemyEditPlanRepository from packages.domain.edit_plan import EditPlan, EditPlanStatus from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus @@ -140,7 +133,7 @@ class RenderAdapter: ) logger.info( - "开始渲染: plan_id=%s job_id=%s ready_clips=%d", + "开始渲染: plan_id=%s job_id=%s ready_clips=%d engine=unified", plan_id, job_id, len(ready_clips), @@ -176,6 +169,18 @@ class RenderAdapter: self._report_progress(progress_cb, 100.0, "渲染完成") + logger.info( + "[render-adapter] render success: plan_id=%s job_id=%s engine=unified " + "duration=%.2fs file_size=%d resolution=%dx%d clip_count=%d", + plan_id, + job_id, + result.duration, + result.file_size, + result.width, + result.height, + len(ready_clips), + ) + return RenderAdapterResult( success=True, output_url=output_url or "", @@ -188,7 +193,12 @@ class RenderAdapter: ) except Exception as exc: - logger.exception("渲染失败: plan_id=%s", plan_id) + logger.exception( + "[render-adapter] render failed: plan_id=%s job_id=%s engine=unified error=%s", + plan_id, + job_id, + str(exc)[:200], + ) return RenderAdapterResult( success=False, error_message=str(exc)[:500], diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index 8f2012b36..aba94148c 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -24,6 +24,7 @@ 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 @@ -356,6 +357,8 @@ def _resolve_layer_role(clip_type: str, config: dict[str, Any]) -> str: # main type if role == "b_roll": return "broll" + if role == "audio": + return "audio" return "main" @@ -415,9 +418,16 @@ class UnifiedRenderService: 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: @@ -432,18 +442,88 @@ class UnifiedRenderService: # 4. 生成 ASS 字幕文件（如果有 title/subtitle 配置） ass_path = self._maybe_generate_ass(video_duration) - output_path = self.work_dir / f"rendered_{self.plan.id}.mp4" + # 灰度埋点：开始渲染 + 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. 视频主渲染 - if self._can_use_pass_through(layers): - self._render_pass_through(layers, output_path, ass_path=ass_path) + 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, output_path) + self._execute_ffmpeg(filter_complex, input_args, video_only_path) - # 6. 探测输出 + 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, @@ -470,14 +550,7 @@ class UnifiedRenderService: 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 - ) + total = sum(UnifiedRenderService._clip_effective_duration(c) for c in main_layer.clips) # 减去转场重叠时间（粗略估算） n_clips = len(main_layer.clips) @@ -547,30 +620,36 @@ class UnifiedRenderService: return True def _render_pass_through( - self, layers: list[RenderLayer], output_path: Path, *, ass_path: Path | None = None - ) -> None: - """单图层单 clip 直通渲染（使用 -vf 而非 -filter_complex）。 + self, + layers: list[RenderLayer], + output_path: Path, + *, + ass_path: Path | None = None, + video_duration: float = 0.0, + ) -> bool: + """单图层单 clip 直通渲染（使用 -vf 而非 -filter_complex），一次性输出带音频的最终视频。 - 性能优化：避免 filter_complex 的解析和调度开销， - 对于一镜到底场景性能提升 ~30%，接近链路A水平。 + 性能优化： + - 避免 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 中预处理逻辑一致） + # 构建视频滤镜链（与 _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 + effective_duration = UnifiedRenderService._clip_effective_duration(clip) if effective_duration > 0: filters.append(f"trim=duration={effective_duration}") @@ -592,12 +671,16 @@ class UnifiedRenderService: # 字幕叠加 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) + # 最终输出时长：取 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", @@ -615,16 +698,27 @@ class UnifiedRenderService: "yuv420p", "-movflags", "+faststart", - "-an", # 直通模式暂不处理音频，音频统一在后续混音阶段处理 - str(output_path), ] + # 音频处理：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", + "直通渲染: 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) @@ -638,6 +732,8 @@ class UnifiedRenderService: ) raise + return has_audio + # ── 内部方法 ────────────────────────────────────────────────────────────── def _resolve_clips(self) -> list[ResolvedClip]: @@ -759,14 +855,7 @@ class UnifiedRenderService: 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 + effective_duration = UnifiedRenderService._clip_effective_duration(clip) if effective_duration > 0: filters.append(f"trim=duration={effective_duration}") @@ -803,14 +892,7 @@ class UnifiedRenderService: 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_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: @@ -930,7 +1012,7 @@ class UnifiedRenderService: raise def _probe_output(self, output_path: Path) -> tuple[float, int, int, int]: - """探测输出文件的时长、大小、宽高。 + """探测输出文件的时长、大小、宽高. Returns: (duration, file_size, width, height) @@ -943,3 +1025,304 @@ class UnifiedRenderService: 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] diff --git a/tests/unit/test_unified_render_service.py b/tests/unit/test_unified_render_service.py index 2ed2d4dad..c9102baa9 100755 --- a/tests/unit/test_unified_render_service.py +++ b/tests/unit/test_unified_render_service.py @@ -573,6 +573,8 @@ class TestPassThrough: patch("video_processing.unified_render_service.probe_duration", return_value=5.0), patch.object(svc, "_render_pass_through") as mock_pass, patch.object(svc, "_execute_ffmpeg") as mock_exec, + patch.object(svc, "_mix_audio", return_value=None), + patch("shutil.copy2"), patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)), ): result = svc.render() @@ -598,6 +600,8 @@ class TestPassThrough: patch("video_processing.unified_render_service.probe_duration", return_value=5.0), patch.object(svc, "_render_pass_through") as mock_pass, patch.object(svc, "_execute_ffmpeg") as mock_exec, + patch.object(svc, "_mix_audio", return_value=None), + patch("shutil.copy2"), patch.object(svc, "_probe_output", return_value=(5.5, 2048, 1280, 720)), ): result = svc.render() @@ -816,6 +820,8 @@ class TestRender: _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0), patch.object(svc, "_render_pass_through") as mock_pass, + patch.object(svc, "_mix_audio", return_value=None), + patch("shutil.copy2"), patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)), ): result = svc.render() @@ -843,6 +849,8 @@ class TestRender: _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0), patch.object(svc, "_execute_ffmpeg") as mock_exec, + patch.object(svc, "_mix_audio", return_value=None), + patch("shutil.copy2"), patch.object(svc, "_probe_output", return_value=(5.5, 2048, 1280, 720)), ): result = svc.render() @@ -853,3 +861,541 @@ class TestRender: assert result.width == 1280 assert result.height == 720 mock_exec.assert_called_once() + + +# ── 测试音频后处理 ────────────────────────────────────────────────────────── + + +class TestAudioMixing: + """测试音频后处理混音功能。""" + + def test_clip_effective_duration_with_both(self): + """指定时长和实际时长都有时取较小值。""" + clip = ResolvedClip( + clip_id="c1", + asset_id="a1", + local_path=Path("/tmp/c1.mp4"), + clip_type="main", + order=0, + duration=3.0, + actual_duration=5.0, + ) + assert UnifiedRenderService._clip_effective_duration(clip) == 3.0 + + def test_clip_effective_duration_only_actual(self): + """只有实际时长时用实际时长。""" + clip = ResolvedClip( + clip_id="c1", + asset_id="a1", + local_path=Path("/tmp/c1.mp4"), + clip_type="main", + order=0, + duration=0.0, + actual_duration=5.0, + ) + assert UnifiedRenderService._clip_effective_duration(clip) == 5.0 + + def test_clip_effective_duration_only_specified(self): + """只有指定时长时用指定时长。""" + clip = ResolvedClip( + clip_id="c1", + asset_id="a1", + local_path=Path("/tmp/c1.mp4"), + clip_type="main", + order=0, + duration=3.0, + actual_duration=0.0, + ) + assert UnifiedRenderService._clip_effective_duration(clip) == 3.0 + + def test_clip_effective_duration_zero(self): + """都没有时返回0。""" + clip = ResolvedClip( + clip_id="c1", + asset_id="a1", + local_path=Path("/tmp/c1.mp4"), + clip_type="main", + order=0, + duration=0.0, + actual_duration=0.0, + ) + assert UnifiedRenderService._clip_effective_duration(clip) == 0.0 + + def test_mix_audio_single_main_clip(self): + """单主clip时直接提取音频。""" + clips = [_make_clip("c1", "main", order=0, duration=5.0)] + asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")} + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + result = svc._mix_audio(layers, 5.0) + + assert result is not None + assert result.name == "audio_plan_001.aac" + mock_run.assert_called_once() + # 验证命令包含 -vn（无视频）和 aac 编码 + cmd = mock_run.call_args[0][0] + assert "-vn" in cmd + assert "aac" in cmd + + def test_mix_audio_multi_main_clips(self): + """多主clip时用concat拼接音频。""" + clips = [ + _make_clip("c1", "main", order=0, duration=3.0), + _make_clip("c2", "main", order=1, duration=2.0), + ] + asset_paths = { + "asset_c1.mp4": Path("/tmp/asset_c1.mp4"), + "asset_c2.mp4": Path("/tmp/asset_c2.mp4"), + } + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + result = svc._mix_audio(layers, 4.5) + + assert result is not None + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + # 验证有 filter_complex 和 concat + assert "-filter_complex" in cmd + cmd_str = " ".join(cmd) + assert "concat=n=2:v=0:a=1" in cmd_str + + def test_mix_audio_with_independent_audio_track(self): + """有独立音频轨时用amix混音。""" + clips = [ + _make_clip("c1", "main", order=0, duration=5.0), + _make_clip( + "bgm1", + "main", + order=0, + duration=5.0, + config={"role": "audio", "volume": 0.5}, + ), + ] + asset_paths = { + "asset_c1.mp4": Path("/tmp/asset_c1.mp4"), + "asset_bgm1.mp4": Path("/tmp/asset_bgm1.mp4"), + } + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + result = svc._mix_audio(layers, 5.0) + + assert result is not None + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + cmd_str = " ".join(cmd) + assert "amix" in cmd_str + assert "volume=0.5" in cmd_str + + def test_mix_audio_no_audio_returns_none(self): + """没有音频素材时返回None。""" + # 构造一个没有音频的场景（比如纯文字） + clips = [_make_clip("t1", "title", order=0, duration=3.0)] + clips[0].asset_id = "" # 无素材 + asset_paths: dict[str, Path] = {} + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=3.0), + ): + # 没有素材的clip会被跳过，layers为空 + resolved = svc._resolve_clips() + layers = svc._group_clips_into_layers(resolved) + result = svc._mix_audio(layers, 3.0) + + assert result is None + + def test_mix_audio_background_not_used_as_main(self): + """background 图层不参与主音频，main 优先级更高。""" + clips = [ + _make_clip("bg1", "background", order=0, duration=5.0), + _make_clip("c1", "main", order=0, duration=5.0), + ] + asset_paths = { + "asset_bg1.mp4": Path("/tmp/asset_bg1.mp4"), + "asset_c1.mp4": Path("/tmp/asset_c1.mp4"), + } + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + result = svc._mix_audio(layers, 5.0) + + assert result is not None + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + # 验证主音频源是 main 的 c1，不是 background 的 bg1 + # 单 main clip 走直接提取路径，输入文件应该只有 c1 + cmd_str = " ".join(cmd) + assert "asset_c1.mp4" in cmd_str + assert "asset_bg1.mp4" not in cmd_str + + def test_mix_audio_main_priority_over_broll(self): + """main 图层优先级高于 broll。""" + clips = [ + _make_clip("b1", "b_roll", order=0, duration=5.0), + _make_clip("c1", "main", order=0, duration=5.0), + ] + asset_paths = { + "asset_b1.mp4": Path("/tmp/asset_b1.mp4"), + "asset_c1.mp4": Path("/tmp/asset_c1.mp4"), + } + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + result = svc._mix_audio(layers, 5.0) + + assert result is not None + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + cmd_str = " ".join(cmd) + # 主音频源应该是 main 的 c1，不是 broll 的 b1 + assert "asset_c1.mp4" in cmd_str + assert "asset_b1.mp4" not in cmd_str + + def test_mix_audio_broll_used_when_no_main(self): + """没有 main 时，broll 作为主音频源。""" + clips = [_make_clip("b1", "b_roll", order=0, duration=5.0)] + asset_paths = {"asset_b1.mp4": Path("/tmp/asset_b1.mp4")} + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + result = svc._mix_audio(layers, 5.0) + + assert result is not None + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + assert "-vn" in cmd + assert "asset_b1.mp4" in " ".join(cmd) + + def test_mix_audio_single_clip_truncated_to_video_duration(self): + """单clip音频截断到 video_duration（video_duration < clip有效时长）。""" + clips = [_make_clip("c1", "main", order=0, duration=10.0)] + asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")} + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=10.0), + patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + # video_duration 只有 3.0，小于 clip 的 10.0 + result = svc._mix_audio(layers, 3.0) + + assert result is not None + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + # 验证 -t 参数是 3.0 不是 10.0 + t_index = cmd.index("-t") + assert t_index >= 0 + t_value = float(cmd[t_index + 1]) + assert t_value == 3.0 + + def test_merge_audio_video(self): + """合并音视频命令正确。""" + svc = _make_service([], {}) + video_path = Path("/tmp/video.mp4") + audio_path = Path("/tmp/audio.aac") + output_path = Path("/tmp/output.mp4") + + with patch("video_processing.unified_render_service.run_ffmpeg") as mock_run: + svc._merge_audio_video(video_path, audio_path, output_path) + + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + assert "-c:v" in cmd + assert "copy" in cmd + assert "-map" in cmd + assert "-shortest" in cmd + + def test_render_calls_audio_mixing(self): + """多clip完整render流程会调用音频混音（非直通路径）。""" + clips = [ + _make_clip("c1", "main", order=0, duration=3.0), + _make_clip("c2", "main", order=1, duration=2.0), + ] + asset_paths = { + "asset_c1.mp4": Path("/tmp/asset_c1.mp4"), + "asset_c2.mp4": Path("/tmp/asset_c2.mp4"), + } + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch.object(svc, "_execute_ffmpeg"), + patch.object(svc, "_mix_audio", return_value=Path("/tmp/audio.aac")) as mock_mix, + patch.object(svc, "_merge_audio_video") as mock_merge, + patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)), + ): + result = svc.render() + + mock_mix.assert_called_once() + mock_merge.assert_called_once() + assert result.duration == 5.0 + + def test_render_without_audio_copies_video(self): + """多clip无音频时走copy路径（非直通路径）。""" + clips = [ + _make_clip("c1", "main", order=0, duration=3.0), + _make_clip("c2", "main", order=1, duration=2.0), + ] + asset_paths = { + "asset_c1.mp4": Path("/tmp/asset_c1.mp4"), + "asset_c2.mp4": Path("/tmp/asset_c2.mp4"), + } + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch.object(svc, "_execute_ffmpeg"), + patch.object(svc, "_mix_audio", return_value=None), + patch("shutil.copy2") as mock_copy, + patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)), + ): + result = svc.render() + + mock_copy.assert_called_once() + assert result.duration == 5.0 + + def test_render_pass_through_skips_audio_mix(self): + """直通场景下视频+音频一次完成，跳过 _mix_audio 和 _merge_audio_video。""" + clips = [_make_clip("c1", "main", order=0, duration=5.0)] + asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")} + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch.object(svc, "_render_pass_through", return_value=True) as mock_pt, + patch.object(svc, "_mix_audio") as mock_mix, + patch.object(svc, "_merge_audio_video") as mock_merge, + patch("shutil.copy2") as mock_copy, + patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)), + ): + result = svc.render() + + # 直通场景调用了 _render_pass_through，跳过了 _mix_audio / _merge / copy + mock_pt.assert_called_once() + mock_mix.assert_not_called() + mock_merge.assert_not_called() + mock_copy.assert_not_called() + assert result.duration == 5.0 + + def test_pass_through_main_has_aac_audio(self): + """直通main/broll场景输出带aac音频。""" + clips = [_make_clip("c1", "main", order=0, duration=5.0)] + asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")} + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + result = svc._render_pass_through(layers, Path("/tmp/out.mp4"), video_duration=5.0) + + assert result is True # main 类型返回有音频 + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + assert "-an" not in cmd # 不再是无声 + assert "aac" in cmd # 有aac音频编码 + assert "-b:a" in cmd + + def test_pass_through_background_no_audio(self): + """直通background场景不带音频（图片素材）。""" + clips = [_make_clip("bg1", "background", order=0, duration=5.0)] + asset_paths = {"asset_bg1.mp4": Path("/tmp/asset_bg1.mp4")} + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + result = svc._render_pass_through(layers, Path("/tmp/out.mp4")) + + assert result is False # background 返回无音频 + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + assert "aac" not in cmd # 没有音频编码参数 + + # ── 无音轨视频防御测试 ── + + def test_mix_audio_main_no_audio_stream_returns_none(self): + """主图层clip无音频流且无独立音频轨时，返回None（不报错）。""" + clips = [_make_clip("c1", "main", order=0, duration=5.0)] + asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")} + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch("video_processing.ffmpeg_utils.probe_has_audio", return_value=False), + patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + result = svc._mix_audio(layers, 5.0) + + assert result is None + # 没有音频流时不应调用 FFmpeg + mock_run.assert_not_called() + + def test_mix_audio_partial_clips_no_audio_filtered(self): + """部分主图层clip无音频流时，过滤掉无音轨的，剩余有音频的正常concat。""" + clips = [ + _make_clip("c1", "main", order=0, duration=3.0), # 无音频 + _make_clip("c2", "main", order=1, duration=2.0), # 有音频 + ] + asset_paths = { + "asset_c1.mp4": Path("/tmp/asset_c1.mp4"), + "asset_c2.mp4": Path("/tmp/asset_c2.mp4"), + } + svc = _make_service(clips, asset_paths) + + # 模拟：c1 无音频，c2 有音频 + def fake_has_audio(path): + return "c2" in str(path) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch("video_processing.ffmpeg_utils.probe_has_audio", side_effect=fake_has_audio), + patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + result = svc._mix_audio(layers, 5.0) + + assert result is not None + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + cmd_str = " ".join(cmd) + # 只剩 1 个有效音频 clip，走单clip路径（-vn），不走 filter_complex concat + assert "-vn" in cmd + assert "concat=n=2" not in cmd_str + + def test_mix_audio_all_main_no_audio_but_independent_track(self): + """主图层全部无音频，但有独立音频轨时，正常走amix混音。""" + clips = [ + _make_clip("c1", "main", order=0, duration=5.0), # 无音频 + _make_clip( + "bgm1", + "main", + order=0, + duration=5.0, + config={"role": "audio", "volume": 0.5}, + ), # 独立音频轨（有音频） + ] + asset_paths = { + "asset_c1.mp4": Path("/tmp/asset_c1.mp4"), + "asset_bgm1.mp4": Path("/tmp/asset_bgm1.mp4"), + } + svc = _make_service(clips, asset_paths) + + def fake_has_audio(path): + return "bgm" in str(path) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch("video_processing.ffmpeg_utils.probe_has_audio", side_effect=fake_has_audio), + patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + result = svc._mix_audio(layers, 5.0) + + assert result is not None + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + cmd_str = " ".join(cmd) + # 只有独立音频轨参与混音，amix 输入数=1 + assert "amix=inputs=1" in cmd_str + + def test_mix_audio_both_no_audio_returns_none(self): + """主图层和独立音频轨都无音频时，返回None。""" + clips = [ + _make_clip("c1", "main", order=0, duration=5.0), + _make_clip( + "bgm1", + "main", + order=0, + duration=5.0, + config={"role": "audio"}, + ), + ] + asset_paths = { + "asset_c1.mp4": Path("/tmp/asset_c1.mp4"), + "asset_bgm1.mp4": Path("/tmp/asset_bgm1.mp4"), + } + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + patch("video_processing.ffmpeg_utils.probe_has_audio", return_value=False), + patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + ): + layers = svc._group_clips_into_layers(svc._resolve_clips()) + result = svc._mix_audio(layers, 5.0) + + assert result is None + mock_run.assert_not_called() + + def test_clip_has_audio_cache(self): + """_clip_has_audio 带缓存，同一clip只探测一次。""" + clips = [_make_clip("c1", "main", order=0, duration=5.0)] + asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")} + svc = _make_service(clips, asset_paths) + + with ( + _patch_path_exists(), + patch("video_processing.unified_render_service.probe_duration", return_value=5.0), + ): + resolved = svc._resolve_clips() + clip = resolved[0] + + with patch("video_processing.ffmpeg_utils.probe_has_audio", return_value=True) as mock_probe: + # 调用 3 次 + r1 = svc._clip_has_audio(clip) + r2 = svc._clip_has_audio(clip) + r3 = svc._clip_has_audio(clip) + + assert r1 is True and r2 is True and r3 is True + # 实际只探测了 1 次 + assert mock_probe.call_count == 1