fix(worker): render_plan 保留视频原声并尊重每clip音量
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 30s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m34s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m39s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m18s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 39s
AI Code Review / AI Code Review (pull_request) Failing after 2m14s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 32s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 2m35s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m17s
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 567h28m26s
CI/CD Pipeline / PR Build Web Image (pull_request) Failing after 567h28m32s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Failing after 567h29m13s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 567h30m47s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 567h30m49s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 567h28m28s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 567h29m15s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 567h30m51s
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 568h2m20s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 568h3m22s

根因:render_audio.mix_audio 中 main_clips = [] 无条件丢弃主图层源视频
原声(PR #1249 "避免录入杂音"),导致走新路径 render_plan 生成的成片
没有原声,与预览不一致(用户反馈"音频没有")。

修复:
- mix_audio 保留 main/broll 原声,按 clip.config.volume 控制音量,
  无音频流/静音(volume=0)的 clip 安全过滤
- concat_main_audio 单clip与多clip路径均应用 per-clip volume 滤镜
- 直通渲染(_render_pass_through)改为先 probe_has_audio 再决定是否编码
  音频,并应用 per-clip volume;不再无条件假设 main 带音频
- stream copy 在音量非默认时回退重编码
- 新增 _clip_volume 辅助函数(render_audio + UnifiedRenderService)
- 更新 test_unified_render_service 旧"丢弃原声"断言为新行为
- 新增 test_original_audio_retained.py 9 个回归测试

验证:161 个相关单测全部通过
This commit is contained in:
saas-backend-agent
2026-08-23 23:54:44 +08:00
parent bbf27ec3f6
commit c2a510331e
4 changed files with 357 additions and 94 deletions
+38 -10
View File
@@ -67,6 +67,16 @@ def clip_has_audio(ctx: RenderContext, clip: ResolvedClip) -> bool:
return ctx._audio_cache[key]
def _clip_volume(clip: ResolvedClip) -> float:
"""读取 clip 的音量配置(0.0~1.0,>1 放大)。缺省 1.0 原声。"""
cfg = getattr(clip, "config", None) or {}
try:
vol = float(cfg.get("volume", 1.0))
except (TypeError, ValueError):
return 1.0
return max(0.0, vol)
# ── 音频混音 ──────────────────────────────────────────────────────────────────
@@ -82,12 +92,13 @@ def mix_audio(
"""音频后处理混音.
处理逻辑:
1. 丢弃主图层(main/broll/overlay/corner_voice)的原始音频,避免录入源视频杂音
2. 仅使用独立音频轨(audio role,TTS/配音)作为主音频
3. 如果提供了 bgm_path,则额外混入 BGM(支持淡入淡出、循环、人声闪避)
4. 如果配置了 audio_tracks,则混入多轨道音频(配音、音效等)
5. 输出时长截断到 video_duration
6. 如果配置了降噪,最后应用降噪
1. 保留主图层(main/broll/overlay/corner_voice)视频素材的原声,按顺序 concat 拼接
2. 每个 clip 按 config.volume 应用音量(volume=0 静音,=1 原声)
3. 独立音频轨(audio role,TTS/配音)通过 amix 混入
4. 如果提供了 bgm_path,则额外混入 BGM(支持淡入淡出、循环、人声闪避)
5. 如果配置了 audio_tracks,则混入多轨道音频(配音、音效等)
6. 输出时长截断到 video_duration
7. 如果配置了降噪,最后应用降噪
Args:
ctx: 渲染上下文
@@ -123,10 +134,14 @@ def mix_audio(
if "audio" in layer_map:
audio_clips = layer_map["audio"].clips
# ── 丢弃源视频的原始音频(避免录入杂音),成片仅保留 TTS 配音 + BGM ──
main_clips = []
# ── 保留源视频原声:过滤掉无音频流的 main clip(图片/无声素材) ──
# 每个 clip 按 config.volume 控制音量(0=静音,1=原声),静音 clip 不送入 concat
main_clips = [
c for c in main_clips
if clip_has_audio(ctx, c) and _clip_volume(c) > 0
]
# ── 防御:过滤掉无音频流的 clip ──
# ── 防御:过滤掉无音频流的独立音频轨 ──
audio_clips = [c for c in audio_clips if clip_has_audio(ctx, c)]
if not main_clips and not audio_clips:
@@ -144,7 +159,7 @@ def mix_audio(
# 构建音频处理命令
output_path = ctx.work_dir / f"audio_{ctx.plan_id}.aac"
# 源视频原始音频已被丢弃(main_clips = []),最终音频完全由独立音频轨 + BGM + 多轨配置组成。
# 主音频为视频素材原声 concat;独立音频轨(TTS/配音)通过 amix 混入。
# 当无 main_clips 时,将独立音频轨作为主音频走 concat 拼接;当二者均有则走 amix 混音。
if main_clips:
effective_main = main_clips
@@ -282,6 +297,9 @@ def concat_main_audio(
"-ac",
"2",
]
vol = _clip_volume(clip)
if abs(vol - 1.0) >= 1e-6:
command.extend(["-af", f"volume={vol:.4f}"])
if trim_start > 0:
command.extend(["-ss", f"{trim_start:.3f}"])
if final_duration > 0:
@@ -312,6 +330,11 @@ def concat_main_audio(
if reverse_filter:
audio_filters.append(reverse_filter)
# 音量
vol = _clip_volume(clip)
if abs(vol - 1.0) >= 1e-6:
audio_filters.append(f"volume={vol:.4f}")
# aformat 归一化:统一输出格式为 48000Hz + stereo + fltp
audio_filters.append("aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp")
@@ -378,6 +401,11 @@ def concat_main_audio(
if reverse_filter:
audio_filters.append(reverse_filter)
# 音量(0=静音,1=原声)
vol = _clip_volume(clip)
if abs(vol - 1.0) >= 1e-6:
audio_filters.append(f"volume={vol:.4f}")
# aformat 归一化:统一采样率48000Hz + 双声道stereo + fltp采样格式
# concat filter 要求所有输入音频参数完全一致,否则 exit=234 失败
audio_filters.append("aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp")
@@ -36,6 +36,7 @@ from video_processing.ffmpeg_utils import (
DEFAULT_TRANSITION_DURATION,
FFMPEG_BIN,
probe_duration,
probe_has_audio,
probe_video_info,
run_ffmpeg,
)
@@ -1000,6 +1001,11 @@ class UnifiedRenderService:
if abs(speed - 1.0) >= 1e-6:
return False, f"有调速: speed={speed:.2f}x"
# 音量非默认(静音/放大)→ 需要音频滤镜重编码 → 不能 copy
vol = UnifiedRenderService._clip_volume(clip)
if abs(vol - 1.0) >= 1e-6:
return False, f"音量非默认: volume={vol:.2f}"
# 有倒放 → 需要重编码 → 不能 copy
reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
if reverse_config.enabled and (reverse_config.reverse_video or reverse_config.reverse_audio):
@@ -1270,12 +1276,23 @@ class UnifiedRenderService:
"+faststart",
]
# 音频处理:background 通常是图片无音频,跳过;其他编码为 aac
# background 以外的视频素材,默认带音频
has_audio = role != "background"
# 音频处理:background 通常是图片无音频,跳过;其他角色先探测是否真有音频流。
# volume=0 表示该片段静音,直接丢弃音频(与多片段 mix_audio 行为一致)。
clip_volume = UnifiedRenderService._clip_volume(clip)
if role != "background" and clip_volume > 0:
try:
has_audio = probe_has_audio(clip.local_path)
except Exception:
has_audio = True
else:
has_audio = False
if has_audio:
af_parts: list[str] = []
# 片段音量(原声=1.0,静音已在上层排除)
if abs(clip_volume - 1.0) >= 1e-6:
af_parts.append(f"volume={clip_volume:.4f}")
# 音频降噪
try:
from video_processing.noise_reduction_engine import NoiseReductionConfig, NoiseReductionEngine
@@ -1976,6 +1993,16 @@ class UnifiedRenderService:
"""
return _clip_playback_speed_pure(getattr(clip, "playback_speed", 1.0))
@staticmethod
def _clip_volume(clip: ResolvedClip) -> float:
"""获取 clip 的音量(config.volume)。缺省 1.0 原声,0.0 静音。"""
cfg = getattr(clip, "config", None) or {}
try:
vol = float(cfg.get("volume", 1.0))
except (TypeError, ValueError):
return 1.0
return max(0.0, vol)
@staticmethod
def _clip_adjusted_duration(clip: ResolvedClip) -> float:
"""计算调速后的 clip 实际时长(用于拼接计算)。
+193
View File
@@ -0,0 +1,193 @@
"""回归测试:render_plan 新路径必须保留视频素材原声。
历史 Bug:mix_audio 中 `main_clips = []` 无条件丢弃源视频原声,
导致最终生成视频没有原声(与预览不一致)。本测试钉住新行为:
- 有音频流的 main/broll clip 原声必须进入最终音轨
- clip.config.volume=0 静音,volume≠1.0 应用音量滤镜
- 无音频流的素材被安全过滤
- 直通(pass-through)路径同样尊重 probe + volume
"""
from __future__ import annotations
from pathlib import Path
from unittest.mock import patch
import pytest
from video_processing.render_audio import (
RenderContext,
_clip_volume,
mix_audio,
)
from video_processing.unified_render_service import (
ResolvedClip,
UnifiedRenderService,
)
def _ctx() -> RenderContext:
return RenderContext(work_dir=Path("/tmp/test_render_audio_fix"), plan_id="plan_audio")
def _clip(
cid: str,
*,
clip_type: str = "main",
order: int = 0,
duration: float = 5.0,
config: dict | None = None,
asset: str | None = None,
) -> ResolvedClip:
return ResolvedClip(
clip_id=cid,
asset_id=asset or f"asset_{cid}.mp4",
clip_type=clip_type,
order=order,
local_path=Path(f"/tmp/asset_{cid}.mp4"),
duration=duration,
actual_duration=duration,
config=config or {},
)
def _layers(svc, clips):
# 直接把 ResolvedClip 分组为图层,跳过 _resolve_clips(后者要求原始 EditPlanClip)
return svc._group_clips_into_layers(clips)
def _service(clips):
paths = {c.asset_id: c.local_path for c in clips}
return UnifiedRenderService(
plan=type("P", (), {"id": "plan_audio", "config": {}})(),
clips=clips,
asset_path_map=paths,
work_dir=Path("/tmp/test_render_audio_fix"),
output_width=1080,
output_height=1920,
output_fps=25,
)
class TestOriginalAudioRetained:
"""钉住原声不再被丢弃。"""
def test_single_main_clip_audio_kept(self):
svc = _service([_clip("c1")])
with (
patch("pathlib.Path.exists", return_value=True),
patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
patch("video_processing.render_audio.probe_has_audio", return_value=True),
patch("video_processing.render_audio.run_ffmpeg") as mock_run,
):
result = mix_audio(_ctx(), _layers(svc, [_clip("c1")]), 5.0)
assert result is not None
cmd_str = " ".join(mock_run.call_args[0][0])
assert "asset_c1.mp4" in cmd_str
def test_multi_main_clips_concat_audio(self):
clips = [_clip("c1", order=0), _clip("c2", order=1)]
svc = _service(clips)
with (
patch("pathlib.Path.exists", return_value=True),
patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
patch("video_processing.render_audio.probe_has_audio", return_value=True),
patch("video_processing.render_audio.run_ffmpeg") as mock_run,
):
result = mix_audio(_ctx(), _layers(svc, clips), 9.0)
assert result is not None
cmd_str = " ".join(mock_run.call_args[0][0])
assert "asset_c1.mp4" in cmd_str
assert "asset_c2.mp4" in cmd_str
assert "concat=n=2:v=0:a=1" in cmd_str
def test_main_audio_plus_independent_track_amix(self):
clips = [
_clip("c1", order=0),
_clip("tts1", order=1, config={"role": "audio", "volume": 0.5}),
]
svc = _service(clips)
with (
patch("pathlib.Path.exists", return_value=True),
patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
patch("video_processing.render_audio.probe_has_audio", return_value=True),
patch("video_processing.render_audio.run_ffmpeg") as mock_run,
):
result = mix_audio(_ctx(), _layers(svc, clips), 5.0)
assert result is not None
cmd_str = " ".join(mock_run.call_args[0][0])
assert "amix=inputs=2" in cmd_str
assert "asset_c1.mp4" in cmd_str
assert "asset_tts1.mp4" in cmd_str
assert "volume=0.5" in cmd_str
def test_silent_clip_volume_zero_excluded(self):
"""volume=0 的素材不送入原声 concat。"""
clips = [_clip("mute", order=0, config={"volume": 0})]
svc = _service(clips)
with (
patch("pathlib.Path.exists", return_value=True),
patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
patch("video_processing.render_audio.probe_has_audio", return_value=True),
patch("video_processing.render_audio.run_ffmpeg") as mock_run,
):
result = mix_audio(_ctx(), _layers(svc, clips), 5.0)
assert result is None
mock_run.assert_not_called()
def test_no_audio_stream_returns_none(self):
clips = [_clip("c1")]
svc = _service(clips)
with (
patch("pathlib.Path.exists", return_value=True),
patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
patch("video_processing.render_audio.probe_has_audio", return_value=False),
patch("video_processing.render_audio.run_ffmpeg") as mock_run,
):
result = mix_audio(_ctx(), _layers(svc, clips), 5.0)
assert result is None
mock_run.assert_not_called()
def test_volume_helper_default_and_override(self):
assert _clip_volume(_clip("c1")) == 1.0
assert _clip_volume(_clip("c2", config={"volume": 0.3})) == pytest.approx(0.3)
assert _clip_volume(_clip("c3", config={"volume": 0})) == 0.0
class TestPassThroughAudioProbe:
"""直通路径必须先探测音频,不能无条件假设 main 有音频。"""
def test_pass_through_probes_audio_before_encoding(self):
clips = [_clip("c1")]
svc = _service(clips)
with (
patch("pathlib.Path.exists", return_value=True),
patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
patch(
"video_processing.unified_render_service.probe_has_audio",
return_value=False,
) as mock_probe,
patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
):
layers = svc._group_clips_into_layers(clips)
has_audio = svc._render_pass_through(layers, Path("/tmp/out.mp4"), video_duration=5.0)
assert has_audio is False
mock_probe.assert_called()
cmd_str = " ".join(mock_run.call_args[0][0])
assert "aac" not in cmd_str
def test_pass_through_volume_non_default_disables_stream_copy(self):
svc = _service([_clip("c1", config={"volume": 0.5})])
clip = _clip("c1", config={"volume": 0.5})
can_copy, reason = svc._can_use_stream_copy(clip)
assert can_copy is False
assert "音量" in reason
def test_clip_volume_static_helper(self):
assert UnifiedRenderService._clip_volume(_clip("c1")) == 1.0
assert UnifiedRenderService._clip_volume(_clip("c2", config={"volume": 0.7})) == pytest.approx(0.7)
+96 -81
View File
@@ -60,6 +60,17 @@ class FakePlan:
config: dict[str, Any] = field(default_factory=dict)
@pytest.fixture(autouse=True)
def _assume_source_clips_have_audio():
"""默认假设测试中的视频素材都带音频流。
新行为:保留视频素材原声(不再无条件丢弃)。需要模拟无音频流的用例
自行 patch probe_has_audio=False(如 test_mix_audio_main_no_audio_stream_returns_none)。
"""
with patch("video_processing.render_audio.probe_has_audio", return_value=True):
yield
def _make_clip(
clip_id: str,
clip_type: str = "main",
@@ -976,7 +987,7 @@ class TestAudioMixing:
assert UnifiedRenderService._clip_effective_duration(clip) == 0.0
def test_mix_audio_single_main_clip(self):
"""只有 main clip(无独立音频轨)→ 源视频音频被丢弃,返回 None。"""
"""只有 main clip(无独立音频轨)→ 保留源视频原声,走单轨 concat。"""
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)
@@ -990,14 +1001,15 @@ class TestAudioMixing:
ctx = _make_ctx()
result = mix_audio(ctx, layers, 5.0)
# 源视频音频被丢弃,没有独立音频轨 → 无音频
assert result is None
mock_run.assert_not_called()
# 保留源视频原声
assert result is not None
mock_run.assert_called_once()
assert "asset_c1.mp4" in " ".join(mock_run.call_args[0][0])
def test_mix_audio_multi_main_clips(self):
"""多个独立音频轨用 concat 拼接(main 图层音频被丢弃)。"""
"""main 原声与独立音频轨通过 amix 混音。"""
clips = [
_make_clip("c1", "main", order=0, duration=3.0), # 源视频音频被丢弃
_make_clip("c1", "main", order=0, duration=3.0),
_make_clip("tts1", "main", order=0, duration=3.0, config={"role": "audio"}),
_make_clip("tts2", "main", order=1, duration=2.0, config={"role": "audio"}),
]
@@ -1021,15 +1033,17 @@ class TestAudioMixing:
mock_run.assert_called_once()
cmd = mock_run.call_args[0][0]
cmd_str = " ".join(cmd)
# 2 个独立音频轨 concat
assert "concat=n=2:v=0:a=1" in cmd_str
# main 的 c1 不参与音频(源视频杂音被丢弃)
assert "asset_c1.mp4" not in cmd_str
# 原声 + 2 个独立音频轨 → amix 混音(3 路输入)
assert "amix=inputs=3" in cmd_str
# main 原声 c1 参与混音
assert "asset_c1.mp4" in cmd_str
assert "asset_tts1.mp4" in cmd_str
assert "asset_tts2.mp4" in cmd_str
def test_mix_audio_with_independent_audio_track(self):
"""独立音频轨生效;main 图层源视频音频被丢弃。"""
"""main 原声与独立音频轨通过 amix 混音。"""
clips = [
_make_clip("c1", "main", order=0, duration=5.0), # 源视频音频被丢弃
_make_clip("c1", "main", order=0, duration=5.0),
_make_clip(
"bgm1",
"main",
@@ -1057,9 +1071,9 @@ class TestAudioMixing:
mock_run.assert_called_once()
cmd = mock_run.call_args[0][0]
cmd_str = " ".join(cmd)
# main 的源视频 c1 不参与音频
assert "asset_c1.mp4" not in cmd_str
# 独立音频轨 bgm1 作为最终音频生效
# main 原声 c1 与独立音频轨 bgm1 都参与 amix
assert "amix=inputs=2" in cmd_str
assert "asset_c1.mp4" in cmd_str
assert "asset_bgm1.mp4" in cmd_str
def test_mix_with_independent_audio_amix(self):
@@ -1122,7 +1136,7 @@ class TestAudioMixing:
assert result is None
def test_mix_audio_background_not_used_as_main(self):
"""main/background 图层音频都被丢弃,只有独立音频轨参与混音。"""
"""background 图层不参与主音频;main 原声与独立音频轨混音。"""
clips = [
_make_clip("bg1", "background", order=0, duration=5.0),
_make_clip("c1", "main", order=0, duration=5.0),
@@ -1148,14 +1162,15 @@ class TestAudioMixing:
mock_run.assert_called_once()
cmd = mock_run.call_args[0][0]
cmd_str = " ".join(cmd)
# 源视频(background + main)音频都被丢弃
# background 图层不参与主音频
assert "asset_bg1.mp4" not in cmd_str
assert "asset_c1.mp4" not in cmd_str
# 只有独立音频轨
# main 原声 + 独立音频轨都参与
assert "amix=inputs=2" in cmd_str
assert "asset_c1.mp4" in cmd_str
assert "asset_tts1.mp4" in cmd_str
def test_mix_audio_main_priority_over_broll(self):
"""main/broll 图层的源视频音频都被丢弃,只使用独立音频轨。"""
"""main 图层优先作为主音频,broll 不参与;与独立音频轨 amix。"""
clips = [
_make_clip("b1", "b_roll", order=0, duration=5.0),
_make_clip("c1", "main", order=0, duration=5.0),
@@ -1181,13 +1196,14 @@ class TestAudioMixing:
mock_run.assert_called_once()
cmd = mock_run.call_args[0][0]
cmd_str = " ".join(cmd)
# main 和 broll 的源视频音频都被丢弃
assert "asset_c1.mp4" not in cmd_str
# main 原声 c1 优先参与;broll b1 不参与主音频
assert "amix=inputs=2" in cmd_str
assert "asset_c1.mp4" in cmd_str
assert "asset_b1.mp4" not in cmd_str
assert "asset_tts1.mp4" in cmd_str
def test_mix_audio_broll_used_when_no_main(self):
"""broll 图层源视频音频也被丢弃;无独立音频轨 → 返回 None。"""
"""无 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)
@@ -1201,14 +1217,15 @@ class TestAudioMixing:
ctx = _make_ctx()
result = mix_audio(ctx, layers, 5.0)
# 源视频音频被丢弃,无独立音频轨 → 无音频
assert result is None
mock_run.assert_not_called()
# 保留 broll 原声
assert result is not None
mock_run.assert_called_once()
assert "asset_b1.mp4" in " ".join(mock_run.call_args[0][0])
def test_mix_audio_single_clip_truncated_to_video_duration(self):
"""单独立音频轨截断到 video_duration(video_duration < clip有效时长)。"""
"""主音频截断到 video_duration(video_duration < clip有效时长)。"""
clips = [
_make_clip("c1", "main", order=0, duration=10.0), # 源视频音频被丢弃
_make_clip("c1", "main", order=0, duration=10.0),
_make_clip("tts1", "main", order=0, duration=10.0, config={"role": "audio"}),
]
asset_paths = {
@@ -1233,8 +1250,8 @@ class TestAudioMixing:
# 验证截断到 3.0(-t 3.0 或 atrim=0:3.000)
cmd_str = " ".join(cmd)
assert "3.000" in cmd_str or "3.0" in cmd_str
# 源视频不参与
assert "asset_c1.mp4" not in cmd_str
# main 原声参与
assert "asset_c1.mp4" in cmd_str
def test_merge_audio_video(self):
"""合并音视频命令正确。"""
@@ -1391,11 +1408,11 @@ class TestAudioMixing:
mock_run.assert_not_called()
def test_mix_audio_partial_clips_no_audio_filtered(self):
"""main 图层音频全部丢弃;独立音频轨有/无音频时按预期过滤。"""
"""main 原声正常保留;无音频流的 clip 被过滤。"""
clips = [
_make_clip("c1", "main", order=0, duration=3.0), # 源视频音频被丢弃
_make_clip("c2", "main", order=1, duration=2.0), # 源视频音频被丢弃
_make_clip("tts1", "main", order=0, duration=2.0, config={"role": "audio"}), # 独立音频轨
_make_clip("c1", "main", order=0, duration=3.0),
_make_clip("c2", "main", order=1, duration=2.0),
_make_clip("tts1", "main", order=0, duration=2.0, config={"role": "audio"}),
]
asset_paths = {
"asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
@@ -1417,16 +1434,15 @@ class TestAudioMixing:
mock_run.assert_called_once()
cmd = mock_run.call_args[0][0]
cmd_str = " ".join(cmd)
# 只有独立音频轨 tts1 参与
# main 原声 c1/c2 + 独立音频轨 tts1 全部参与
assert "asset_c1.mp4" in cmd_str
assert "asset_c2.mp4" in cmd_str
assert "asset_tts1.mp4" in cmd_str
# main 的 c1/c2 源视频音频被丢弃
assert "asset_c1.mp4" not in cmd_str
assert "asset_c2.mp4" not in cmd_str
def test_mix_audio_all_main_no_audio_but_independent_track(self):
"""main 图层源视频音频全部丢弃;仅独立音频轨生效,走 concat 单轨路径。"""
def test_mix_audio_main_plus_independent_amix(self):
"""main 原声与独立音频轨 amix 混音。"""
clips = [
_make_clip("c1", "main", order=0, duration=5.0), # 源视频音频被丢弃
_make_clip("c1", "main", order=0, duration=5.0),
_make_clip(
"bgm1",
"main",
@@ -1454,9 +1470,9 @@ class TestAudioMixing:
mock_run.assert_called_once()
cmd = mock_run.call_args[0][0]
cmd_str = " ".join(cmd)
# main 的源视频 c1 不参与音频
assert "asset_c1.mp4" not in cmd_str
# 只有独立音频轨 bgm1 作为主音频走单轨拼接
# main 原声 c1 与独立音频轨 bgm1 都参与 amix
assert "amix=inputs=2" in cmd_str
assert "asset_c1.mp4" in cmd_str
assert "asset_bgm1.mp4" in cmd_str
def test_mix_audio_both_no_audio_returns_none(self):
@@ -2116,9 +2132,9 @@ class TestConcatNormalizeAudioFormat:
"""
def test_multi_clip_concat_has_aformat(self):
"""多独立音频轨 concat 前,每个轨都有 aformat 归一化(main 图层源视频音频被丢弃)。"""
"""main 原声 + 独立音频轨在 concat/amix 前都有 aformat 归一化。"""
clips = [
_make_clip("c1", "main", order=0, duration=3.0), # 源视频音频被丢弃
_make_clip("c1", "main", order=0, duration=3.0),
_make_clip("tts1", "main", order=0, duration=3.0, config={"role": "audio"}),
_make_clip("tts2", "main", order=1, duration=2.0, config={"role": "audio"}),
]
@@ -2148,14 +2164,14 @@ class TestConcatNormalizeAudioFormat:
assert "channel_layouts=stereo" in cmd_str, "声道应统一为 stereo"
assert "sample_fmts=fltp" in cmd_str, "采样格式应统一为 fltp"
# 2 个独立音频轨都应有 aformat
# main 原声 + 2 个独立音频轨都应有 aformat
aformat_count = cmd_str.count("aformat=")
assert aformat_count >= 2, f"每个独立音频轨都应有 aformat,实际 {aformat_count} 个"
assert aformat_count >= 3, f"3 路音频都应有 aformat,实际 {aformat_count} 个"
# 有 concat
assert "concat=n=2:v=0:a=1" in cmd_str
# main 源视频不参与
assert "asset_c1.mp4" not in cmd_str
# amix 3 路输入
assert "amix=inputs=3" in cmd_str
# main 源视频原声参与
assert "asset_c1.mp4" in cmd_str
def test_aformat_before_concat(self):
"""aformat 应在 concat 之前(每个独立音频轨处理链中 aformat 在 concat 之前)。"""
@@ -2190,9 +2206,9 @@ class TestConcatNormalizeAudioFormat:
assert aformat_before_count >= 2, f"concat 之前每个独立音频轨都应有 aformat,实际 {aformat_before_count} 个"
def test_single_clip_audio_has_normalized_output(self):
"""单独立音频轨输出也应统一格式(一致性保障)。"""
"""原声+独立音频轨输出统一格式(一致性保障)。"""
clips = [
_make_clip("c1", "main", order=0, duration=5.0), # 源视频音频被丢弃
_make_clip("c1", "main", order=0, duration=5.0),
_make_clip("tts1", "main", order=0, duration=5.0, config={"role": "audio"}),
]
asset_paths = {
@@ -2212,21 +2228,20 @@ class TestConcatNormalizeAudioFormat:
assert mock_run.called
cmd = mock_run.call_args[0][0]
# 单独立音频轨简单路径应有 -ar 48000 和 -ac 2
assert "-ar" in cmd, "单独立音频轨应指定采样率"
ar_idx = cmd.index("-ar")
assert cmd[ar_idx + 1] == "48000", "采样率应为 48000"
assert "-ac" in cmd, "单独立音频轨应指定声道数"
ac_idx = cmd.index("-ac")
assert cmd[ac_idx + 1] == "2", "声道数应为 2(stereo)"
cmd_str = " ".join(cmd)
assert "asset_c1.mp4" not in cmd_str
# 原声 + 独立音频轨走 amix:两路都有 aformat 归一化
assert "amix=inputs=2" in cmd_str
assert cmd_str.count("aformat=") >= 2
assert "sample_rates=48000" in cmd_str
assert "channel_layouts=stereo" in cmd_str
assert "asset_c1.mp4" in cmd_str
assert "asset_tts1.mp4" in cmd_str
def test_independent_audio_track_has_aformat(self):
"""独立音频轨输出也应统一格式(48000Hz + stereo + aac)。"""
"""原声+独立音频轨输出统一格式(48000Hz + stereo + aac)。"""
clips = [
_make_clip("c1", "main", order=0, duration=5.0), # 源视频音频被丢弃
_make_clip("c1", "main", order=0, duration=5.0),
_make_clip(
"audio1",
"main",
@@ -2254,15 +2269,13 @@ class TestConcatNormalizeAudioFormat:
cmd = mock_run.call_args[0][0]
cmd_str = " ".join(cmd)
# 单独立音频轨走 concat 单轨简单路径:-ar 48000 -ac 2
assert "-ar" in cmd
ar_idx = cmd.index("-ar")
assert cmd[ar_idx + 1] == "48000"
assert "-ac" in cmd
ac_idx = cmd.index("-ac")
assert cmd[ac_idx + 1] == "2"
# main 源视频 c1 不参与音频
assert "asset_c1.mp4" not in cmd_str
# 原声 + 独立音频轨走 amix:两路都有 aformat 归一化
assert "amix=inputs=2" in cmd_str
assert cmd_str.count("aformat=") >= 2
assert "sample_rates=48000" in cmd_str
assert "channel_layouts=stereo" in cmd_str
# main 原声 c1 与独立音频轨 audio1 都参与
assert "asset_c1.mp4" in cmd_str
assert "asset_audio1.mp4" in cmd_str
@@ -2299,9 +2312,9 @@ class TestConcatNormalizeAudioCodec:
assert "-b:a" in cmd, "应指定音频码率"
def test_single_clip_output_is_aac(self):
"""单独立音频轨输出编码为 aac。"""
"""原声+独立音频轨输出编码为 aac。"""
clips = [
_make_clip("c1", "main", order=0, duration=5.0), # 源视频音频被丢弃
_make_clip("c1", "main", order=0, duration=5.0),
_make_clip("tts1", "main", order=0, duration=5.0, config={"role": "audio"}),
]
asset_paths = {
@@ -2322,7 +2335,8 @@ class TestConcatNormalizeAudioCodec:
assert mock_run.called
cmd = mock_run.call_args[0][0]
assert "aac" in cmd
assert "asset_c1.mp4" not in " ".join(cmd)
# main 原声参与
assert "asset_c1.mp4" in " ".join(cmd)
class TestConcatNormalizeFourItemsComplete:
@@ -2372,7 +2386,7 @@ class TestConcatNormalizeFourItemsComplete:
assert fps_count >= 3, f"3个 clip 都应有 fps=30,实际 {fps_count} 个"
def test_audio_format_normalized(self):
"""[3/4] 音频格式:3 个独立音频轨 concat 前都有 aformat 归一化(main 图层源视频音频被丢弃)。"""
"""[3/4] 音频格式:3 个 main 原声 + 3 个独立音频轨都有 aformat 归一化。"""
clips = self._make_one_take_clips() + [
_make_clip("tts1", "main", order=10, duration=5.0, config={"role": "audio"}),
_make_clip("tts2", "main", order=11, duration=4.0, config={"role": "audio"}),
@@ -2396,13 +2410,13 @@ class TestConcatNormalizeFourItemsComplete:
cmd_str = " ".join(cmd)
aformat_count = cmd_str.count("aformat=")
assert aformat_count >= 3, f"3个独立音频轨都应有 aformat,实际 {aformat_count} 个"
assert aformat_count >= 6, f"6 路音频(3原声+3独立轨)都应有 aformat,实际 {aformat_count} 个"
assert "sample_rates=48000" in cmd_str
assert "channel_layouts=stereo" in cmd_str
assert "sample_fmts=fltp" in cmd_str
# main 图层源视频不参与
# main 图层原声参与
for i in range(1, 4):
assert f"asset_c{i}.mp4" not in cmd_str
assert f"asset_c{i}.mp4" in cmd_str
def test_audio_codec_aac(self):
"""[4/4] 音频编码:输出为 aac(仅独立音频轨参与)。"""
@@ -2427,6 +2441,7 @@ class TestConcatNormalizeFourItemsComplete:
assert mock_run.called
cmd = mock_run.call_args[0][0]
assert "aac" in cmd
# 3 个 main 原声 concat 后再与独立轨 amix
assert "concat=n=3:v=0:a=1" in " ".join(cmd)