xiaoxia/xiaoxia-saas
1
0
Fork 0
Code Issues 24 Pull Requests 4 Actions Packages Projects Releases 7 Wiki Activity

feat(unified-render): Phase 3 - 音频统一混音 + 灰度观测埋点
CI/CD Pipeline / Validate Code Quality And Tests (push) Successful in 1m27s
Details
CI/CD Pipeline / Frontend Lint (push) Successful in 2m4s
Details
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
Details
CI/CD Pipeline / Deploy Production (push) Has been skipped
Details
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
Details
CI/CD Pipeline / Integration Tests (push) Successful in 1m44s
Details
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Successful in 9m39s
Details
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
Details
CI/CD Pipeline / Staging API Integration Tests (push) Has been skipped
Details

Browse Source 
Phase 3 unified render with audio mixing + gray scale observation metrics
This commit was merged in pull request #232.
This commit is contained in:
xiaoxia
2026-07-12 23:00:51 +08:00
parent 9e87ac05c6
commit fdcf48103e
4 changed files with 1031 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/worker/video_processing/unified_render_service.py
+428 -45
View File
@@ -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]