| 1 | import unittest |
| 2 | import os |
| 3 | import shutil |
| 4 | import sys |
| 5 | import tempfile |
| 6 | from pathlib import Path |
| 7 | from unittest.mock import patch |
| 8 | |
| 9 | # add project root to python path |
| 10 | sys.path.insert(0, str(Path(__file__).parent.parent.parent)) |
| 11 | |
| 12 | from app.services import task as tm |
| 13 | from app.models.schema import MaterialInfo, VideoParams |
| 14 | from app.utils import utils |
| 15 | |
| 16 | resources_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "resources") |
| 17 | RUN_INTEGRATION_TESTS = os.environ.get("MPT_RUN_INTEGRATION_TESTS", "").lower() in { |
| 18 | "1", |
| 19 | "true", |
| 20 | "yes", |
| 21 | } |
| 22 | |
| 23 | class TestTaskService(unittest.TestCase): |
| 24 | def setUp(self): |
| 25 | pass |
| 26 | |
| 27 | def tearDown(self): |
| 28 | pass |
| 29 | |
| 30 | def test_generate_script_forwards_advanced_prompt_options(self): |
| 31 | """ |
| 32 | 任务生成入口和 WebUI/API 共用 VideoParams。这里验证自动生成文案时， |
| 33 | 高级提示词参数会继续传到 LLM 服务层，避免只在 /scripts 接口生效。 |
| 34 | """ |
| 35 | params = VideoParams( |
| 36 | video_subject="咖啡", |
| 37 | video_script="", |
| 38 | video_language="zh-CN", |
| 39 | paragraph_number=2, |
| 40 | video_script_prompt="语气轻松", |
| 41 | custom_system_prompt="Only write short narration.", |
| 42 | ) |
| 43 | |
| 44 | with patch.object(tm.llm, "generate_script", return_value="生成的文案") as generate: |
| 45 | result = tm.generate_script("task-id", params) |
| 46 | |
| 47 | self.assertEqual(result, "生成的文案") |
| 48 | generate.assert_called_once_with( |
| 49 | video_subject="咖啡", |
| 50 | language="zh-CN", |
| 51 | paragraph_number=2, |
| 52 | video_script_prompt="语气轻松", |
| 53 | custom_system_prompt="Only write short narration.", |
| 54 | ) |
| 55 | |
| 56 | def test_generate_terms_uses_script_order_mode_when_enabled(self): |
| 57 | """ |
| 58 | 默认模式不受影响；只有用户显式开启素材按文案顺序匹配时，任务层才 |
| 59 | 要求 LLM 生成有序关键词，并适当增加关键词数量以覆盖更多脚本片段。 |
| 60 | """ |
| 61 | params = VideoParams( |
| 62 | video_subject="城市通勤", |
| 63 | video_script="", |
| 64 | match_materials_to_script=True, |
| 65 | ) |
| 66 | |
| 67 | with patch.object(tm.llm, "generate_terms", return_value=["city", "train"]) as generate: |
| 68 | result = tm.generate_terms("task-id", params, "先城市，再地铁") |
| 69 | |
| 70 | self.assertEqual(result, ["city", "train"]) |
| 71 | generate.assert_called_once_with( |
| 72 | video_subject="城市通勤", |
| 73 | video_script="先城市，再地铁", |
| 74 | amount=8, |
| 75 | match_script_order=True, |
| 76 | ) |
| 77 | |
| 78 | def test_generate_audio_uses_custom_file_inside_task_directory(self): |
| 79 | task_id = "test-custom-audio-safe" |
| 80 | task_dir = utils.task_dir(task_id) |
| 81 | custom_audio_file = os.path.join(task_dir, "custom-audio.mp3") |
| 82 | with open(custom_audio_file, "wb") as audio: |
| 83 | audio.write(b"fake audio") |
| 84 | |
| 85 | params = VideoParams( |
| 86 | video_subject="custom audio", |
| 87 | video_script="", |
| 88 | custom_audio_file=custom_audio_file, |
| 89 | voice_name="test-voice", |
| 90 | ) |
| 91 | |
| 92 | try: |
| 93 | with ( |
| 94 | patch.object(tm.voice, "tts") as tts, |
| 95 | patch.object(tm.voice, "get_audio_duration", return_value=7), |
| 96 | ): |
| 97 | audio_file, audio_duration, sub_maker = tm.generate_audio( |
| 98 | task_id, params, "script" |
| 99 | ) |
| 100 | finally: |
| 101 | shutil.rmtree(task_dir, ignore_errors=True) |
| 102 | |
| 103 | self.assertEqual(audio_file, os.path.realpath(custom_audio_file)) |
| 104 | self.assertEqual(audio_duration, 7) |
| 105 | self.assertIsNone(sub_maker) |
| 106 | tts.assert_not_called() |
| 107 | |
| 108 | def test_generate_audio_accepts_server_side_custom_file(self): |
| 109 | task_id = "test-custom-audio-server-side" |
| 110 | task_dir = utils.task_dir(task_id) |
| 111 | |
| 112 | with tempfile.NamedTemporaryFile(suffix=".mp3") as server_audio: |
| 113 | server_audio.write(b"fake audio") |
| 114 | server_audio.flush() |
| 115 | params = VideoParams( |
| 116 | video_subject="custom audio", |
| 117 | video_script="", |
| 118 | custom_audio_file=server_audio.name, |
| 119 | voice_name="test-voice", |
| 120 | ) |
| 121 | |
| 122 | try: |
| 123 | with ( |
| 124 | patch.object(tm.voice, "tts") as tts, |
| 125 | patch.object(tm.voice, "get_audio_duration", return_value=6), |
| 126 | ): |
| 127 | audio_file, audio_duration, result_sub_maker = tm.generate_audio( |
| 128 | task_id, params, "script" |
| 129 | ) |
| 130 | finally: |
| 131 | shutil.rmtree(task_dir, ignore_errors=True) |
| 132 | |
| 133 | self.assertEqual(audio_file, os.path.realpath(server_audio.name)) |
| 134 | self.assertEqual(audio_duration, 6) |
| 135 | self.assertIsNone(result_sub_maker) |
| 136 | tts.assert_not_called() |
| 137 | |
| 138 | def test_generate_audio_rejects_missing_custom_file_without_tts(self): |
| 139 | task_id = "test-custom-audio-missing" |
| 140 | task_dir = utils.task_dir(task_id) |
| 141 | missing_audio_file = os.path.join(task_dir, "missing.mp3") |
| 142 | params = VideoParams( |
| 143 | video_subject="custom audio", |
| 144 | video_script="", |
| 145 | custom_audio_file=missing_audio_file, |
| 146 | voice_name="test-voice", |
| 147 | ) |
| 148 | |
| 149 | try: |
| 150 | with ( |
| 151 | patch.object(tm.voice, "tts") as tts, |
| 152 | patch.object(tm.sm.state, "update_task") as update_task, |
| 153 | ): |
| 154 | audio_file, audio_duration, result_sub_maker = tm.generate_audio( |
| 155 | task_id, params, "script" |
| 156 | ) |
| 157 | finally: |
| 158 | shutil.rmtree(task_dir, ignore_errors=True) |
| 159 | |
| 160 | self.assertIsNone(audio_file) |
| 161 | self.assertIsNone(audio_duration) |
| 162 | self.assertIsNone(result_sub_maker) |
| 163 | tts.assert_not_called() |
| 164 | update_task.assert_called_with(task_id, state=tm.const.TASK_STATE_FAILED) |
| 165 | |
| 166 | def test_generate_subtitle_uses_whisper_for_custom_audio_without_sub_maker(self): |
| 167 | """ |
| 168 | 自定义音频不会经过 TTS，所以没有 sub_maker。 |
| 169 | Whisper 可以直接从音频文件转写，此时不能被 sub_maker 为空的保护逻辑提前跳过。 |
| 170 | """ |
| 171 | task_id = "test-custom-audio-whisper-subtitle" |
| 172 | task_dir = utils.task_dir(task_id) |
| 173 | audio_file = os.path.join(task_dir, "custom-audio.mp3") |
| 174 | Path(audio_file).write_bytes(b"fake audio") |
| 175 | params = VideoParams( |
| 176 | video_subject="custom audio", |
| 177 | video_script="Hello world.", |
| 178 | subtitle_enabled=True, |
| 179 | ) |
| 180 | |
| 181 | def fake_whisper_create(audio_file, subtitle_file): |
| 182 | Path(subtitle_file).write_text( |
| 183 | "1\n00:00:00,000 --> 00:00:01,000\nHello world.\n\n", |
| 184 | encoding="utf-8", |
| 185 | ) |
| 186 | |
| 187 | try: |
| 188 | with ( |
| 189 | patch.object( |
| 190 | tm.config, |
| 191 | "app", |
| 192 | dict(tm.config.app, subtitle_provider="whisper"), |
| 193 | ), |
| 194 | patch.object( |
| 195 | tm.subtitle, "create", side_effect=fake_whisper_create |
| 196 | ) as create, |
| 197 | patch.object(tm.subtitle, "correct") as correct, |
| 198 | ): |
| 199 | subtitle_path = tm.generate_subtitle( |
| 200 | task_id=task_id, |
| 201 | params=params, |
| 202 | video_script="Hello world.", |
| 203 | sub_maker=None, |
| 204 | audio_file=audio_file, |
| 205 | ) |
| 206 | finally: |
| 207 | shutil.rmtree(task_dir, ignore_errors=True) |
| 208 | |
| 209 | self.assertTrue(subtitle_path.endswith("subtitle.srt")) |
| 210 | create.assert_called_once_with(audio_file=audio_file, subtitle_file=subtitle_path) |
| 211 | correct.assert_called_once_with( |
| 212 | subtitle_file=subtitle_path, video_script="Hello world." |
| 213 | ) |
| 214 | |
| 215 | def test_generate_subtitle_skips_edge_provider_without_sub_maker(self): |
| 216 | """ |
| 217 | Edge 字幕依赖 TTS 返回的 sub_maker 时间轴。 |
| 218 | 自定义音频缺少该对象时应继续跳过，避免产生不可信的字幕时间轴。 |
| 219 | """ |
| 220 | task_id = "test-custom-audio-edge-no-submaker" |
| 221 | task_dir = utils.task_dir(task_id) |
| 222 | audio_file = os.path.join(task_dir, "custom-audio.mp3") |
| 223 | Path(audio_file).write_bytes(b"fake audio") |
| 224 | params = VideoParams( |
| 225 | video_subject="custom audio", |
| 226 | video_script="Hello world.", |
| 227 | subtitle_enabled=True, |
| 228 | ) |
| 229 | |
| 230 | try: |
| 231 | with ( |
| 232 | patch.object( |
| 233 | tm.config, |
| 234 | "app", |
| 235 | dict(tm.config.app, subtitle_provider="edge"), |
| 236 | ), |
| 237 | patch.object(tm.voice, "create_subtitle") as create_subtitle, |
| 238 | patch.object(tm.subtitle, "create") as whisper_create, |
| 239 | ): |
| 240 | subtitle_path = tm.generate_subtitle( |
| 241 | task_id=task_id, |
| 242 | params=params, |
| 243 | video_script="Hello world.", |
| 244 | sub_maker=None, |
| 245 | audio_file=audio_file, |
| 246 | ) |
| 247 | finally: |
| 248 | shutil.rmtree(task_dir, ignore_errors=True) |
| 249 | |
| 250 | self.assertEqual(subtitle_path, "") |
| 251 | create_subtitle.assert_not_called() |
| 252 | whisper_create.assert_not_called() |
| 253 | |
| 254 | @unittest.skipUnless( |
| 255 | RUN_INTEGRATION_TESTS, |
| 256 | "MPT_RUN_INTEGRATION_TESTS not set", |
| 257 | ) |
| 258 | def test_task_local_materials(self): |
| 259 | task_id = "00000000-0000-0000-0000-000000000000" |
| 260 | video_materials=[] |
| 261 | for i in range(1, 4): |
| 262 | video_materials.append(MaterialInfo( |
| 263 | provider="local", |
| 264 | url=os.path.join(resources_dir, f"{i}.png"), |
| 265 | duration=0 |
| 266 | )) |
| 267 | |
| 268 | params = VideoParams( |
| 269 | video_subject="金钱的作用", |
| 270 | video_script="金钱不仅是交换媒介，更是社会资源的分配工具。它能满足基本生存需求，如食物和住房，也能提供教育、医疗等提升生活品质的机会。拥有足够的金钱意味着更多选择权，比如职业自由或创业可能。但金钱的作用也有边界，它无法直接购买幸福、健康或真诚的人际关系。过度追逐财富可能导致价值观扭曲，忽视精神层面的需求。理想的状态是理性看待金钱，将其作为实现目标的工具而非终极目的。", |
| 271 | video_terms="money importance, wealth and society, financial freedom, money and happiness, role of money", |
| 272 | video_aspect="9:16", |
| 273 | video_concat_mode="random", |
| 274 | video_transition_mode="None", |
| 275 | video_clip_duration=3, |
| 276 | video_count=1, |
| 277 | video_source="local", |
| 278 | video_materials=video_materials, |
| 279 | video_language="", |
| 280 | voice_name="zh-CN-XiaoxiaoNeural-Female", |
| 281 | voice_volume=1.0, |
| 282 | voice_rate=1.0, |
| 283 | bgm_type="random", |
| 284 | bgm_file="", |
| 285 | bgm_volume=0.2, |
| 286 | subtitle_enabled=True, |
| 287 | subtitle_position="bottom", |
| 288 | custom_position=70.0, |
| 289 | font_name="MicrosoftYaHeiBold.ttc", |
| 290 | text_fore_color="#FFFFFF", |
| 291 | text_background_color=True, |
| 292 | font_size=60, |
| 293 | stroke_color="#000000", |
| 294 | stroke_width=1.5, |
| 295 | n_threads=2, |
| 296 | paragraph_number=1 |
| 297 | ) |
| 298 | result = tm.start(task_id=task_id, params=params) |
| 299 | print(result) |
| 300 | |
| 301 | |
| 302 | if __name__ == "__main__": |
| 303 | unittest.main() |
| 304 |