mirror of
https://github.com/linyqh/NarratoAI.git
synced 2026-06-30 03:15:16 +00:00
dc12f390bb
- 为VideoClipParams新增原字幕路径配置字段，支持单条/多条字幕路径 - 完善webui参数获取逻辑，处理字幕路径兼容性并对接前端选择 - 重构后端字幕处理流程，支持自动匹配视频对应原字幕，合并原声字幕 - 优化视频合并逻辑，新增ffmpeg无损copy合并判断，自动回退重编码提升效率 - 新增ffmpeg快速素材合并路径，支持自定义字幕样式与多音轨混合 - 新增多个单元测试覆盖字幕匹配、合并及视频合并场景
121 lines
4.4 KiB
Python
121 lines
4.4 KiB
Python
import subprocess
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from app.services import merger_video
|
|
|
|
|
|
class MergerVideoConcatTests(unittest.TestCase):
|
|
def test_can_concat_video_copy_when_signatures_match(self):
|
|
signature = {
|
|
"codec_name": "h264",
|
|
"profile": "High",
|
|
"width": 1080,
|
|
"height": 1920,
|
|
"pix_fmt": "yuv420p",
|
|
"r_frame_rate": "30/1",
|
|
"avg_frame_rate": "30/1",
|
|
"time_base": "1/15360",
|
|
"sample_aspect_ratio": "1:1",
|
|
}
|
|
|
|
with mock.patch.object(
|
|
merger_video,
|
|
"_get_video_stream_signature",
|
|
side_effect=[signature, dict(signature)],
|
|
):
|
|
self.assertTrue(merger_video._can_concat_video_copy(["1.mp4", "2.mp4"]))
|
|
|
|
def test_can_concat_video_copy_rejects_mismatched_signature(self):
|
|
base_signature = {
|
|
"codec_name": "h264",
|
|
"profile": "High",
|
|
"width": 1080,
|
|
"height": 1920,
|
|
"pix_fmt": "yuv420p",
|
|
"r_frame_rate": "30/1",
|
|
"avg_frame_rate": "30/1",
|
|
"time_base": "1/15360",
|
|
"sample_aspect_ratio": "1:1",
|
|
}
|
|
mismatch_signature = dict(base_signature, r_frame_rate="24000/1001")
|
|
|
|
with mock.patch.object(
|
|
merger_video,
|
|
"_get_video_stream_signature",
|
|
side_effect=[base_signature, mismatch_signature],
|
|
):
|
|
self.assertFalse(merger_video._can_concat_video_copy(["1.mp4", "2.mp4"]))
|
|
|
|
def test_concat_video_streams_prefers_copy_when_compatible(self):
|
|
completed = subprocess.CompletedProcess(args=["ffmpeg"], returncode=0)
|
|
|
|
with (
|
|
mock.patch.object(merger_video, "_can_concat_video_copy", return_value=True),
|
|
mock.patch.object(merger_video, "_concat_duration_matches", return_value=True),
|
|
mock.patch.object(merger_video.subprocess, "run", return_value=completed) as run_mock,
|
|
):
|
|
merger_video._concat_video_streams(
|
|
["1.mp4", "2.mp4"],
|
|
"concat.txt",
|
|
"video_concat.mp4",
|
|
threads=4,
|
|
)
|
|
|
|
cmd = run_mock.call_args.args[0]
|
|
self.assertEqual("copy", cmd[cmd.index("-c:v") + 1])
|
|
self.assertNotIn("libx264", cmd)
|
|
|
|
def test_concat_video_streams_falls_back_when_copy_duration_mismatches(self):
|
|
completed = subprocess.CompletedProcess(args=["ffmpeg"], returncode=0)
|
|
|
|
with (
|
|
mock.patch.object(merger_video, "_can_concat_video_copy", return_value=True),
|
|
mock.patch.object(merger_video, "_concat_duration_matches", return_value=False),
|
|
mock.patch.object(merger_video.os.path, "exists", return_value=False),
|
|
mock.patch.object(merger_video.subprocess, "run", return_value=completed) as run_mock,
|
|
):
|
|
merger_video._concat_video_streams(
|
|
["1.mp4", "2.mp4"],
|
|
"concat.txt",
|
|
"video_concat.mp4",
|
|
threads=6,
|
|
)
|
|
|
|
self.assertEqual(2, run_mock.call_count)
|
|
fallback_cmd = run_mock.call_args_list[1].args[0]
|
|
self.assertEqual("libx264", fallback_cmd[fallback_cmd.index("-c:v") + 1])
|
|
self.assertEqual("6", fallback_cmd[fallback_cmd.index("-threads") + 1])
|
|
|
|
def test_concat_video_streams_falls_back_to_reencode_when_copy_fails(self):
|
|
copy_error = subprocess.CalledProcessError(
|
|
returncode=1,
|
|
cmd=["ffmpeg"],
|
|
stderr=b"copy failed",
|
|
)
|
|
completed = subprocess.CompletedProcess(args=["ffmpeg"], returncode=0)
|
|
|
|
with (
|
|
mock.patch.object(merger_video, "_can_concat_video_copy", return_value=True),
|
|
mock.patch.object(
|
|
merger_video.subprocess,
|
|
"run",
|
|
side_effect=[copy_error, completed],
|
|
) as run_mock,
|
|
):
|
|
merger_video._concat_video_streams(
|
|
["1.mp4", "2.mp4"],
|
|
"concat.txt",
|
|
"video_concat.mp4",
|
|
threads=8,
|
|
)
|
|
|
|
self.assertEqual(2, run_mock.call_count)
|
|
fallback_cmd = run_mock.call_args_list[1].args[0]
|
|
self.assertEqual("libx264", fallback_cmd[fallback_cmd.index("-c:v") + 1])
|
|
self.assertEqual("8", fallback_cmd[fallback_cmd.index("-threads") + 1])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|