fix(unified-render): 无音轨视频防御 + probe_has_audio 工具
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 1m4s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 2m3s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 4m23s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped

- 新增 ffmpeg_utils.probe_has_audio:ffprobe 探测音频流
- _mix_audio 入口过滤无音频流的 clip,避免 FFmpeg 引用 [i:a] 失败
- 新增 _clip_has_audio 缓存方法,同 clip 只探测一次
- 6个新增单测覆盖:全无音频、部分无音频、主图层无但独立音轨有、缓存、双无音频兜底
This commit is contained in:
灵应
2026-07-12 22:16:34 +08:00
parent ca86240c06
commit 73e853127a
3 changed files with 199 additions and 0 deletions
@@ -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 获取视频时长(秒)。
@@ -1036,6 +1036,7 @@ class UnifiedRenderService:
2. 主图层音频按顺序 concat 拼接
3. 独立音频轨(audio role)用 amix 混入
4. 输出时长截断到 video_duration
5. 无音频流的 clip 会被自动跳过,避免 FFmpeg 引用 [i:a] 失败
Args:
layers: 图层列表
@@ -1067,6 +1068,11 @@ class UnifiedRenderService:
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
@@ -1306,3 +1312,17 @@ class UnifiedRenderService:
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]
+144
View File
@@ -1255,3 +1255,147 @@ class TestAudioMixing:
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