| 1 | import math |
| 2 | import os.path |
| 3 | import re |
| 4 | from os import path |
| 5 | |
| 6 | from loguru import logger |
| 7 | |
| 8 | from app.config import config |
| 9 | from app.models import const |
| 10 | from app.models.schema import VideoConcatMode, VideoParams |
| 11 | from app.services import llm, material, subtitle, twelvelabs, video, voice, upload_post |
| 12 | from app.services import state as sm |
| 13 | from app.utils import file_security, utils |
| 14 | |
| 15 | |
| 16 | def generate_script(task_id, params): |
| 17 | logger.info("\n\n## generating video script") |
| 18 | video_script = params.video_script.strip() |
| 19 | if not video_script: |
| 20 | video_script = llm.generate_script( |
| 21 | video_subject=params.video_subject, |
| 22 | language=params.video_language, |
| 23 | paragraph_number=params.paragraph_number, |
| 24 | video_script_prompt=params.video_script_prompt, |
| 25 | custom_system_prompt=params.custom_system_prompt, |
| 26 | ) |
| 27 | else: |
| 28 | logger.debug(f"video script: \n{video_script}") |
| 29 | |
| 30 | if not video_script: |
| 31 | sm.state.update_task(task_id, state=const.TASK_STATE_FAILED) |
| 32 | logger.error("failed to generate video script.") |
| 33 | return None |
| 34 | |
| 35 | return video_script |
| 36 | |
| 37 | |
| 38 | def generate_terms(task_id, params, video_script): |
| 39 | logger.info("\n\n## generating video terms") |
| 40 | video_terms = params.video_terms |
| 41 | if not video_terms: |
| 42 | # 开启素材按文案顺序匹配后，关键词本身也必须按脚本叙事顺序生成； |
| 43 | # 否则后续即使顺序下载和顺序拼接，也只能复用一组全局主题词， |
| 44 | # 无法改善“后面内容的画面提前出现”的问题。 |
| 45 | video_terms = llm.generate_terms( |
| 46 | video_subject=params.video_subject, |
| 47 | video_script=video_script, |
| 48 | amount=8 if params.match_materials_to_script else 5, |
| 49 | match_script_order=params.match_materials_to_script, |
| 50 | ) |
| 51 | else: |
| 52 | if isinstance(video_terms, str): |
| 53 | video_terms = [term.strip() for term in re.split(r"[,，]", video_terms)] |
| 54 | elif isinstance(video_terms, list): |
| 55 | video_terms = [term.strip() for term in video_terms] |
| 56 | else: |
| 57 | raise ValueError("video_terms must be a string or a list of strings.") |
| 58 | |
| 59 | logger.debug(f"video terms: {utils.to_json(video_terms)}") |
| 60 | |
| 61 | if not video_terms: |
| 62 | sm.state.update_task(task_id, state=const.TASK_STATE_FAILED) |
| 63 | logger.error("failed to generate video terms.") |
| 64 | return None |
| 65 | |
| 66 | # 可选的 TwelveLabs Marengo 语义重排：未启用时返回原顺序，无任何副作用。 |
| 67 | # 顺序匹配模式下关键词顺序本身就是脚本叙事顺序，必须保持原样，故跳过。 |
| 68 | if not params.match_materials_to_script: |
| 69 | video_terms = twelvelabs.rerank_terms_by_subject( |
| 70 | video_subject=params.video_subject, |
| 71 | search_terms=video_terms, |
| 72 | ) |
| 73 | |
| 74 | return video_terms |
| 75 | |
| 76 | |
| 77 | def save_script_data(task_id, video_script, video_terms, params): |
| 78 | script_file = path.join(utils.task_dir(task_id), "script.json") |
| 79 | script_data = { |
| 80 | "script": video_script, |
| 81 | "search_terms": video_terms, |
| 82 | "params": params, |
| 83 | } |
| 84 | |
| 85 | with open(script_file, "w", encoding="utf-8") as f: |
| 86 | f.write(utils.to_json(script_data)) |
| 87 | |
| 88 | |
| 89 | def resolve_custom_audio_file(task_id: str, custom_audio_file: str | None) -> str: |
| 90 | requested_file = (custom_audio_file or "").strip() |
| 91 | if not requested_file: |
| 92 | return "" |
| 93 | |
| 94 | task_dir = utils.task_dir(task_id) |
| 95 | try: |
| 96 | return file_security.resolve_path_within_directory( |
| 97 | task_dir, |
| 98 | requested_file, |
| 99 | ) |
| 100 | except ValueError as exc: |
| 101 | task_dir_error = exc |
| 102 | |
| 103 | server_audio_file = path.realpath( |
| 104 | requested_file |
| 105 | if path.isabs(requested_file) |
| 106 | else path.join(utils.root_dir(), requested_file) |
| 107 | ) |
| 108 | if not path.isabs(requested_file): |
| 109 | project_root = path.realpath(utils.root_dir()) |
| 110 | try: |
| 111 | if path.commonpath([project_root, server_audio_file]) != project_root: |
| 112 | raise ValueError( |
| 113 | "relative custom audio paths must stay within the project directory" |
| 114 | ) |
| 115 | except ValueError as exc: |
| 116 | raise ValueError( |
| 117 | "custom audio file must be task-local or an existing server-side file" |
| 118 | ) from exc |
| 119 | |
| 120 | if not path.isfile(server_audio_file): |
| 121 | raise ValueError( |
| 122 | "custom audio file does not exist or is not a file" |
| 123 | ) from task_dir_error |
| 124 | |
| 125 | return server_audio_file |
| 126 | |
| 127 | |
| 128 | def generate_audio(task_id, params, video_script): |
| 129 | ''' |
| 130 | Generate audio for the video script. |
| 131 | If a custom audio file is provided, it will be used directly. |
| 132 | There will be no subtitle maker object returned in this case. |
| 133 | Otherwise, TTS will be used to generate the audio. |
| 134 | Returns: |
| 135 | - audio_file: path to the generated or provided audio file |
| 136 | - audio_duration: duration of the audio in seconds |
| 137 | - sub_maker: subtitle maker object if TTS is used, None otherwise |
| 138 | ''' |
| 139 | logger.info("\n\n## generating audio") |
| 140 | # /audio 和 /subtitle 请求模型不包含 custom_audio_file， |
| 141 | # 这里统一做兼容读取，避免直调接口时抛属性错误。 |
| 142 | requested_custom_audio_file = getattr(params, "custom_audio_file", None) |
| 143 | try: |
| 144 | custom_audio_file = resolve_custom_audio_file( |
| 145 | task_id, requested_custom_audio_file |
| 146 | ) |
| 147 | except ValueError as exc: |
| 148 | logger.error( |
| 149 | "custom audio file is invalid, " |
| 150 | f"task_id: {task_id}, path: {requested_custom_audio_file}, error: {str(exc)}" |
| 151 | ) |
| 152 | sm.state.update_task(task_id, state=const.TASK_STATE_FAILED) |
| 153 | return None, None, None |
| 154 | |
| 155 | if not custom_audio_file: |
| 156 | logger.info("no custom audio file provided, using TTS to generate audio.") |
| 157 | audio_file = path.join(utils.task_dir(task_id), "audio.mp3") |
| 158 | sub_maker = voice.tts( |
| 159 | text=video_script, |
| 160 | voice_name=voice.parse_voice_name(params.voice_name), |
| 161 | voice_rate=params.voice_rate, |
| 162 | voice_file=audio_file, |
| 163 | ) |
| 164 | if sub_maker is None: |
| 165 | sm.state.update_task(task_id, state=const.TASK_STATE_FAILED) |
| 166 | logger.error( |
| 167 | """failed to generate audio: |
| 168 | 1. check if the language of the voice matches the language of the video script. |
| 169 | 2. check if the network is available. If you are in China, it is recommended to use a VPN and enable the global traffic mode. |
| 170 | """.strip() |
| 171 | ) |
| 172 | return None, None, None |
| 173 | audio_duration = math.ceil(voice.get_audio_duration(sub_maker)) |
| 174 | if audio_duration == 0: |
| 175 | sm.state.update_task(task_id, state=const.TASK_STATE_FAILED) |
| 176 | logger.error("failed to get audio duration.") |
| 177 | return None, None, None |
| 178 | return audio_file, audio_duration, sub_maker |
| 179 | else: |
| 180 | logger.info(f"using custom audio file: {custom_audio_file}") |
| 181 | audio_duration = voice.get_audio_duration(custom_audio_file) |
| 182 | if audio_duration == 0: |
| 183 | sm.state.update_task(task_id, state=const.TASK_STATE_FAILED) |
| 184 | logger.error("failed to get audio duration from custom audio file.") |
| 185 | return None, None, None |
| 186 | return custom_audio_file, audio_duration, None |
| 187 | |
| 188 | def generate_subtitle(task_id, params, video_script, sub_maker, audio_file): |
| 189 | ''' |
| 190 | Generate subtitle for the video script. |
| 191 | If subtitle generation is disabled or no subtitle maker is provided, it will return an empty string. |
| 192 | Otherwise, it will generate the subtitle using the specified provider. |
| 193 | Returns: |
| 194 | - subtitle_path: path to the generated subtitle file |
| 195 | ''' |
| 196 | logger.info("\n\n## generating subtitle") |
| 197 | if not params.subtitle_enabled: |
| 198 | return "" |
| 199 | |
| 200 | subtitle_path = path.join(utils.task_dir(task_id), "subtitle.srt") |
| 201 | subtitle_provider = config.app.get("subtitle_provider", "edge").strip().lower() |
| 202 | logger.info(f"\n\n## generating subtitle, provider: {subtitle_provider}") |
| 203 | |
| 204 | if sub_maker is None and subtitle_provider != "whisper": |
| 205 | # 自定义音频不会经过 TTS，因此没有 Edge/Azure 等 TTS 返回的 |
| 206 | # sub_maker 时间轴。只有 Whisper 可以直接从音频文件转写字幕； |
| 207 | # 其他字幕提供方继续保持原有行为，避免生成错误的空时间轴。 |
| 208 | logger.warning( |
| 209 | "subtitle maker is missing, skip subtitle generation for provider: " |
| 210 | f"{subtitle_provider}" |
| 211 | ) |
| 212 | return "" |
| 213 | |
| 214 | subtitle_fallback = False |
| 215 | if subtitle_provider == "edge": |
| 216 | voice.create_subtitle( |
| 217 | text=video_script, sub_maker=sub_maker, subtitle_file=subtitle_path |
| 218 | ) |
| 219 | if not os.path.exists(subtitle_path): |
| 220 | subtitle_fallback = True |
| 221 | logger.warning("subtitle file not found, fallback to whisper") |
| 222 | |
| 223 | if subtitle_provider == "whisper" or subtitle_fallback: |
| 224 | subtitle.create(audio_file=audio_file, subtitle_file=subtitle_path) |
| 225 | logger.info("\n\n## correcting subtitle") |
| 226 | subtitle.correct(subtitle_file=subtitle_path, video_script=video_script) |
| 227 | |
| 228 | subtitle_lines = subtitle.file_to_subtitles(subtitle_path) |
| 229 | if not subtitle_lines: |
| 230 | logger.warning(f"subtitle file is invalid: {subtitle_path}") |
| 231 | return "" |
| 232 | |
| 233 | return subtitle_path |
| 234 | |
| 235 | |
| 236 | def get_video_materials(task_id, params, video_terms, audio_duration): |
| 237 | if params.video_source == "local": |
| 238 | logger.info("\n\n## preprocess local materials") |
| 239 | materials = video.preprocess_video( |
| 240 | materials=params.video_materials, clip_duration=params.video_clip_duration |
| 241 | ) |
| 242 | if not materials: |
| 243 | sm.state.update_task(task_id, state=const.TASK_STATE_FAILED) |
| 244 | logger.error( |
| 245 | "no valid materials found, please check the materials and try again." |
| 246 | ) |
| 247 | return None |
| 248 | return [material_info.url for material_info in materials] |
| 249 | else: |
| 250 | logger.info(f"\n\n## downloading videos from {params.video_source}") |
| 251 | # 顺序匹配模式只在用户显式开启时生效。这里强制素材下载按关键词顺序 |
| 252 | # 轮询，避免某个早期关键词下载太多素材，把后续脚本主题挤出最终时间线。 |
| 253 | downloaded_videos = material.download_videos( |
| 254 | task_id=task_id, |
| 255 | search_terms=video_terms, |
| 256 | source=params.video_source, |
| 257 | video_aspect=params.video_aspect, |
| 258 | video_concat_mode=( |
| 259 | VideoConcatMode.sequential |
| 260 | if params.match_materials_to_script |
| 261 | else params.video_concat_mode |
| 262 | ), |
| 263 | audio_duration=audio_duration * params.video_count, |
| 264 | max_clip_duration=params.video_clip_duration, |
| 265 | match_script_order=params.match_materials_to_script, |
| 266 | ) |
| 267 | if not downloaded_videos: |
| 268 | sm.state.update_task(task_id, state=const.TASK_STATE_FAILED) |
| 269 | logger.error( |
| 270 | "failed to download videos, maybe the network is not available. if you are in China, please use a VPN." |
| 271 | ) |
| 272 | return None |
| 273 | return downloaded_videos |
| 274 | |
| 275 | |
| 276 | def generate_final_videos( |
| 277 | task_id, params, downloaded_videos, audio_file, subtitle_path |
| 278 | ): |
| 279 | final_video_paths = [] |
| 280 | combined_video_paths = [] |
| 281 | # 多视频生成默认会打散素材以增加差异；但“按文案顺序匹配素材”追求的是 |
| 282 | # 时间线稳定性和可解释性，所以开启后所有输出都使用顺序拼接。 |
| 283 | if params.match_materials_to_script: |
| 284 | video_concat_mode = VideoConcatMode.sequential |
| 285 | elif params.video_count == 1: |
| 286 | video_concat_mode = params.video_concat_mode |
| 287 | else: |
| 288 | video_concat_mode = VideoConcatMode.random |
| 289 | video_transition_mode = params.video_transition_mode |
| 290 | |
| 291 | _progress = 50 |
| 292 | for i in range(params.video_count): |
| 293 | index = i + 1 |
| 294 | combined_video_path = path.join( |
| 295 | utils.task_dir(task_id), f"combined-{index}.mp4" |
| 296 | ) |
| 297 | logger.info(f"\n\n## combining video: {index} => {combined_video_path}") |
| 298 | video.combine_videos( |
| 299 | combined_video_path=combined_video_path, |
| 300 | video_paths=downloaded_videos, |
| 301 | audio_file=audio_file, |
| 302 | video_aspect=params.video_aspect, |
| 303 | video_concat_mode=video_concat_mode, |
| 304 | video_transition_mode=video_transition_mode, |
| 305 | max_clip_duration=params.video_clip_duration, |
| 306 | threads=params.n_threads, |
| 307 | ) |
| 308 | |
| 309 | _progress += 50 / params.video_count / 2 |
| 310 | sm.state.update_task(task_id, progress=_progress) |
| 311 | |
| 312 | final_video_path = path.join(utils.task_dir(task_id), f"final-{index}.mp4") |
| 313 | |
| 314 | logger.info(f"\n\n## generating video: {index} => {final_video_path}") |
| 315 | video.generate_video( |
| 316 | video_path=combined_video_path, |
| 317 | audio_path=audio_file, |
| 318 | subtitle_path=subtitle_path, |
| 319 | output_file=final_video_path, |
| 320 | params=params, |
| 321 | ) |
| 322 | |
| 323 | _progress += 50 / params.video_count / 2 |
| 324 | sm.state.update_task(task_id, progress=_progress) |
| 325 | |
| 326 | final_video_paths.append(final_video_path) |
| 327 | combined_video_paths.append(combined_video_path) |
| 328 | |
| 329 | return final_video_paths, combined_video_paths |
| 330 | |
| 331 | |
| 332 | def start(task_id, params: VideoParams, stop_at: str = "video"): |
| 333 | logger.info(f"start task: {task_id}, stop_at: {stop_at}") |
| 334 | sm.state.update_task(task_id, state=const.TASK_STATE_PROCESSING, progress=5) |
| 335 | |
| 336 | # 1. Generate script |
| 337 | video_script = generate_script(task_id, params) |
| 338 | if not video_script or "Error: " in video_script: |
| 339 | sm.state.update_task(task_id, state=const.TASK_STATE_FAILED) |
| 340 | return |
| 341 | |
| 342 | sm.state.update_task(task_id, state=const.TASK_STATE_PROCESSING, progress=10) |
| 343 | |
| 344 | if stop_at == "script": |
| 345 | sm.state.update_task( |
| 346 | task_id, state=const.TASK_STATE_COMPLETE, progress=100, script=video_script |
| 347 | ) |
| 348 | return {"script": video_script} |
| 349 | |
| 350 | # 2. Generate terms |
| 351 | video_terms = "" |
| 352 | if params.video_source != "local": |
| 353 | video_terms = generate_terms(task_id, params, video_script) |
| 354 | if not video_terms: |
| 355 | sm.state.update_task(task_id, state=const.TASK_STATE_FAILED) |
| 356 | return |
| 357 | |
| 358 | save_script_data(task_id, video_script, video_terms, params) |
| 359 | |
| 360 | if stop_at == "terms": |
| 361 | sm.state.update_task( |
| 362 | task_id, state=const.TASK_STATE_COMPLETE, progress=100, terms=video_terms |
| 363 | ) |
| 364 | return {"script": video_script, "terms": video_terms} |
| 365 | |
| 366 | sm.state.update_task(task_id, state=const.TASK_STATE_PROCESSING, progress=20) |
| 367 | |
| 368 | # 3. Generate audio |
| 369 | audio_file, audio_duration, sub_maker = generate_audio( |
| 370 | task_id, params, video_script |
| 371 | ) |
| 372 | if not audio_file: |
| 373 | sm.state.update_task(task_id, state=const.TASK_STATE_FAILED) |
| 374 | return |
| 375 | |
| 376 | sm.state.update_task(task_id, state=const.TASK_STATE_PROCESSING, progress=30) |
| 377 | |
| 378 | if stop_at == "audio": |
| 379 | sm.state.update_task( |
| 380 | task_id, |
| 381 | state=const.TASK_STATE_COMPLETE, |
| 382 | progress=100, |
| 383 | audio_file=audio_file, |
| 384 | ) |
| 385 | return {"audio_file": audio_file, "audio_duration": audio_duration} |
| 386 | |
| 387 | # 4. Generate subtitle |
| 388 | subtitle_path = generate_subtitle( |
| 389 | task_id, params, video_script, sub_maker, audio_file |
| 390 | ) |
| 391 | |
| 392 | if stop_at == "subtitle": |
| 393 | sm.state.update_task( |
| 394 | task_id, |
| 395 | state=const.TASK_STATE_COMPLETE, |
| 396 | progress=100, |
| 397 | subtitle_path=subtitle_path, |
| 398 | ) |
| 399 | return {"subtitle_path": subtitle_path} |
| 400 | |
| 401 | sm.state.update_task(task_id, state=const.TASK_STATE_PROCESSING, progress=40) |
| 402 | |
| 403 | # 5. Get video materials |
| 404 | downloaded_videos = get_video_materials( |
| 405 | task_id, params, video_terms, audio_duration |
| 406 | ) |
| 407 | if not downloaded_videos: |
| 408 | sm.state.update_task(task_id, state=const.TASK_STATE_FAILED) |
| 409 | return |
| 410 | |
| 411 | if stop_at == "materials": |
| 412 | sm.state.update_task( |
| 413 | task_id, |
| 414 | state=const.TASK_STATE_COMPLETE, |
| 415 | progress=100, |
| 416 | materials=downloaded_videos, |
| 417 | ) |
| 418 | return {"materials": downloaded_videos} |
| 419 | |
| 420 | sm.state.update_task(task_id, state=const.TASK_STATE_PROCESSING, progress=50) |
| 421 | |
| 422 | # 仅完整视频生成流程才需要处理视频拼接模式； |
| 423 | # 这样可以避免 /subtitle 和 /audio 这类请求访问不存在的字段。 |
| 424 | if type(params.video_concat_mode) is str: |
| 425 | params.video_concat_mode = VideoConcatMode(params.video_concat_mode) |
| 426 | |
| 427 | # 6. Generate final videos |
| 428 | final_video_paths, combined_video_paths = generate_final_videos( |
| 429 | task_id, params, downloaded_videos, audio_file, subtitle_path |
| 430 | ) |
| 431 | |
| 432 | if not final_video_paths: |
| 433 | sm.state.update_task(task_id, state=const.TASK_STATE_FAILED) |
| 434 | return |
| 435 | |
| 436 | logger.success( |
| 437 | f"task {task_id} finished, generated {len(final_video_paths)} videos." |
| 438 | ) |
| 439 | |
| 440 | # 7. Cross-post to social platforms (if enabled) |
| 441 | cross_post_results = [] |
| 442 | if upload_post.upload_post_service.is_configured() and upload_post.upload_post_service.auto_upload: |
| 443 | platforms = upload_post.upload_post_service.platforms |
| 444 | logger.info(f"\n\n## cross-posting videos to {', '.join(platforms)}") |
| 445 | |
| 446 | youtube_extra = None |
| 447 | if any(p.startswith("youtube") for p in platforms): |
| 448 | metadata = llm.generate_social_metadata( |
| 449 | video_subject=params.video_subject, |
| 450 | video_script=video_script, |
| 451 | language=params.video_language or "", |
| 452 | platform="youtube_shorts", |
| 453 | ) |
| 454 | youtube_extra = { |
| 455 | "youtube_title": metadata.get("title", params.video_subject), |
| 456 | "youtube_description": metadata.get("caption", ""), |
| 457 | "tags": metadata.get("hashtags", []), |
| 458 | "privacyStatus": upload_post.upload_post_service.youtube_privacy_status, |
| 459 | "containsSyntheticMedia": True, |
| 460 | } |
| 461 | |
| 462 | for video_path in final_video_paths: |
| 463 | result = upload_post.cross_post_video( |
| 464 | video_path=video_path, |
| 465 | title=params.video_subject or "Check out this video! #shorts #viral", |
| 466 | youtube_extra=youtube_extra, |
| 467 | ) |
| 468 | cross_post_results.append(result) |
| 469 | if result.get('success'): |
| 470 | logger.info(f"✅ Cross-posted: {video_path}") |
| 471 | else: |
| 472 | logger.warning(f"⚠️ Failed to cross-post: {video_path} - {result.get('error', 'Unknown error')}") |
| 473 | |
| 474 | kwargs = { |
| 475 | "videos": final_video_paths, |
| 476 | "combined_videos": combined_video_paths, |
| 477 | "script": video_script, |
| 478 | "terms": video_terms, |
| 479 | "audio_file": audio_file, |
| 480 | "audio_duration": audio_duration, |
| 481 | "subtitle_path": subtitle_path, |
| 482 | "materials": downloaded_videos, |
| 483 | "cross_post_results": cross_post_results if cross_post_results else None, |
| 484 | } |
| 485 | sm.state.update_task( |
| 486 | task_id, state=const.TASK_STATE_COMPLETE, progress=100, **kwargs |
| 487 | ) |
| 488 | return kwargs |
| 489 | |
| 490 | |
| 491 | if __name__ == "__main__": |
| 492 | task_id = "task_id" |
| 493 | params = VideoParams( |
| 494 | video_subject="金钱的作用", |
| 495 | voice_name="zh-CN-XiaoyiNeural-Female", |
| 496 | voice_rate=1.0, |
| 497 | ) |
| 498 | start(task_id, params, stop_at="video") |
| 499 |