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

feat(unified-render): Phase 3 - 音频统一混音 + 灰度观测埋点

Browse Source 
- 音频后处理：主图层concat拼接 + 独立音频轨amix混音
- 音视频分离：视频先渲染无声版，音频后处理后合并
- 支持音量调节（audio clip config.volume）
- 灰度埋点：渲染开始/视频渲染完成/音频混音完成/总完成 四个阶段日志
- 新增11个音频混音单元测试
- _resolve_layer_role 支持 audio role 映射
This commit is contained in:
灵应
2026-07-12 19:40:57 +08:00
committed by CI Bot
parent 9e87ac05c6
commit 3af353eca7
3 changed files with 588 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/worker/video_processing/render_adapter.py
+19 -2
View File
@@ -140,7 +140,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 +176,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 +200,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],
apps/worker/video_processing/unified_render_service.py
+344 -5
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,76 @@ class UnifiedRenderService:
# 4. 生成 ASS 字幕文件（如果有 title/subtitle 配置）
ass_path = self._maybe_generate_ass(video_duration)
# 灰度埋点：开始渲染
layer_roles = [l.role for l in layers]
clip_counts = {l.role: len(l.clips) for l 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"
# 5. 视频主渲染
if self._can_use_pass_through(layers):
self._render_pass_through(layers, output_path, ass_path=ass_path)
self._render_pass_through(layers, video_only_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)
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,
self._can_use_pass_through(layers),
)
# 6. 音频后处理混音
t_audio_start = time.time()
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,
@@ -930,7 +998,7 @@ class UnifiedRenderService:
raise
def _probe_output(self, output_path: Path) -> tuple[float, int, int, int]:
"""探测输出文件的时长、大小、宽高。
"""探测输出文件的时长、大小、宽高.
Returns:
(duration, file_size, width, height)
@@ -943,3 +1011,274 @@ class UnifiedRenderService:
info["width"],
info["height"],
)
# ── 音频后处理 ────────────────────────────────────────────────────────
def _mix_audio(self, layers: list[RenderLayer], video_duration: float) -> Path | None:
"""音频后处理混音.
处理逻辑：
1. 主图层（main/broll/background）音频按顺序 concat 拼接
2. 独立音频轨（audio role）用 amix 混入
3. 输出时长截断到 video_duration
Args:
layers: 图层列表
video_duration: 视频总时长（用于截断音频）
Returns:
混音后的音频文件路径，无音频时返回 None
"""
# 收集主图层的视频 clips（带音频）
main_clips: list[ResolvedClip] = []
for layer in layers:
if layer.role in ("main", "broll", "background"):
main_clips.extend(layer.clips)
break # 只取第一个主图层
if not main_clips:
# 没有主视频图层，检查其他带音频的图层
for layer in layers:
if layer.role in ("overlay", "corner_voice"):
main_clips.extend(layer.clips)
break
# 收集独立音频轨
audio_clips: list[ResolvedClip] = []
for layer in layers:
if layer.role == "audio":
audio_clips.extend(layer.clips)
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，直接提取音频
clip = clips[0]
effective_duration = self._clip_effective_duration(clip)
command = [
FFMPEG_BIN,
"-y",
"-i",
str(clip.local_path),
"-vn",
"-acodec",
"aac",
"-b:a",
"128k",
]
if effective_duration > 0:
command.extend(["-t", f"{effective_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"[{l}]" for l 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
tests/unit/test_unified_render_service.py
+225
View File
@@ -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,220 @@ 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_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):
"""完整render流程会调用音频混音。"""
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"),
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):
"""无音频时直接copy视频文件。"""
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"),
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