NarratoAI/app/services/test_jianying_task_unittest.py
viccy f6bda521b2 feat(webui, jianying): 添加自动字幕匹配功能并修复webui状态问题
- 为剪映任务模块新增自动根据视频文件名匹配对应字幕文件的逻辑,当未传入原始字幕路径时自动查找并选择最新的匹配字幕
- 修复webui脚本设置页的selectbox状态同步问题,改用session_state作为唯一状态源,避免同时传递index和key导致的冲突
- 更新webui脚本路径的特殊路径判断列表,新增MODE_FILE的特殊情况处理
- 新增两个单元测试用例验证自动字幕匹配和原片字幕导入功能
2026-06-08 16:23:10 +08:00

427 lines
19 KiB
Python

import json
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from app.models.schema import VideoClipParams
from app.services import jianying_draft_builder, jianying_task
DraftPathPlaceholder = "##_draftpath_placeholder_0E685133-18CE-45ED-8CB8-2904A212EC80_##"
class JianyingTaskTests(unittest.TestCase):
def test_normalize_indextts_uses_valid_param_reference(self):
with tempfile.NamedTemporaryFile(suffix=".wav") as ref:
params = VideoClipParams(tts_engine="indextts", voice_name=ref.name)
jianying_task._normalize_indextts_reference_audio(params)
self.assertEqual(f"indextts:{ref.name}", params.voice_name)
def test_normalize_indextts_uses_config_reference_when_param_is_stale(self):
with tempfile.TemporaryDirectory() as temp_dir:
ref_path = Path(temp_dir) / "reference.wav"
ref_path.write_bytes(b"fake wav")
params = VideoClipParams(tts_engine="indextts", voice_name="zh-CN-YunjianNeural")
with patch.dict(jianying_task.config.indextts, {"reference_audio": str(ref_path)}, clear=False):
jianying_task._normalize_indextts_reference_audio(params)
self.assertEqual(f"indextts:{ref_path}", params.voice_name)
def test_normalize_indextts2_uses_valid_param_reference(self):
with tempfile.NamedTemporaryFile(suffix=".wav") as ref:
params = VideoClipParams(tts_engine="indextts2", voice_name=f"indextts2:{ref.name}")
jianying_task._normalize_indextts_reference_audio(params)
self.assertEqual("indextts2", params.tts_engine)
self.assertEqual(f"indextts2:{ref.name}", params.voice_name)
def test_normalize_indextts2_uses_config_reference_when_param_is_stale(self):
with tempfile.TemporaryDirectory() as temp_dir:
ref_path = Path(temp_dir) / "reference.wav"
ref_path.write_bytes(b"fake wav")
params = VideoClipParams(tts_engine="indextts2", voice_name="zh-CN-YunjianNeural")
with patch.dict(jianying_task.config.indextts2, {"reference_audio": str(ref_path)}, clear=False):
jianying_task._normalize_indextts_reference_audio(params)
self.assertEqual(f"indextts2:{ref_path}", params.voice_name)
def test_normalize_omnivoice_clone_uses_valid_param_reference(self):
with tempfile.NamedTemporaryFile(suffix=".wav") as ref:
params = VideoClipParams(tts_engine="omnivoice", voice_name=f"omnivoice:{ref.name}")
with patch.dict(jianying_task.config.omnivoice, {"mode": "voice_clone"}, clear=False):
jianying_task._normalize_indextts_reference_audio(params)
self.assertEqual(f"omnivoice:{ref.name}", params.voice_name)
def test_normalize_omnivoice_auto_does_not_require_reference(self):
params = VideoClipParams(tts_engine="omnivoice", voice_name="omnivoice:auto")
with patch.dict(jianying_task.config.omnivoice, {"mode": "auto", "reference_audio": ""}, clear=False):
jianying_task._normalize_indextts_reference_audio(params)
self.assertEqual("omnivoice:auto", params.voice_name)
def test_normalize_indextts_requires_existing_reference_audio(self):
params = VideoClipParams(tts_engine="indextts", voice_name="zh-CN-YunjianNeural")
with patch.dict(jianying_task.config.indextts, {"reference_audio": ""}, clear=False):
with self.assertRaisesRegex(ValueError, "IndexTTS-1.5 参考音频不存在"):
jianying_task._normalize_indextts_reference_audio(params)
def test_floor_duration_to_milliseconds(self):
self.assertAlmostEqual(6.997, jianying_task._floor_duration_to_milliseconds(6.997333))
self.assertAlmostEqual(7.0, jianying_task._floor_duration_to_milliseconds(7.000999))
def test_clamp_duration_to_media_uses_actual_media_duration(self):
duration_cache = {}
with patch.object(jianying_task, "get_media_duration_ffprobe", return_value=4.2809):
duration = jianying_task._clamp_duration_to_media(
requested_duration=4.31,
media_file="/tmp/clip.mp4",
duration_cache=duration_cache,
media_label="视频素材",
)
self.assertAlmostEqual(4.28, duration)
def test_clamp_duration_to_media_respects_source_start_time(self):
duration_cache = {}
with patch.object(jianying_task, "get_media_duration_ffprobe", return_value=10.0):
duration = jianying_task._clamp_duration_to_media(
requested_duration=4.0,
media_file="/tmp/original.mp4",
duration_cache=duration_cache,
media_label="原始视频素材",
source_start_time=8.5,
)
self.assertAlmostEqual(1.5, duration)
def test_format_seconds_for_trange_uses_millisecond_precision(self):
self.assertEqual("4.280s", jianying_task._format_seconds_for_trange(4.28))
def test_write_plaintext_jianying_draft_creates_root_package(self):
with tempfile.TemporaryDirectory() as temp_dir:
root_path = Path(temp_dir) / "drafts"
output_dir = Path(temp_dir) / "task"
root_path.mkdir()
output_dir.mkdir()
video_path = output_dir / "clip:01.mp4"
audio_path = output_dir / "audio_00_00_00,000-00_00_04,310.mp3"
video_path.write_bytes(b"fake video")
audio_path.write_bytes(b"fake audio")
params = VideoClipParams(
video_origin_path=str(video_path),
original_volume=0.4,
tts_volume=0.9,
)
script = [
{
"OST": 0,
"start_time": 0.0,
"duration": 4.31,
"timestamp": "00:00:00,000-00:00:04,310",
"video": str(video_path),
"audio": str(audio_path),
}
]
def fake_duration(file_path):
return 4.2809 if file_path == str(video_path) else 5.0
with (
patch.object(jianying_draft_builder, "_get_media_duration_ffprobe", side_effect=fake_duration),
patch.object(
jianying_draft_builder,
"_get_video_metadata_ffprobe",
return_value=(4_280_000, 720, 1280),
),
):
draft_path, draft_name = jianying_draft_builder.write_plaintext_jianying_draft(
str(root_path),
"NarratoAI_test",
script,
params,
str(output_dir),
)
draft_dir = Path(draft_path)
self.assertEqual("NarratoAI_test", draft_name)
self.assertTrue((draft_dir / "draft_info.json").exists())
self.assertTrue((draft_dir / "template-2.tmp").exists())
self.assertTrue((draft_dir / "template.tmp").exists())
self.assertTrue((draft_dir / "draft_cover.jpg").exists())
self.assertFalse((draft_dir / "draft_content_legacy.json").exists())
self.assertFalse((draft_dir / "Timelines" / "project.json").exists())
self.assertTrue((draft_dir / "assets" / "video" / "clip_01.mp4").exists())
self.assertTrue((draft_dir / "assets" / "audio" / audio_path.name).exists())
draft_info = json.loads((draft_dir / "draft_info.json").read_text(encoding="utf-8"))
self.assertEqual("169.0.0", draft_info["new_version"])
self.assertEqual("NarratoAI_test", draft_info["name"])
self.assertEqual(54, len(draft_info["materials"]))
self.assertEqual(
f"{DraftPathPlaceholder}/assets/video/clip_01.mp4",
draft_info["materials"]["videos"][0]["path"],
)
self.assertEqual(
f"{DraftPathPlaceholder}/assets/audio/{audio_path.name}",
draft_info["materials"]["audios"][0]["path"],
)
self.assertEqual(4_280_000, draft_info["tracks"][0]["segments"][0]["source_timerange"]["duration"])
self.assertEqual(4_280_000, draft_info["tracks"][1]["segments"][0]["source_timerange"]["duration"])
attachment_editing = json.loads((draft_dir / "attachment_editing.json").read_text(encoding="utf-8"))
self.assertEqual("1.0.0", attachment_editing["editing_draft"]["version"])
self.assertFalse(attachment_editing["editing_draft"]["is_use_audio_separation"])
empty_template = json.loads((draft_dir / "template.tmp").read_text(encoding="utf-8"))
self.assertEqual("75.0.0", empty_template["new_version"])
self.assertEqual([], empty_template["tracks"])
root_meta = json.loads((root_path / "root_meta_info.json").read_text(encoding="utf-8"))
self.assertEqual("NarratoAI_test", root_meta["all_draft_store"][0]["draft_name"])
self.assertEqual(str(draft_dir / "draft_info.json"), root_meta["all_draft_store"][0]["draft_json_file"])
def test_write_plaintext_jianying_draft_uses_source_timerange_and_writes_subtitles(self):
with tempfile.TemporaryDirectory() as temp_dir:
root_path = Path(temp_dir) / "drafts"
output_dir = Path(temp_dir) / "task"
root_path.mkdir()
output_dir.mkdir()
video_path = output_dir / "source.mp4"
audio_path = output_dir / "audio_00_00_02,000-00_00_04,000.mp3"
subtitle_path = output_dir / "script_subtitles.srt"
video_path.write_bytes(b"fake source video")
audio_path.write_bytes(b"fake audio")
subtitle_path.write_text(
"1\n00:00:00,000 --> 00:00:01,500\n测试字幕\n",
encoding="utf-8",
)
params = VideoClipParams(
video_origin_path=str(video_path),
original_volume=0.4,
tts_volume=0.9,
subtitle_enabled=True,
font_size=60,
text_fore_color="#FFFFFF",
)
script = [
{
"OST": 0,
"start_time": 2.0,
"source_start_time": 2.0,
"duration": 3.0,
"timestamp": "00:00:02,000-00:00:05,000",
"video": str(video_path),
"audio": str(audio_path),
"use_source_timerange": True,
}
]
def fake_duration(file_path):
return 10.0 if file_path == str(video_path) else 3.0
with (
patch.object(jianying_draft_builder, "_get_media_duration_ffprobe", side_effect=fake_duration),
patch.object(
jianying_draft_builder,
"_get_video_metadata_ffprobe",
return_value=(10_000_000, 1920, 1080),
),
):
draft_path, _ = jianying_draft_builder.write_plaintext_jianying_draft(
str(root_path),
"NarratoAI_source",
script,
params,
str(output_dir),
subtitle_path=str(subtitle_path),
)
draft_info = json.loads((Path(draft_path) / "draft_info.json").read_text(encoding="utf-8"))
self.assertEqual(1, len(draft_info["materials"]["videos"]))
self.assertEqual(1, len(draft_info["materials"]["texts"]))
self.assertIn("测试字幕", draft_info["materials"]["texts"][0]["content"])
video_segment = draft_info["tracks"][0]["segments"][0]
self.assertEqual(2_000_000, video_segment["source_timerange"]["start"])
self.assertEqual(3_000_000, video_segment["source_timerange"]["duration"])
self.assertEqual(0.0, video_segment["volume"])
text_tracks = [track for track in draft_info["tracks"] if track["type"] == "text"]
self.assertEqual(1, len(text_tracks))
self.assertEqual(1, len(text_tracks[0]["segments"]))
self.assertEqual(1_500_000, text_tracks[0]["segments"][0]["target_timerange"]["duration"])
def test_build_jianying_draft_script_references_original_video(self):
with tempfile.TemporaryDirectory() as temp_dir:
video_one = Path(temp_dir) / "one.mp4"
video_two = Path(temp_dir) / "two.mp4"
audio_path = Path(temp_dir) / "audio.mp3"
video_one.write_bytes(b"one")
video_two.write_bytes(b"two")
audio_path.write_bytes(b"audio")
params = VideoClipParams(
video_origin_path=str(video_one),
video_origin_paths=[str(video_one), str(video_two)],
)
script = [
{
"_id": 9,
"video_id": 2,
"timestamp": "00:00:05,000-00:00:07,000",
"narration": "解说",
"OST": 0,
}
]
tts_results = [
{
"_id": 9,
"timestamp": "00:00:05,000-00:00:07,000",
"audio_file": str(audio_path),
"subtitle_file": "",
"duration": 1.25,
}
]
draft_script = jianying_task._build_jianying_draft_script(script, params, tts_results)
self.assertEqual(str(video_two), draft_script[0]["video"])
self.assertEqual(str(audio_path), draft_script[0]["audio"])
self.assertEqual(5.0, draft_script[0]["source_start_time"])
self.assertEqual(1.25, draft_script[0]["duration"])
self.assertTrue(draft_script[0]["use_source_timerange"])
def test_get_original_subtitle_paths_falls_back_to_matching_video_name(self):
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
video_path = temp_path / "episode_20260608010240.mp4"
older_subtitle = temp_path / "episode_fun_asr_20260608000100.srt"
newer_subtitle = temp_path / "episode_fun_asr_20260608010100.srt"
video_path.write_bytes(b"video")
older_subtitle.write_text("old", encoding="utf-8")
newer_subtitle.write_text("new", encoding="utf-8")
params = VideoClipParams(video_origin_path=str(video_path))
with patch.object(jianying_task.utils, "subtitle_dir", return_value=str(temp_path)):
subtitle_paths = jianying_task._get_original_subtitle_paths(params)
self.assertEqual([str(newer_subtitle)], subtitle_paths)
def test_create_jianying_subtitle_file_includes_original_audio_subtitles(self):
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
task_dir = temp_path / "task"
task_dir.mkdir()
video_path = temp_path / "episode.mp4"
subtitle_path = temp_path / "episode.srt"
video_path.write_bytes(b"video")
subtitle_path.write_text(
"1\n00:00:05,000 --> 00:00:06,500\n原片对白\n",
encoding="utf-8",
)
params = VideoClipParams(video_origin_path=str(video_path), subtitle_enabled=True)
draft_script = jianying_task._build_jianying_draft_script(
[
{
"_id": 1,
"timestamp": "00:00:05,000-00:00:07,000",
"narration": "播放原片1",
"OST": 1,
}
],
params,
[],
)
with (
patch.object(jianying_task.utils, "subtitle_dir", return_value=str(temp_path)),
patch.object(jianying_task.utils, "task_dir", return_value=str(task_dir)),
):
output_path = jianying_task._create_jianying_subtitle_file(
"task-id",
draft_script,
params,
)
self.assertTrue(output_path)
self.assertIn("原片对白", Path(output_path).read_text(encoding="utf-8"))
def test_start_export_jianying_draft_does_not_clip_video(self):
with tempfile.TemporaryDirectory() as temp_dir:
root_path = Path(temp_dir) / "drafts"
task_dir = Path(temp_dir) / "task"
root_path.mkdir()
task_dir.mkdir()
video_path = Path(temp_dir) / "source.mp4"
audio_path = task_dir / "audio.mp3"
script_path = Path(temp_dir) / "script.json"
subtitle_path = task_dir / "script_subtitles.srt"
video_path.write_bytes(b"video")
audio_path.write_bytes(b"audio")
script_path.write_text(
json.dumps([
{
"_id": 1,
"timestamp": "00:00:01,000-00:00:03,000",
"narration": "测试解说",
"OST": 0,
}
], ensure_ascii=False),
encoding="utf-8",
)
params = VideoClipParams(
video_clip_json_path=str(script_path),
video_origin_path=str(video_path),
tts_engine="edge_tts",
voice_name="zh-CN-YunjianNeural",
subtitle_enabled=True,
draft_name="NarratoAI_no_clip",
)
tts_results = [
{
"_id": 1,
"timestamp": "00:00:01,000-00:00:03,000",
"audio_file": str(audio_path),
"subtitle_file": "",
"duration": 1.5,
}
]
with (
patch.dict(jianying_task.config.ui, {"jianying_draft_path": str(root_path)}, clear=False),
patch.object(jianying_task.utils, "task_dir", return_value=str(task_dir)),
patch.object(jianying_task.voice, "tts_multiple", return_value=tts_results),
patch.object(jianying_task, "_create_jianying_subtitle_file", return_value=str(subtitle_path)),
patch.object(jianying_task, "write_plaintext_jianying_draft", return_value=(str(root_path / "draft"), "NarratoAI_no_clip")) as write_draft,
patch.object(jianying_task.clip_video, "clip_video_unified") as clip_video_unified,
):
result = jianying_task.start_export_jianying_draft("task-id", params)
clip_video_unified.assert_not_called()
write_kwargs = write_draft.call_args.kwargs
self.assertTrue(write_kwargs["new_script_list"][0]["use_source_timerange"])
self.assertEqual(str(audio_path), write_kwargs["new_script_list"][0]["audio"])
self.assertEqual(str(subtitle_path), write_kwargs["subtitle_path"])
self.assertEqual(str(subtitle_path), result["subtitles"][0])
if __name__ == "__main__":
unittest.main()