from __future__ import annotations import argparse import copy import json import math import os import re import shutil import sys from typing import TYPE_CHECKING, Any, Sequence from uuid import UUID, uuid4 from loguru import logger if TYPE_CHECKING: from app.models.schema import MaterialInfo, VideoParams DEFAULT_VOICE_NAME = "zh-CN-XiaoxiaoNeural-Female" # 对应 webui/Main.py 的 VOICE_MODE_NONE 和 VOICE_MODE_UPLOAD。两端目前没有 # 共享这些常量，因此在这里保留字面值并注明来源。 UI_VOICE_MODE_NONE = "none" # 字幕位置的内置默认值。VideoParams 也有同名默认值，但它是 Pydantic 字段 # 默认，只在模块导入时读取一次 config.ui：此时 config.toml 里的非法值会被 # 直接冻结进模型，既不会校验也无法在测试中替换。CLI 因此自己保存一份， # 保证“非法保存值回退到默认值”对命令行始终成立。 DEFAULT_SUBTITLE_POSITION = "bottom" DEFAULT_CUSTOM_POSITION = 70.0 UI_VOICE_MODE_UPLOAD = "upload" # 这两种保存的配音方式都表示不要自动配音。 UI_VOICE_MODES_WITHOUT_TTS = frozenset({UI_VOICE_MODE_NONE, UI_VOICE_MODE_UPLOAD}) _PIPELINE_STAGES = ("script", "terms", "audio", "subtitle", "materials", "video") _CUSTOM_AUDIO_EXTENSIONS = {".mp3", ".wav", ".m4a", ".aac", ".flac", ".ogg"} _BATCH_FILE_MAX_BYTES = 1024 * 1024 _BATCH_TASK_MAX_COUNT = 100 # 单任务 argparse 和批量清单必须共享同一来源集合。此前两处手工维护导致 # openai_image 只在单任务入口可用；集中定义后，新增 Provider 不会再次遗漏 # 批量校验。这里仅包含 CLI 已公开支持的来源，不强行暴露 WebUI 专属流程。 _CLI_VIDEO_SOURCES = ( "pexels", "pixabay", "coverr", "wavespeed", "volcengine_seedance", "ofox", "metaso_minimax", "muapi", "openai_image", "local", ) class _CliHelpFormatter( argparse.ArgumentDefaultsHelpFormatter, argparse.RawDescriptionHelpFormatter, ): """在保留多行示例排版的同时，自动展示有意义的默认值。""" def _get_help_string(self, action): help_text = action.help or "" if ( "%(default)" not in help_text and action.default not in (None, "", argparse.SUPPRESS) and action.option_strings and "default:" not in help_text.lower() ): help_text += " (default: %(default)s)" return help_text def _positive_int(value: str) -> int: parsed = int(value) if parsed < 1: raise argparse.ArgumentTypeError(f"value must be >= 1, got {parsed}") return parsed def _paragraph_count(value: str) -> int: parsed = int(value) if parsed < 1 or parsed > 10: raise argparse.ArgumentTypeError( f"paragraph-number must be between 1 and 10, got {parsed}" ) return parsed def _non_negative_float(value: str) -> float: parsed = float(value) if not math.isfinite(parsed) or parsed < 0: raise argparse.ArgumentTypeError(f"value must be a finite number >= 0, got {value!r}") return parsed def _positive_float(value: str) -> float: parsed = float(value) if not math.isfinite(parsed) or parsed <= 0: raise argparse.ArgumentTypeError(f"value must be a finite number > 0, got {value!r}") return parsed # 片段播放速度的取值范围。app/utils/utils.py 的 normalize_clip_speed() 用同一 # 范围做限幅，webui/Main.py 的滑块也使用 0.5~2.0。集中定义后，单任务参数、 # 批量清单校验和 [ui] 保存值不会再次出现取值不一致。 _CLIP_SPEED_MIN = 0.5 _CLIP_SPEED_MAX = 2.0 def _clip_speed(value: str) -> float: parsed = float(value) if not math.isfinite(parsed) or not _CLIP_SPEED_MIN <= parsed <= _CLIP_SPEED_MAX: raise argparse.ArgumentTypeError( "video-clip-speed must be a finite number between " f"{_CLIP_SPEED_MIN} and {_CLIP_SPEED_MAX}, got {value!r}" ) return parsed def _percent_position(value: str) -> float: parsed = float(value) if not math.isfinite(parsed) or parsed < 0 or parsed > 100: raise argparse.ArgumentTypeError( f"custom-position must be a finite number between 0 and 100, got {value!r}" ) return parsed def _hex_color(value: str) -> str: if not re.fullmatch(r"#[0-9a-fA-F]{6}", value): raise argparse.ArgumentTypeError( f"color must use #RRGGBB format, got {value!r}" ) return value def _subtitle_position(value: str) -> str: """校验保存的字幕位置，取值范围与命令行参数保持一致。""" if value not in _SUBTITLE_POSITION_VALUES: raise argparse.ArgumentTypeError( "subtitle-position must be one of: " f"{', '.join(_SUBTITLE_POSITION_VALUES)}, got {value!r}" ) return value def _subtitle_display_mode(value: str) -> str: """校验保存的字幕展示模式，取值范围与命令行参数保持一致。""" if value not in _SUBTITLE_DISPLAY_MODE_VALUES: raise argparse.ArgumentTypeError( "subtitle-display-mode must be one of: " f"{', '.join(_SUBTITLE_DISPLAY_MODE_VALUES)}, got {value!r}" ) return value def _subtitle_animation(value: str) -> str: """校验保存的字幕动画，取值范围与命令行参数保持一致。""" if value not in _SUBTITLE_ANIMATION_VALUES: raise argparse.ArgumentTypeError( "subtitle-animation must be one of: " f"{', '.join(_SUBTITLE_ANIMATION_VALUES)}, got {value!r}" ) return value def _task_id(value: str) -> str: """CLI 自定义任务标识只接受 UUID，避免该值被解释为文件系统路径。""" try: return str(UUID(value.strip())) except (AttributeError, ValueError) as exc: raise argparse.ArgumentTypeError( f"task-id must be a valid UUID, got {value!r}" ) from exc _TRANSITION_MODE_VALUES = { "none": None, "shuffle": "Shuffle", "fade-in": "FadeIn", "fade-out": "FadeOut", "slide-in": "SlideIn", "slide-out": "SlideOut", "zoom-in": "ZoomIn", "zoom-out": "ZoomOut", } # 单任务 argparse 和批量清单必须共享同一份字幕位置取值，避免再次出现 # 「WebUI 能保存、CLI 却拒绝」的落差。取值需与 app/services/video.py 的 # 渲染分支及 webui/Main.py 的下拉框保持一致。 _SUBTITLE_POSITION_VALUES = ( "top", "center", "bottom", "two_thirds_bottom", "custom", ) # 字幕展示模式与入场动画由 WebUI 保存进 [ui]。两者随「逐词字幕 + 弹跳动画」 # 加入时只改了 WebUI 与模型字段默认值，命令行一直缺少开关。取值需与 # app/models/schema.py 的 _SUBTITLE_DISPLAY_MODES / _SUBTITLE_ANIMATIONS 一致。 _SUBTITLE_DISPLAY_MODE_VALUES = ("sentence", "word_by_word") _SUBTITLE_ANIMATION_VALUES = ("none", "pop_spring") def _transition_mode(value: str) -> str | None: normalized = value.strip().lower() if normalized not in _TRANSITION_MODE_VALUES: allowed = ", ".join(_TRANSITION_MODE_VALUES) raise argparse.ArgumentTypeError( f"video-transition-mode must be one of: {allowed}" ) return _TRANSITION_MODE_VALUES[normalized] def _video_fit_mode(value: str) -> str: normalized = value.strip().lower() if normalized not in {"cover", "contain"}: raise argparse.ArgumentTypeError( "video-fit-mode must be one of: cover, contain" ) return normalized # 只有这些取值会请求外部音乐生成服务，因此只有它们消费配乐提示词。取值需与 # app/services/task.py 的 _VIDEO_MUSIC_PROVIDERS 保持一致；集中定义后，新增 # 配乐供应商不会再次遗漏 --bgm-type 校验、批量校验和提示词参数。 _CLI_MUSIC_BGM_TYPES = ("sonilo", "elevenlabs") _BGM_TYPE_CHOICES_TEXT = "none, random, custom, " + ", ".join(_CLI_MUSIC_BGM_TYPES) _BGM_TYPE_METAVAR = "{none,random,custom," + ",".join(_CLI_MUSIC_BGM_TYPES) + "}" def _bgm_type(value: str) -> str: normalized = value.strip().lower() if normalized == "none": return "" if normalized in {"", "random", "custom", *_CLI_MUSIC_BGM_TYPES}: return normalized raise argparse.ArgumentTypeError( f"bgm-type must be one of: {_BGM_TYPE_CHOICES_TEXT}" ) def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Generate MoneyPrinterTurbo videos without the WebUI.\n\n" "Provider settings and credentials are read from config.toml.\n" "Default full-video generation requires a configured LLM and Pexels API key.\n" "The default Edge TTS voice requires no API key." ), epilog=""" Examples: Generate a complete video with the default Edge TTS voice: uv run python cli.py --video-subject "How AI is changing everyday life" Generate from local files. Relative paths use the current working directory; absolute paths are also accepted: uv run python cli.py --video-subject "How AI is changing everyday life" \\ --video-source local --video-materials "./1.mp4,./2.mp4" Generate with a prepared script and no voiceover: uv run python cli.py --video-script "Your complete script" \\ --voice-name no-voice --stop-at video Stop after script generation: uv run python cli.py --video-subject "How AI is changing everyday life" --stop-at script Run a JSON array or JSONL manifest. CLI options provide defaults and each object overrides VideoParams fields for one task: uv run python cli.py --batch-file ./tasks.jsonl --stop-at script Pipeline stages: script Generate or return the script. terms Generate material search terms; unavailable with local materials. audio Generate TTS, silent audio, or use --custom-audio-file. subtitle Generate subtitles when enabled. materials Download online materials or preprocess local files. video Generate the final video and run configured cross-posting. The command stops immediately after the selected stage and prints that stage's result. Output and exit status: Task files are written to storage/tasks//. A successful command prints one JSON object to stdout and exits with 0. Batch mode prints one JSON summary after all tasks finish. Task failures exit with 1; argument or manifest errors exit with 2 before any batch task starts. Runtime logs are written to stderr. Batch manifests: --batch-file accepts a UTF-8 JSON array or JSONL file with one object per non-empty line (up to 100 tasks and 1 MiB). Objects may override VideoParams fields; unknown fields are rejected. Every merged task needs video_subject or video_script. The manifest path is relative to the current working directory. Relative custom_audio_file and local video_materials[].url values declared in a manifest are relative to the manifest directory. Relative paths supplied by CLI options remain relative to the current working directory. bgm_file and font_name keep their managed storage/resource lookup rules. --stop-at is a batch-wide CLI option. The summary has total, succeeded, failed, and tasks keys; every task entry has index, task_id, status, result, failed_stage, and error. """, formatter_class=_CliHelpFormatter, ) content_group = parser.add_argument_group("script and content") content_group.add_argument( "--video-subject", default="", help="video topic; required unless --video-script is provided", ) content_group.add_argument( "--video-script", default="", help="complete script; skips LLM script generation when provided", ) content_group.add_argument( "--video-terms", default=None, help="comma-separated material search terms; generated automatically when omitted", ) content_group.add_argument( "--video-language", default=None, help=( "script language code, such as zh-CN or en-US (default: auto-detect)" ), ) content_group.add_argument( "--paragraph-number", type=_paragraph_count, default=None, help="number of generated script paragraphs, from 1 to 10 (default: 1)", ) content_group.add_argument( "--video-script-prompt", default=None, help="additional requirements for LLM script generation", ) content_group.add_argument( "--custom-system-prompt", default=None, help="replace the default LLM system prompt for script generation", ) material_group = parser.add_argument_group("materials and pipeline") material_group.add_argument( "--video-source", default="pexels", choices=_CLI_VIDEO_SOURCES, help="video material provider; online providers require matching API keys in config.toml", ) material_group.add_argument( "--video-materials", default="", metavar="PATH[,PATH...]", help=( "comma-separated local image/video paths for --video-source local; relative " "paths use the current working directory, then storage/local_videos as a " "compatibility fallback; absolute paths are accepted" ), ) material_group.add_argument( "--stop-at", default="video", choices=_PIPELINE_STAGES, help="stop after this pipeline stage; see the stage order below", ) material_group.add_argument( "--confirm-wavespeed-charge", action="store_true", help=( "confirm that WaveSpeed video generation creates paid tasks; required " "with --video-source wavespeed for materials or video output" ), ) material_group.add_argument( "--confirm-seedance-charge", action="store_true", help=( "confirm that Volcano Engine Seedance creates paid Ark tasks; required " "with --video-source volcengine_seedance for materials or video output" ), ) material_group.add_argument( "--confirm-ofox-charge", action="store_true", help=( "confirm that OFox video generation creates paid tasks; required " "with --video-source ofox for materials or video output" ), ) material_group.add_argument( "--confirm-metaso-minimax-charge", action="store_true", help=( "confirm that Metaso MiniMax H3 creates paid video tasks; required " "with --video-source metaso_minimax for materials or video output" ), ) material_group.add_argument( "--confirm-muapi-charge", action="store_true", help=( "confirm that MuAPI video generation creates paid tasks; required " "with --video-source muapi for materials or video output" ), ) video_group = parser.add_argument_group("video output") video_group.add_argument( "--video-count", type=_positive_int, default=1, help="number of output videos, at least 1", ) video_group.add_argument( "--video-aspect", choices=["9:16", "16:9", "1:1"], default="9:16", help="output aspect ratio: portrait, landscape, or square", ) video_group.add_argument( "--video-fit-mode", type=_video_fit_mode, choices=["cover", "contain"], default=None, help=( "fit mismatched source clips by filling and center-cropping (cover) " "or preserving the full frame with black bars (contain); default: cover" ), ) video_group.add_argument( "--video-concat-mode", choices=["random", "sequential"], default=None, help="source clip concatenation order (default: random)", ) video_group.add_argument( "--video-transition-mode", type=_transition_mode, default=None, metavar="{none,shuffle,fade-in,fade-out,slide-in,slide-out,zoom-in,zoom-out}", help="transition applied between source clips (default: none)", ) video_group.add_argument( "--video-clip-duration", type=_positive_int, default=None, help=( "maximum duration of each source clip in seconds, at least 1 (default: 5)" ), ) video_group.add_argument( "--video-clip-speed", type=_clip_speed, default=None, help=( "playback speed multiplier applied to every source clip, between " "0.5 and 2.0 (default: 1.0)" ), ) video_group.add_argument( "--match-materials-to-script", default=None, action=argparse.BooleanOptionalAction, help=( "preserve script keyword order while selecting and concatenating materials " "(default: disabled)" ), ) video_group.add_argument( "--n-threads", type=_positive_int, default=None, help="FFmpeg worker thread count, at least 1 (default: 2)", ) audio_group = parser.add_argument_group("voiceover and background music") audio_group.add_argument( "--voice-name", default=None, help=( f"TTS voice identifier. Defaults to [ui].voice_name from " f"config.toml, otherwise {DEFAULT_VOICE_NAME}. A saved " "[ui].voice_mode of 'none' or 'upload' resolves to no-voice " "instead, unless this option is given. " "Use 'no-voice' for silent output. Provider-specific identifiers " "use prefixes such as gemini:, mimo:, elevenlabs:, chatterbox:, " "kokoro:, and voxcpm:" ), ) audio_group.add_argument( "--voice-volume", type=_non_negative_float, default=None, help=( "final voiceover volume multiplier, a finite number >= 0 (default: " "[ui].voice_volume from config.toml; 1.0 when unset)" ), ) audio_group.add_argument( "--voice-rate", type=_positive_float, default=None, help=( "speech rate multiplier, a finite number > 0 (default: " "[ui].voice_rate from config.toml; 1.0 when unset)" ), ) audio_group.add_argument( "--custom-audio-file", default=None, metavar="PATH", help=( "existing MP3/WAV/M4A/AAC/FLAC/OGG voiceover; relative paths use the " "current working directory. This skips TTS; set subtitle_provider=whisper " "to transcribe it" ), ) audio_group.add_argument( "--bgm-type", type=_bgm_type, default=None, metavar=_BGM_TYPE_METAVAR, help=( "background music mode; Sonilo reads its API key from config.toml or " "SONILO_API_KEY, ElevenLabs from config.toml or ELEVENLABS_API_KEY; " "--bgm-file implies custom when omitted (default: random)" ), ) audio_group.add_argument( "--sonilo-bgm-prompt", default=None, help="optional music style prompt for Sonilo, up to 2000 characters", ) audio_group.add_argument( "--video-music-prompt", default=None, help=( "optional music style prompt for the selected AI background music " "provider (sonilo or elevenlabs); the provider's own length limit " "is enforced before generation" ), ) audio_group.add_argument( "--bgm-file", default=None, metavar="PATH", help=( "custom supported audio file inside storage/bgm or resource/songs; " "accepts a filename or an allowed managed path" ), ) audio_group.add_argument( "--bgm-volume", type=_non_negative_float, default=None, help=( "background music volume multiplier, a finite number >= 0 (default: 0.2)" ), ) subtitle_group = parser.add_argument_group("subtitles") subtitle_group.add_argument( "--subtitle-enabled", default=None, action=argparse.BooleanOptionalAction, help=( "enable subtitles; use --no-subtitle-enabled to disable " "(default: [ui].subtitle_enabled from config.toml; enabled when " "unset)" ), ) subtitle_group.add_argument( "--font-name", default=None, help=( "subtitle font filename inside resource/fonts " "(default: [ui].font_name from config.toml; " "STHeitiMedium.ttc when unset)" ), ) subtitle_group.add_argument( "--subtitle-position", choices=_SUBTITLE_POSITION_VALUES, default=None, help=( "subtitle vertical position (default: [ui].subtitle_position from " "config.toml; bottom when unset)" ), ) subtitle_group.add_argument( "--subtitle-display-mode", choices=_SUBTITLE_DISPLAY_MODE_VALUES, default=None, help=( "subtitle timing: sentence by sentence, or one word at a time " "(default: [ui].subtitle_display_mode from config.toml; sentence)" ), ) subtitle_group.add_argument( "--subtitle-animation", choices=_SUBTITLE_ANIMATION_VALUES, default=None, help=( "subtitle entrance animation (default: [ui].subtitle_animation from " "config.toml; none when unset)" ), ) subtitle_group.add_argument( "--custom-position", type=_percent_position, default=None, help=( "custom position as percent from top, 0-100; requires " "--subtitle-position custom (default: [ui].custom_position from " "config.toml; 70 when unset)" ), ) subtitle_group.add_argument( "--text-fore-color", type=_hex_color, default=None, help=( "subtitle text color in #RRGGBB format; quote the value in shells " "that treat # as a comment (default: [ui].text_fore_color from " "config.toml; #FFFFFF when unset)" ), ) subtitle_group.add_argument( "--font-size", type=_positive_int, default=None, help=( "subtitle font size (default: [ui].font_size from config.toml; " "60 when unset)" ), ) subtitle_group.add_argument( "--stroke-color", type=_hex_color, default=None, help=( "subtitle outline color in #RRGGBB format (default: " "[ui].stroke_color from config.toml; #000000 when unset)" ), ) subtitle_group.add_argument( "--stroke-width", type=_non_negative_float, default=None, help=( "subtitle outline width, a finite number >= 0 (default: " "[ui].stroke_width from config.toml; 1.5 when unset)" ), ) subtitle_group.add_argument( "--subtitle-background-enabled", default=None, action=argparse.BooleanOptionalAction, help=( "enable subtitle background; use --no-subtitle-background-enabled to " "disable (default: [ui].subtitle_background_enabled from " "config.toml; disabled when unset)" ), ) subtitle_group.add_argument( "--subtitle-background-color", type=_hex_color, default=None, help=( "subtitle background color in #RRGGBB format (default: " "[ui].subtitle_background_color from config.toml)" ), ) subtitle_group.add_argument( "--rounded-subtitle-background", default=None, action=argparse.BooleanOptionalAction, help=( "use a rounded subtitle background (default: " "[ui].rounded_subtitle_background from config.toml; " "disabled when unset)" ), ) execution_group = parser.add_argument_group("execution") execution_mode = execution_group.add_mutually_exclusive_group() execution_mode.add_argument( "--task-id", type=_task_id, default=None, help="custom UUID used for storage/tasks/; generated automatically when omitted", ) execution_mode.add_argument( "--batch-file", default=None, metavar="PATH", help=( "UTF-8 JSON array or JSONL task manifest; conflicts with --task-id and " "generates a UUID for every task" ), ) args = parser.parse_args(argv) if ( not args.batch_file and not args.video_subject.strip() and not args.video_script.strip() ): parser.error("one of --video-subject or --video-script is required") if not args.batch_file and args.video_source == "local" and args.stop_at == "terms": parser.error( "--stop-at terms has no effect with --video-source local " "(search terms are not generated for local sources)" ) stage_requires_materials = args.stop_at in {"materials", "video"} has_video_materials = bool((args.video_materials or "").strip()) if ( not args.batch_file and args.video_source == "local" and stage_requires_materials and not has_video_materials ): parser.error( "--video-materials is required with --video-source local when " "--stop-at is materials or video" ) if not args.batch_file and args.video_source != "local" and has_video_materials: parser.error("--video-materials can only be used with --video-source local") if ( not args.batch_file and args.video_source == "wavespeed" and stage_requires_materials and not args.confirm_wavespeed_charge ): parser.error( "--confirm-wavespeed-charge is required with " "--video-source wavespeed" ) if ( not args.batch_file and args.video_source == "volcengine_seedance" and stage_requires_materials and not args.confirm_seedance_charge ): parser.error( "--confirm-seedance-charge is required with " "--video-source volcengine_seedance" ) if ( not args.batch_file and args.video_source == "ofox" and stage_requires_materials and not args.confirm_ofox_charge ): parser.error( "--confirm-ofox-charge is required with --video-source ofox" ) if ( not args.batch_file and args.video_source == "metaso_minimax" and stage_requires_materials and not args.confirm_metaso_minimax_charge ): parser.error( "--confirm-metaso-minimax-charge is required with " "--video-source metaso_minimax" ) if ( not args.batch_file and args.video_source == "muapi" and stage_requires_materials and not args.confirm_muapi_charge ): parser.error( "--confirm-muapi-charge is required with --video-source muapi" ) if args.bgm_file: if args.bgm_type in (None, "custom"): args.bgm_type = "custom" elif not args.batch_file: parser.error("--bgm-file can only be combined with --bgm-type custom") if args.sonilo_bgm_prompt: if args.bgm_type in (None, "sonilo"): args.bgm_type = "sonilo" elif not args.batch_file: parser.error( "--sonilo-bgm-prompt can only be combined with --bgm-type sonilo" ) # 提示词字段本身与供应商无关，所以不像 Sonilo 专用参数那样推断 bgm_type： # 推断出来的供应商可能不是用户想要的那个。必须显式选择 AI 配乐供应商， # 否则该提示词会被静默丢弃。批量清单仍可按任务单独设置该字段。 if ( not args.batch_file and args.video_music_prompt and args.bgm_type not in _CLI_MUSIC_BGM_TYPES ): parser.error( "--video-music-prompt requires --bgm-type sonilo or elevenlabs" ) if ( not args.batch_file and args.custom_position is not None and args.subtitle_position != "custom" ): parser.error("--custom-position requires --subtitle-position custom") # 只有显式的 --no-subtitle-enabled 才算冲突。默认值现在是 None， # 保存的关闭状态在 build_video_params 中处理，不应在这里报参数错误。 if ( not args.batch_file and args.stop_at == "subtitle" and args.subtitle_enabled is False ): parser.error("--stop-at subtitle cannot be combined with --no-subtitle-enabled") if not args.batch_file and args.subtitle_background_enabled is False and ( args.subtitle_background_color is not None or args.rounded_subtitle_background is True ): parser.error( "subtitle background color or rounding cannot be enabled together with " "--no-subtitle-background-enabled" ) return args def _ui_config_value(ui_config, key: str, expected_type, checker=None): """ 读取 ``[ui]`` 中保存的 WebUI 设置，取不到可用值时返回 ``None``。 该配置段也可能被手工编辑，因此这里直接丢弃不可用的条目，让调用方回退到 内置默认值，而不是把脏数据继续传下去，触发 traceback 或 ``VideoParams`` 的校验错误。 ``checker`` 复用命令行使用的同一批类型函数（如 ``_hex_color``），因此保存 值和命令行参数遵循完全相同的取值规则，例如音量不能为负、颜色必须是 ``#RRGGBB``。 """ value = ui_config.get(key) if value is None: return None # ``isinstance(True, int)`` 为真，所以在期望数字时必须显式排除 bool。 if isinstance(value, bool) != (expected_type is bool): return None # TOML 中的 ``1`` 是整数，但对音量、语速这类字段同样是合法取值。 if expected_type is float and isinstance(value, int): value = float(value) if not isinstance(value, expected_type): return None if expected_type is str and not value.strip(): return None if checker is not None: try: return checker(str(value)) except argparse.ArgumentTypeError: return None return value def _resolve_subtitle_enabled(args: argparse.Namespace, ui_config) -> bool: """ 按优先级解析字幕开关：命令行 > WebUI 保存值 > 默认开启。 ``--stop-at subtitle`` 明确要求生成字幕，因此保存的关闭状态不能让该阶段 变成空操作；显式的 --no-subtitle-enabled 与该阶段的组合在参数校验中已被 拒绝，所以这里只需处理保存值。 """ if args.subtitle_enabled is not None: return args.subtitle_enabled if args.stop_at == "subtitle": return True saved = _ui_config_value(ui_config, "subtitle_enabled", bool) return True if saved is None else saved def _resolve_voice_name(args: argparse.Namespace, ui_config) -> str: """ 按优先级解析音色：命令行 > WebUI 保存的配音方式和音色 > 内置默认值。 WebUI 把“无配音”保存为独立的 voice_mode，同时保留用户上一次真正选择的 音色，以便切回自动配音后恢复。所以这里不能只读 voice_name，否则保存的 无配音状态会被忽略，并可能重新触发付费供应商请求。 """ from app.services.voice import NO_VOICE_NAME if args.voice_name: return args.voice_name # 无配音和上传自备音频都表示用户不想要自动配音。上传模式的文件路径不会 # 写入 [ui]，CLI 无法复现该上传；此时沿用保存的音色会静默触发付费 TTS # 请求，因此两种模式都映射为 no-voice，需要配音时显式传 --voice-name。 if _ui_config_value(ui_config, "voice_mode", str) in UI_VOICE_MODES_WITHOUT_TTS: return NO_VOICE_NAME return _ui_config_value(ui_config, "voice_name", str) or DEFAULT_VOICE_NAME def build_video_params(args: argparse.Namespace) -> VideoParams: # 参数帮助和校验不需要加载应用配置。仅在真正构建任务参数时导入模型， # 避免执行 ``cli.py -h`` 时产生配置初始化日志。 from app.config import config from app.models.schema import MaterialInfo, VideoParams ui_config = config.ui video_terms = args.video_terms if video_terms: video_terms = [ term.strip() for term in re.split(r"[,，]", video_terms) if term.strip() ] video_materials = None materials_arg = args.video_materials or "" if materials_arg.strip(): video_materials = [ # Actual duration will be detected during video processing; use 0 as placeholder. MaterialInfo(provider="local", url=item.strip(), duration=0) for item in materials_arg.split(",") if item.strip() ] params_kwargs = { "video_subject": args.video_subject.strip(), "video_script": args.video_script, "video_terms": video_terms, "video_source": args.video_source, "video_materials": video_materials, "video_count": args.video_count, "video_aspect": args.video_aspect, "voice_name": _resolve_voice_name(args, ui_config), "subtitle_enabled": _resolve_subtitle_enabled(args, ui_config), } optional_arg_names = [ "video_language", "paragraph_number", "video_script_prompt", "custom_system_prompt", "video_fit_mode", "video_concat_mode", "video_transition_mode", "video_clip_duration", "video_clip_speed", "match_materials_to_script", "n_threads", "voice_volume", "voice_rate", "custom_audio_file", "bgm_type", "bgm_file", "bgm_volume", "sonilo_bgm_prompt", "video_music_prompt", "font_name", "subtitle_position", "subtitle_display_mode", "subtitle_animation", "custom_position", "text_fore_color", "font_size", "stroke_color", "stroke_width", "rounded_subtitle_background", ] for name in optional_arg_names: value = getattr(args, name) if value is not None: params_kwargs[name] = value # 没有显式传入命令行参数时，使用 WebUI 保存的值。只补充上面尚未由命令行 # 设置的字段；若保存值缺失，则继续沿用 VideoParams 的默认值。 # VideoParams 里多数可保存字段（subtitle_position、custom_position 等）在 # 模型导入时就读取 config.ui，video_clip_speed 却是硬编码的 1.0，因此必须 # 在这里显式补上，否则 WebUI 保存的速度在命令行会被静默丢弃。 ui_defaults = ( ("video_clip_speed", float, _clip_speed), ("video_fit_mode", str, _video_fit_mode), ("font_name", str, None), ("subtitle_display_mode", str, _subtitle_display_mode), ("subtitle_animation", str, _subtitle_animation), ("text_fore_color", str, _hex_color), ("font_size", int, _positive_int), ("rounded_subtitle_background", bool, None), ("voice_volume", float, _non_negative_float), ("voice_rate", float, _positive_float), ("stroke_color", str, _hex_color), ("stroke_width", float, _non_negative_float), ) for name, expected_type, checker in ui_defaults: if name in params_kwargs: continue value = _ui_config_value(ui_config, name, expected_type, checker) if value is not None: params_kwargs[name] = value # 字幕位置必须由 CLI 给出确定值，不能交给上面说明的导入期字段默认值。 if "subtitle_position" not in params_kwargs: params_kwargs["subtitle_position"] = ( _ui_config_value(ui_config, "subtitle_position", str, _subtitle_position) or DEFAULT_SUBTITLE_POSITION ) if "custom_position" not in params_kwargs: saved_custom_position = _ui_config_value( ui_config, "custom_position", float, _percent_position ) params_kwargs["custom_position"] = ( DEFAULT_CUSTOM_POSITION if saved_custom_position is None else saved_custom_position ) if args.subtitle_background_enabled is False: params_kwargs["text_background_color"] = False params_kwargs["rounded_subtitle_background"] = False elif args.subtitle_background_color is not None: params_kwargs["text_background_color"] = args.subtitle_background_color elif args.subtitle_background_enabled is True: # 用户只开启了背景而没有覆盖颜色，因此优先沿用 WebUI 保存的颜色， # 只有在没有可用保存值时才回退到默认背景。 params_kwargs["text_background_color"] = ( _ui_config_value( ui_config, "subtitle_background_color", str, _hex_color ) or True ) else: # “关闭背景”加上颜色作为命令行组合是参数错误；但作为保存的设置， # 同样的组合不应中断运行，只表示禁用背景。 ui_enabled = _ui_config_value( ui_config, "subtitle_background_enabled", bool ) ui_color = _ui_config_value( ui_config, "subtitle_background_color", str, _hex_color ) if ui_enabled is False: params_kwargs["text_background_color"] = False if args.rounded_subtitle_background is None: params_kwargs["rounded_subtitle_background"] = False elif ui_color is not None: params_kwargs["text_background_color"] = ui_color elif ui_enabled is True: params_kwargs["text_background_color"] = True return VideoParams(**params_kwargs) def _load_batch_manifest(raw_path: str) -> tuple[str, list[dict[str, Any]]]: expanded_path = os.path.expanduser(raw_path.strip()) if not expanded_path: raise ValueError("--batch-file path cannot be empty") candidate = ( expanded_path if os.path.isabs(expanded_path) else os.path.join(os.getcwd(), expanded_path) ) manifest_path = os.path.realpath(candidate) if not os.path.isfile(manifest_path): raise ValueError(f"batch manifest does not exist or is not a file: {raw_path}") with open(manifest_path, "rb") as manifest_file: payload = manifest_file.read(_BATCH_FILE_MAX_BYTES + 1) if len(payload) > _BATCH_FILE_MAX_BYTES: raise ValueError("batch manifest exceeds the 1 MiB limit") try: text = payload.decode("utf-8-sig") except UnicodeDecodeError as exc: raise ValueError("batch manifest must be UTF-8 encoded") from exc if not text.strip(): raise ValueError("batch manifest must contain at least one task") if text.lstrip().startswith("["): try: parsed = json.loads(text) except json.JSONDecodeError as exc: raise ValueError( f"invalid JSON array in batch manifest: {exc.msg}" ) from exc if not isinstance(parsed, list): raise ValueError("batch manifest JSON must be an array") entries = parsed else: entries = [] for line_number, line in enumerate(text.splitlines(), start=1): if not line.strip(): continue try: entries.append(json.loads(line)) except json.JSONDecodeError as exc: raise ValueError( f"invalid JSONL in batch manifest at line {line_number}: {exc.msg}" ) from exc if not entries: raise ValueError("batch manifest must contain at least one task") if len(entries) > _BATCH_TASK_MAX_COUNT: raise ValueError( f"batch manifest contains {len(entries)} tasks; " f"the limit is {_BATCH_TASK_MAX_COUNT}" ) for index, entry in enumerate(entries, start=1): if not isinstance(entry, dict): raise ValueError(f"batch task {index} must be a JSON object") return manifest_path, entries def _validate_batch_entry_fields( entry: dict[str, Any], *, index: int, allowed_fields: set[str], ) -> None: unknown_fields = sorted(set(entry) - allowed_fields) if unknown_fields: raise ValueError( f"batch task {index} contains unknown VideoParams fields: " f"{', '.join(unknown_fields)}" ) materials = entry.get("video_materials") if not isinstance(materials, list): return allowed_material_fields = {"provider", "url", "duration", "source_info"} for material_index, material in enumerate(materials, start=1): if not isinstance(material, dict): continue unknown_material_fields = sorted(set(material) - allowed_material_fields) if unknown_material_fields: raise ValueError( f"batch task {index} material {material_index} contains unknown " f"MaterialInfo fields: {', '.join(unknown_material_fields)}" ) def _manifest_relative_path(raw_path: str, manifest_directory: str) -> str: expanded_path = os.path.expanduser(raw_path.strip()) if not expanded_path or os.path.isabs(expanded_path): return expanded_path return os.path.realpath(os.path.join(manifest_directory, expanded_path)) def _resolve_batch_entry_paths( params: VideoParams, *, override_fields: set[str], manifest_directory: str, ) -> None: if "custom_audio_file" in override_fields and params.custom_audio_file: params.custom_audio_file = _manifest_relative_path( params.custom_audio_file, manifest_directory, ) if ( "video_materials" in override_fields and params.video_source == "local" and params.video_materials ): for material in params.video_materials: material.url = _manifest_relative_path( material.url, manifest_directory, ) def _validate_batch_task_params( params: VideoParams, *, stop_at: str, custom_position_is_explicit: bool, wavespeed_charge_confirmed: bool, seedance_charge_confirmed: bool, ofox_charge_confirmed: bool, metaso_minimax_charge_confirmed: bool, muapi_charge_confirmed: bool, ) -> None: if not params.video_subject.strip() and not params.video_script.strip(): raise ValueError("one of video_subject or video_script is required") if params.video_source not in _CLI_VIDEO_SOURCES: raise ValueError( "video_source must be one of: " + ", ".join(_CLI_VIDEO_SOURCES) ) for field_name, value in ( ("video_aspect", params.video_aspect), ("video_concat_mode", params.video_concat_mode), ): # These schema fields remain Optional for compatibility with historical # API payloads, but the video pipeline always dereferences their enum # values. A manifest's explicit null must fail before any batch task starts. if value is None: raise ValueError(f"{field_name} cannot be null") if params.video_source == "local" and stop_at == "terms": raise ValueError( "stop_at=terms has no effect with video_source=local" ) if ( params.video_source == "local" and stop_at in {"materials", "video"} and not params.video_materials ): raise ValueError( "video_materials is required with video_source=local when " "stop_at is materials or video" ) if params.video_source != "local" and params.video_materials: raise ValueError("video_materials can only be used with video_source=local") if ( params.video_source == "wavespeed" and stop_at in {"materials", "video"} and not wavespeed_charge_confirmed ): raise ValueError( "--confirm-wavespeed-charge is required for WaveSpeed video generation" ) if ( params.video_source == "volcengine_seedance" and stop_at in {"materials", "video"} and not seedance_charge_confirmed ): raise ValueError( "--confirm-seedance-charge is required for Volcano Engine Seedance" ) if ( params.video_source == "ofox" and stop_at in {"materials", "video"} and not ofox_charge_confirmed ): raise ValueError("--confirm-ofox-charge is required for OFox video generation") if ( params.video_source == "metaso_minimax" and stop_at in {"materials", "video"} and not metaso_minimax_charge_confirmed ): raise ValueError( "--confirm-metaso-minimax-charge is required for Metaso MiniMax H3" ) if ( params.video_source == "muapi" and stop_at in {"materials", "video"} and not muapi_charge_confirmed ): raise ValueError("--confirm-muapi-charge is required for MuAPI video generation") if stop_at == "subtitle" and not params.subtitle_enabled: raise ValueError("stop_at=subtitle cannot be combined with disabled subtitles") if params.subtitle_position not in _SUBTITLE_POSITION_VALUES: raise ValueError( "subtitle_position must be one of: " + ", ".join(_SUBTITLE_POSITION_VALUES) ) if custom_position_is_explicit and params.subtitle_position != "custom": raise ValueError("custom_position requires subtitle_position=custom") if not math.isfinite(params.custom_position) or not 0 <= params.custom_position <= 100: raise ValueError("custom_position must be a finite number between 0 and 100") if params.video_clip_speed is not None and ( not math.isfinite(params.video_clip_speed) or not _CLIP_SPEED_MIN <= params.video_clip_speed <= _CLIP_SPEED_MAX ): raise ValueError( "video_clip_speed must be a finite number between " f"{_CLIP_SPEED_MIN} and {_CLIP_SPEED_MAX}" ) if params.text_background_color is False and params.rounded_subtitle_background: raise ValueError( "rounded_subtitle_background requires an enabled subtitle background" ) color_values = { "text_fore_color": params.text_fore_color, "stroke_color": params.stroke_color, "text_background_color": ( params.text_background_color if isinstance(params.text_background_color, str) else None ), } for name, value in color_values.items(): if value is not None and not re.fullmatch(r"#[0-9a-fA-F]{6}", value): raise ValueError(f"{name} must use #RRGGBB format") numeric_constraints = ( ("voice_volume", params.voice_volume, 0, False), ("voice_rate", params.voice_rate, 0, True), ("bgm_volume", params.bgm_volume, 0, False), ("stroke_width", params.stroke_width, 0, False), ) for name, value, minimum, exclusive in numeric_constraints: if value is None: continue if not math.isfinite(value) or ( value <= minimum if exclusive else value < minimum ): comparison = ">" if exclusive else ">=" raise ValueError(f"{name} must be a finite number {comparison} {minimum}") for name, value in ( ("n_threads", params.n_threads), ("font_size", params.font_size), ): if value is not None and value < 1: raise ValueError(f"{name} must be >= 1") if params.bgm_type not in {"", "random", "custom", *_CLI_MUSIC_BGM_TYPES}: raise ValueError(f"bgm_type must be one of: {_BGM_TYPE_CHOICES_TEXT}") if params.bgm_file and params.bgm_type != "custom": raise ValueError("bgm_file requires bgm_type=custom") if params.sonilo_bgm_prompt and params.bgm_type != "sonilo": raise ValueError("sonilo_bgm_prompt requires bgm_type=sonilo") if params.video_music_prompt and params.bgm_type not in _CLI_MUSIC_BGM_TYPES: raise ValueError( "video_music_prompt requires bgm_type=sonilo or elevenlabs" ) def _build_batch_tasks(args: argparse.Namespace) -> list[VideoParams]: from app.models.schema import VideoParams manifest_path, entries = _load_batch_manifest(args.batch_file) manifest_directory = os.path.dirname(manifest_path) base_params = build_video_params(args) allowed_fields = set(VideoParams.model_fields) base_values = { field_name: getattr(base_params, field_name) for field_name in allowed_fields } tasks: list[VideoParams] = [] for index, entry in enumerate(entries, start=1): _validate_batch_entry_fields( entry, index=index, allowed_fields=allowed_fields, ) try: params = VideoParams.model_validate( {**copy.deepcopy(base_values), **entry} ) override_fields = set(entry) _resolve_batch_entry_paths( params, override_fields=override_fields, manifest_directory=manifest_directory, ) _validate_batch_task_params( params, stop_at=args.stop_at, custom_position_is_explicit=( args.custom_position is not None or "custom_position" in override_fields ), seedance_charge_confirmed=args.confirm_seedance_charge, wavespeed_charge_confirmed=args.confirm_wavespeed_charge, ofox_charge_confirmed=args.confirm_ofox_charge, metaso_minimax_charge_confirmed=( args.confirm_metaso_minimax_charge ), muapi_charge_confirmed=args.confirm_muapi_charge, ) except (TypeError, ValueError) as exc: raise ValueError(f"invalid batch task {index}: {exc}") from exc tasks.append(params) validation_plans = [] for index, params in enumerate(tasks, start=1): try: validation_plans.append(_validate_cli_files(params, stop_at=args.stop_at)) except (OSError, ValueError) as exc: raise ValueError(f"invalid batch task {index}: {exc}") from exc prepared_paths: dict[str, str] = {} created_paths: list[str] = [] try: for index, (local_videos_dir, resolved_materials) in enumerate( validation_plans, start=1 ): try: _prepare_cli_materials( local_videos_dir, resolved_materials, prepared_paths=prepared_paths, created_paths=created_paths, ) except OSError as exc: raise ValueError(f"invalid batch task {index}: {exc}") from exc except Exception: _remove_cli_material_copies(created_paths) raise return tasks def _resolve_cli_file( raw_path: str, *, description: str, fallback_dir: str | None = None, ) -> str: """ 将 CLI 文件参数按当前工作目录解析为绝对路径， 并在任务开始前确认存在。 本地素材旧版本始终相对 ``storage/local_videos`` 解析。为兼容已有脚本， 当前目录找不到相对路径时允许回退该目录；绝对路径始终按用户输入 直接解析。 """ expanded_path = os.path.expanduser(raw_path.strip()) if not expanded_path: raise ValueError(f"{description} path cannot be empty") candidate = ( expanded_path if os.path.isabs(expanded_path) else os.path.join(os.getcwd(), expanded_path) ) resolved_path = os.path.realpath(candidate) if not os.path.isfile(resolved_path) and fallback_dir and not os.path.isabs(expanded_path): resolved_path = os.path.realpath(os.path.join(fallback_dir, expanded_path)) if not os.path.isfile(resolved_path): raise ValueError(f"{description} file does not exist: {raw_path}") return resolved_path def _path_is_within_directory(file_path: str, directory: str) -> bool: try: return os.path.commonpath( [os.path.realpath(directory), os.path.realpath(file_path)] ) == os.path.realpath(directory) except ValueError: # Windows 不同盘符无法计算 commonpath，此时文件显然不在目标目录内。 return False def _resolve_managed_resource_file( raw_path: str, *, resource_dir: str, description: str, ) -> str: """解析项目资源文件，并确保绝对路径仍位于对应资源目录内。""" from app.utils import utils expanded_path = os.path.expanduser(raw_path.strip()) candidates = ( [expanded_path] if os.path.isabs(expanded_path) else [ os.path.join(resource_dir, expanded_path), os.path.join(utils.root_dir(), expanded_path), ] ) for candidate in candidates: resolved_path = os.path.realpath(candidate) if os.path.isfile(resolved_path) and _path_is_within_directory( resolved_path, resource_dir ): return resolved_path raise ValueError( f"{description} file must exist inside {resource_dir}: {raw_path}" ) def _validate_cli_files( params: VideoParams, stop_at: str ) -> tuple[str, list[tuple[MaterialInfo, str, str]]]: """ 无副作用地解析并校验 CLI 文件，避免批量预检留下素材副本。 自定义音频、BGM 和字体会被规范化为服务层可使用的路径或名称；本地素材 仅解析来源和扩展名，实际复制由 ``_prepare_cli_materials`` 在所有批量条目 校验通过后统一完成。 """ from app.models import const from app.services import bgm as bgm_service from app.utils import utils # FFmpeg 探测已经移到 app/services/task.py 的共享任务流水线（task.start） # 里统一做硬性检查：探测失败会让任务以 preflight 阶段失败结束，run_cli() # 会据此返回非零退出码。这里不再重复一次不阻断流程的检查，避免与流水线 # 里的判断结果不一致。 local_material_extensions = { *(f".{extension}" for extension in const.FILE_TYPE_VIDEOS), *(f".{extension}" for extension in const.FILE_TYPE_IMAGES), ".avi", ".flv", } if params.custom_audio_file and stop_at not in {"script", "terms"}: params.custom_audio_file = _resolve_cli_file( params.custom_audio_file, description="custom audio", ) audio_extension = os.path.splitext(params.custom_audio_file)[1].lower() if audio_extension not in _CUSTOM_AUDIO_EXTENSIONS: allowed = ", ".join(sorted(_CUSTOM_AUDIO_EXTENSIONS)) raise ValueError( f"unsupported custom audio type {audio_extension or ''}; " f"allowed extensions: {allowed}" ) if params.bgm_type == "custom" and stop_at == "video": if not bgm_service.should_use_bgm(params.bgm_type, params.bgm_volume): # 0 音量时下游会统一跳过所有 BGM。这里同时清空文件参数，避免 # CLI 为一个不会被读取的文件执行路径解析、存在性检查或格式 # 校验。 params.bgm_file = "" elif not params.bgm_file: # 缺少文件是否构成错误取决于通用 BGM 开关，不能在 argparse 阶段 # 无条件拦截，否则 ``custom + 0%`` 会和 WebUI、服务层行为不一致。 raise ValueError("--bgm-file is required when --bgm-type is custom") else: try: # CLI、WebUI 和任务服务必须共用同一个 BGM 文件边界。这里直接 # 复用服务层解析，既支持用户上传目录和内置歌曲目录，也 # 自动继承新增音频格式及路径安全规则，避免多个入口分别 # 维护白名单。 params.bgm_file = bgm_service.resolve_bgm_file(params.bgm_file) except ValueError as exc: supported_extensions = ", ".join( bgm_service.SUPPORTED_BGM_EXTENSIONS ) raise ValueError( "background music must be a supported audio file inside " f"storage/bgm or resource/songs ({supported_extensions}): " f"{params.bgm_file}" ) from exc if params.subtitle_enabled and params.font_name and stop_at == "video": font_path = _resolve_managed_resource_file( params.font_name, resource_dir=utils.font_dir(), description="subtitle font", ) if not font_path.lower().endswith((".ttf", ".ttc")): raise ValueError("subtitle font must use the .ttf or .ttc extension") # 下游根据 resource/fonts 内的文件名拼接路径，因此仍保留纯文件名。 params.font_name = os.path.basename(font_path) if params.video_source != "local" or stop_at not in {"materials", "video"}: return "", [] local_videos_dir = utils.storage_dir("local_videos") resolved_materials: list[tuple[MaterialInfo, str, str]] = [] for material in params.video_materials or []: source_path = _resolve_cli_file( material.url, description="local material", fallback_dir=local_videos_dir, ) extension = os.path.splitext(source_path)[1].lower() if extension not in local_material_extensions: allowed = ", ".join(sorted(local_material_extensions)) raise ValueError( f"unsupported local material type {extension or ''}: " f"{material.url}; allowed extensions: {allowed}" ) resolved_materials.append((material, source_path, extension)) return local_videos_dir, resolved_materials def _remove_cli_material_copies(created_paths: Sequence[str]) -> None: """Best-effort cleanup for managed copies created by one CLI preparation.""" for file_path in reversed(created_paths): try: if os.path.isfile(file_path): os.remove(file_path) except OSError as exc: logger.warning( f"failed to remove prepared CLI material: path={file_path}, " f"error={exc}" ) def _prepare_cli_materials( local_videos_dir: str, resolved_materials: Sequence[tuple[MaterialInfo, str, str]], *, prepared_paths: dict[str, str] | None = None, created_paths: list[str] | None = None, ) -> None: """Copy validated local materials once and update every matching reference.""" if not resolved_materials: return os.makedirs(local_videos_dir, exist_ok=True) if prepared_paths is None: prepared_paths = {} if created_paths is None: created_paths = [] for material, source_path, extension in resolved_materials: prepared_path = prepared_paths.get(source_path) if prepared_path is None: if _path_is_within_directory(source_path, local_videos_dir): prepared_path = source_path else: prepared_path = os.path.join( local_videos_dir, f"cli-material-{uuid4().hex}{extension}", ) try: shutil.copy2(source_path, prepared_path) except OSError: try: if os.path.exists(prepared_path): os.remove(prepared_path) except OSError: pass raise created_paths.append(prepared_path) logger.info( "copied CLI local material into managed storage: " f"source={source_path}, target={prepared_path}" ) prepared_paths[source_path] = prepared_path material.url = prepared_path def prepare_cli_files(params: VideoParams, stop_at: str) -> None: """Validate and prepare files for one trusted local CLI task.""" local_videos_dir, resolved_materials = _validate_cli_files(params, stop_at) created_paths: list[str] = [] try: _prepare_cli_materials( local_videos_dir, resolved_materials, created_paths=created_paths, ) except Exception: _remove_cli_material_copies(created_paths) raise def _run_batch_tasks(args: argparse.Namespace, tasks: list[VideoParams]) -> int: from app.services import task as tm from app.utils import utils task_summaries: list[dict[str, Any]] = [] used_task_ids: set[str] = set() failed_count = 0 for index, params in enumerate(tasks, start=1): task_id = utils.get_uuid() if task_id in used_task_ids: task_id = str(uuid4()) used_task_ids.add(task_id) logger.info( f"start CLI batch task: index={index}, task_id={task_id}, " f"stop_at={args.stop_at}" ) result = None failed_stage = None error = None try: result = tm.start( task_id=task_id, params=params, stop_at=args.stop_at, allow_server_file_input=True, ) except Exception as exc: failed_stage = "runtime" error = str(exc) logger.exception( "CLI batch task failed with an unexpected error: " f"index={index}, task_id={task_id}, error={exc}" ) else: if not isinstance(result, dict) or not result: failed_stage = "unknown" error = "empty or invalid task result" elif result.get("state") == tm.const.TASK_STATE_FAILED: failed_stage = result.get("failed_stage") or "unknown" error = result.get("error") or "unknown task error" if error is not None: logger.error( f"CLI batch task failed: index={index}, task_id={task_id}, " f"stop_at={args.stop_at}, stage={failed_stage}, error={error}" ) status = "failed" if error is not None else "succeeded" if status == "failed": failed_count += 1 task_summaries.append( { "index": index, "task_id": task_id, "status": status, "result": result, "failed_stage": failed_stage, "error": error, } ) print( json.dumps( { "total": len(task_summaries), "succeeded": len(task_summaries) - failed_count, "failed": failed_count, "tasks": task_summaries, }, ensure_ascii=False, ) ) return 1 if failed_count else 0 def run_cli(argv: Sequence[str] | None = None) -> int: args = parse_args(argv) if args.batch_file: try: tasks = _build_batch_tasks(args) except (ValueError, OSError) as exc: logger.error(f"invalid CLI batch input: {exc}") return 2 return _run_batch_tasks(args, tasks) try: params = build_video_params(args) prepare_cli_files(params, stop_at=args.stop_at) except (ValueError, OSError) as exc: logger.error(f"invalid CLI input: {exc}") return 2 # 帮助参数会在 parse_args 中直接退出。把业务服务延迟到这里导入， # 保证 -h/--help 输出干净，同时不改变实际任务的初始化流程。 from app.services import task as tm from app.utils import utils task_id = args.task_id or utils.get_uuid() logger.info(f"start CLI task: task_id={task_id}, stop_at={args.stop_at}") try: result = tm.start( task_id=task_id, params=params, stop_at=args.stop_at, # CLI inputs come from the local operator rather than an HTTP client. # Preserve support for arbitrary local audio paths without weakening the # task service's secure default for API and WebUI callers. allow_server_file_input=True, ) except Exception as exc: logger.exception( f"CLI task failed with an unexpected error: task_id={task_id}, error={exc}" ) return 1 if not result or result.get("state") == tm.const.TASK_STATE_FAILED: failed_stage = result.get("failed_stage", "unknown") if result else "unknown" error = result.get("error", "unknown task error") if result else "empty result" logger.error( f"CLI task failed: task_id={task_id}, stop_at={args.stop_at}, " f"stage={failed_stage}, error={error}" ) if result and result.get("videos"): print(json.dumps({"task_id": task_id, "result": result}, ensure_ascii=False)) return 1 print(json.dumps({"task_id": task_id, "result": result}, ensure_ascii=False)) return 0 def _force_utf8_console() -> None: """Make stdout/stderr UTF-8 before anything is printed. Windows consoles default to a legacy code page (cp1252 in Western Europe). Generating a French video produces U+202F, the narrow no-break space French typography puts before ':' and '!', and Loguru's own progress lines carry circled digits. Either one raises UnicodeEncodeError, which kills the process *after* the video was written successfully -- so the run reports failure and never prints where the file is. """ for stream in (sys.stdout, sys.stderr): reconfigure = getattr(stream, "reconfigure", None) if reconfigure is None: continue try: reconfigure(encoding="utf-8", errors="replace") except (ValueError, OSError): continue if __name__ == "__main__": _force_utf8_console() raise SystemExit(run_cli())