import hashlib
import html
import json
import math
import mimetypes
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
import webbrowser
from collections.abc import Mapping
from datetime import datetime
from pathlib import Path
from uuid import UUID, uuid4
import requests
import streamlit as st
from loguru import logger
from streamlit_tour import Tour
# WebUI 作为独立入口运行时,需要让项目根目录优先于第三方依赖,
# 避免依赖中的同名 app 包遮蔽 MoneyPrinterTurbo 自己的 app 包。
root_dir = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
if root_dir in sys.path:
sys.path.remove(root_dir)
sys.path.insert(0, root_dir)
from app.config import config
from app.models import const
from app.models.llm_provider import (
DEFAULT_LLM_PROVIDER_ID,
LLM_PROVIDER_REGISTRY,
get_llm_provider,
normalize_provider_override,
)
from app.models.schema import (
ImageMotion,
MaterialInfo,
VideoAspect,
VideoConcatMode,
VideoFitMode,
VideoParams,
VideoTransitionMode,
)
from app.services import bgm as bgm_service
from app.services import material_upload as material_upload_service
from app.services import (
cache_manager,
llm,
loomloom,
material,
metaso_minimax,
muapi,
infistar_video,
jalapeno_video,
ofox,
subtitle,
video,
volcengine_seedance,
voice,
webui_task,
)
from app.services import elevenlabs_music as elevenlabs_music_service
from app.services import sonilo as sonilo_service
from app.services import state as sm
from app.services import task as tm
from app.services import version_checker
from app.utils.logging_utils import (
LOG_LEVELS,
configure_terminal_logger,
get_logging_options,
)
from app.utils import utils
from app.utils.request_logging import trace_http_call
st.set_page_config(
page_title="MoneyPrinterTurbo",
page_icon="🤖",
layout="wide",
initial_sidebar_state="auto",
menu_items={
"Report a bug": "https://github.com/harry0703/MoneyPrinterTurbo/issues",
"About": "# MoneyPrinterTurbo\nSimply provide a topic or keyword for a video, and it will "
"automatically generate the video copy, video materials, video subtitles, "
"and video background music before synthesizing a high-definition short "
"video.\n\nhttps://github.com/harry0703/MoneyPrinterTurbo",
},
)
# Streamlit 1.59 会在页面右上角默认展示 Deploy、skills nudge 等平台入口。
# MoneyPrinterTurbo 是面向终端用户的本地工具,这些入口会造成顶部大块空白,
# 也会让新用户误以为需要安装额外组件。这里统一隐藏 Streamlit 平台工具栏,
# 并压缩主容器顶部留白,只保留项目自己的标题、语言选择和业务设置区域。
style_file = Path(__file__).with_name("styles.css")
streamlit_style = f""
st.markdown(streamlit_style, unsafe_allow_html=True)
# 定义资源目录
font_dir = utils.font_dir()
song_dir = utils.song_dir()
i18n_dir = os.path.join(root_dir, "webui", "i18n")
config_file = os.path.join(root_dir, "webui", ".streamlit", "webui.toml")
# 语言列表必须在会话状态初始化前可用,首次访问时才能把浏览器 locale 映射到
# 项目真正支持的语言;自动识别结果只进入当前会话,不修改全局配置。
locales = utils.load_locales(i18n_dir)
DEFAULT_CHATTERBOX_BASE_URL = "http://127.0.0.1:4123/v1"
DEFAULT_CHATTERBOX_MODEL = "chatterbox"
DEFAULT_CHATTERBOX_VOICES = ["default-Female"]
DEFAULT_KOKORO_BASE_URL = "http://127.0.0.1:8880/v1"
DEFAULT_KOKORO_MODEL = "kokoro"
# empty = ask the server for its voice list (GET {base_url}/audio/voices)
DEFAULT_KOKORO_VOICES: list[str] = []
DEFAULT_VOXCPM_BASE_URL = voice.VOXCPM_DEFAULT_BASE_URL
DEFAULT_VOXCPM_VOICE = voice.VOXCPM_DEFAULT_VOICE
VOXCPM_REFERENCE_AUDIO_SESSION_KEY = "voxcpm_reference_audio"
VOXCPM_REFERENCE_AUDIO_ERROR_SESSION_KEY = "voxcpm_reference_audio_error"
VOXCPM_PROMPT_AUDIO_SESSION_KEY = "voxcpm_prompt_audio"
VOXCPM_PROMPT_AUDIO_ERROR_SESSION_KEY = "voxcpm_prompt_audio_error"
VOXCPM_PROMPT_TEXT_SESSION_KEY = "voxcpm_prompt_text_input"
VOXCPM_HIGH_FIDELITY_SESSION_KEY = "voxcpm_high_fidelity_enabled"
VOXCPM_SEPARATE_PROMPT_AUDIO_SESSION_KEY = "voxcpm_separate_prompt_audio_enabled"
VOXCPM_PROMPT_EXAMPLE_MODE_SESSION_KEY = "voxcpm_prompt_example_mode"
ONBOARDING_TOUR_KEY = "mpt-onboarding-v1"
CUSTOM_LLM_ENDPOINT_ID = "custom"
VOICE_MODE_TTS = "tts"
VOICE_MODE_UPLOAD = "upload"
VOICE_MODE_NONE = "none"
LOOMLOOM_MAX_POLL_FAILURES = 5
# WebUI 按素材能力分组展示视频来源,但底层仍保存原有 video_source 值。
# AI 视频组与设置页共用同一业务顺序:合作服务商优先,并按秘塔、OFox、胜算云、
# 火山引擎、无限星河排列;其余服务随后展示。这样两个入口的顺序一致,同时
# 不改变 config.toml、历史任务和 API 请求中的字段语义,旧用户无需迁移配置。
VIDEO_SOURCE_GROUPS = {
"stock_video": ("pexels", "pixabay", "coverr"),
"ai_video": (
"metaso_minimax",
"ofox",
"loomloom",
"volcengine_seedance",
"infistar",
"jalapeno",
"wavespeed",
"muapi",
),
"ai_image": ("openai_image", "codex_image"),
"local": ("local",),
}
# Upload-Post 的 API Key 与发布用户分别在两个页面管理,并且发布用户名称
# 不等于登录邮箱。集中维护入口可以避免多语言文案各自硬编码 URL 后发生偏差,
# 也方便用户从 WebUI 直接完成首次配置和后续账号维护。
UPLOAD_POST_API_KEYS_URL = "https://app.upload-post.com/api-keys"
UPLOAD_POST_MANAGE_USERS_URL = "https://app.upload-post.com/manage-users"
# 素材设置与视频来源说明共用推广入口,避免两个位置的链接参数不一致。
OFOX_REFERRAL_URL = (
"https://ofox.ai/?utm_source=github"
"&utm_medium=sponsorship&utm_content=moneyprinterturbo"
)
# “默认”是 WebUI 专用哨兵,不会写入 config.toml,也不会传给 FFmpeg。
# 后端在 video_codec 未配置时继续采用稳定的 libx264;单独保留该哨兵可以区分
# “跟随项目默认策略”和“用户明确固定 libx264”,便于未来安全调整默认策略。
DEFAULT_VIDEO_CODEC_OPTION = "__default__"
# LoomLoom 的能力接口只返回模型 ID 和展示名,不提供价格。这里仅维护用户确认过
# 的参考价,用于帮助选择模型;最终费用按实际模型调用结算。别名同时覆盖
# 当前展示名和常见模型 ID,未收录的新模型会自然返回空价格,不影响选择或报价。
LOOMLOOM_VIDEO_MODEL_PRICES = (
(("veo31fast", "googleveo31fastpreview"), "¥0.700/秒", "¥0.700/秒"),
(
(
"通义万相22图生视频fastlora",
"通义万相22文生视频fastlora",
"tongyiwanxiang22i2vfastlora",
"tongyiwanxiang22t2vfastlora",
"wanx22i2vfastlora",
"wanx22t2vfastlora",
),
"¥0.350–0.770/条",
"¥0.350/条(480P);¥0.770/条(720P)",
),
(("即梦30文生视频720p", "jimeng30t2v720p"), "¥0.230/秒", "¥0.230/秒"),
(("即梦30pro视频", "jimeng30pro视频", "jimeng30provideo"), "¥1.000/秒", "¥1.000/秒"),
(("veo3", "googleveo3"), "¥1.400/秒", "¥1.400/秒"),
(("veo31", "googleveo31"), "¥1.400/秒", "¥1.400/秒"),
(
("klingv2", "可灵v2"),
"¥10.00–20.00/条",
"¥10.00/条(5 秒);¥20.00/条(10 秒)",
),
(
("klingv21master", "可灵v21master"),
"¥10.00–20.00/条",
"¥10.00/条(5 秒);¥20.00/条(10 秒)",
),
(
("viduq3pro",),
"¥0.440–1.000/秒",
"¥0.440/秒(540P);¥0.940/秒(720P);¥1.000/秒(1080P)",
),
)
DEFAULT_SUBTITLE_SETTINGS = {
"subtitle_enabled": True,
"font_name": "MicrosoftYaHeiBold.ttc",
"subtitle_position": "bottom",
"subtitle_display_mode": "sentence",
"subtitle_animation": "none",
"custom_position": 70.0,
"text_fore_color": "#FFFFFF",
"font_size": 60,
"stroke_color": "#000000",
"stroke_width": 1.5,
"subtitle_background_enabled": False,
"subtitle_background_color": "#000000",
"rounded_subtitle_background": False,
}
LOCAL_MATERIAL_EXTENSIONS = set(material_upload_service.SUPPORTED_MATERIAL_EXTENSIONS)
CUSTOM_AUDIO_EXTENSIONS = {".mp3", ".wav", ".m4a", ".aac", ".flac", ".ogg"}
_FINAL_VIDEO_PATTERN = re.compile(
r"^final-(?P\d+)\.(?Pmp4|mov|mkv|webm)$",
re.IGNORECASE,
)
# 与 app/services/bgm.py、app/services/material_upload.py 的上传文件名规则共用
# 同一份清单:除路径分隔符与 Win32 非法字符外,还要移除所有控制符和会改变显示
# 形态的字符。此前只拦截 ord < 32(C0 控制符),漏掉了同一类问题的另一半:
# * C1 控制符 U+007F-U+009F:会原样进入下载文件名、日志和界面。
# * 双向文本控制符 U+200E/U+200F/U+202A-U+202E/U+2066-U+2069:U+202E 之后的
# 字符会被反向渲染,``holiday4pm`` 看起来像另一个扩展名。
# * U+2028/U+2029(Unicode 行/段分隔符)能在一行日志里制造视觉换行。
# 视频主题是用户自由输入的文本,可以直接粘贴这些字符,因此必须在这里过滤。
# 字符串不使用 r'' 前缀:``\\`` 表示要排除的反斜杠本身,``\x``/``\u`` 转义由
# Python 解析,避免把转义文本写进字符集。
_DOWNLOAD_FILENAME_INVALID_PATTERN = re.compile(
"[<>:\"/\\\\|?*\\x00-\\x1f\\x7f-\\x9f"
"\\u200e\\u200f\\u2028\\u2029\\u202a-\\u202e\\u2066-\\u2069]"
)
_WINDOWS_RESERVED_FILENAMES = frozenset(
{"CON", "PRN", "AUX", "NUL"}
| {
f"{prefix}{number}"
for prefix in ("COM", "LPT")
for number in range(1, 10)
}
# Win32 还会把 Latin-1 上标数字 ¹、²、³ 识别为设备编号。虽然这类主题
# 很少见,但仍会导致 Windows 下载失败,因此与普通数字保留名统一处理。
| {
f"{prefix}{number}"
for prefix in ("COM", "LPT")
for number in ("¹", "²", "³")
}
)
_RUNTIME_CONFIG_SECTIONS = {
"app": config.app,
"azure": config.azure,
"chatterbox": config.chatterbox,
"kokoro": config.kokoro,
"elevenlabs": config.elevenlabs,
"minimax_tts": config.minimax_tts,
"siliconflow": config.siliconflow,
"fish_audio": config.fish_audio,
"voxcpm": config.voxcpm,
"ui": config.ui,
}
# 设置预设与密钥备份使用各自的文件标识。导入时先校验 schema 和版本,
# 避免把任务记录、config.toml 或其它 JSON 误当成本功能的导出文件。
SETTINGS_PRESET_SCHEMA = "moneyprinterturbo.settings-preset"
SETTINGS_PRESET_VERSION = 1
SETTINGS_PRESET_FILE_NAME = "moneyprinterturbo-settings.json"
KEY_BACKUP_SCHEMA = "moneyprinterturbo.key-backup"
KEY_BACKUP_VERSION = 1
KEY_BACKUP_FILE_NAME = "moneyprinterturbo-keys.json"
# Export files contain only settings or credentials, not media. Reject oversized
# uploads before decoding and parsing them in the Streamlit process.
MAX_SETTINGS_TRANSFER_BYTES = 2 * 1024 * 1024
# 预设只描述生成参数。素材、配音和配乐都是本机文件路径,预设通常要在另一台
# 机器或另一个容器里导入,带上这些路径只会指向不存在的文件。
PRESET_EXCLUDED_PARAM_KEYS = frozenset(
{
"video_materials",
"custom_audio_file",
"bgm_file",
}
)
# 密钥按配置项名称后缀识别。新增 Provider 只要沿用现有命名,就会自动进入
# 备份,不需要再维护第二份密钥清单。
CREDENTIAL_KEY_SUFFIXES = (
"api_key",
"api_keys",
"api_token",
"access_key",
"secret_key",
"speech_key",
)
# 只恢复密钥而不恢复配套配置项时,凭据仍然不可用。这些配套项与密钥一起备份。
CREDENTIAL_COMPANION_KEYS = {
# Azure 语音必须同时知道区域。
"azure": ("speech_region",),
# Provider 的额外字段由 Registry 声明,例如 Cloudflare AI Gateway 的
# Account ID 和 Gateway ID。只恢复 API Key 而丢掉这些字段时,换到另一台
# 机器后该 Provider 仍然无法调用。从 Registry 读取可以让以后新增的
# Provider 自动进入备份,不需要在这里维护第二份字段清单。
"app": tuple(
provider.config_key(field.config_suffix)
for provider in LLM_PROVIDER_REGISTRY
for field in provider.extra_fields
),
}
NON_LLM_COMPANION_KEYS = {
"app": ("upload_post_username",)
}
# 同一个密钥在不同面板可能使用各自的控件 key:音频面板直接编辑 Gemini 和
# MiMo 的 LLM 密钥。恢复备份时必须清除每一个别名,否则遗留的旧值
# 会在下一次 rerun 覆盖刚刚恢复的密钥。
CREDENTIAL_WIDGET_STATE_ALIASES = {
("app", "gemini_api_key"): ("gemini_tts_api_key_input",),
("app", "mimo_api_key"): ("mimo_tts_api_key_input",),
}
# ui 分区只保存界面偏好,不含任何凭据,备份时整体跳过。
KEY_BACKUP_EXCLUDED_SECTIONS = frozenset({"ui"})
# -----------------------------------------------------------------------------
# 启动配置、会话状态与本地化
# -----------------------------------------------------------------------------
def _set_runtime_config(section_name, key, value):
"""
更新 WebUI 配置,但不等待正在生成视频的后台任务。
后台任务结束前,配置层只保留同一配置项的最新值;任务释放配置锁时会自动
应用并保存。页面控件值仍由 Streamlit session_state 维护,因此暂存期间的
rerun 不会把用户刚输入的内容重置为旧配置。
"""
config_section = _RUNTIME_CONFIG_SECTIONS[section_name]
updated = config.update_config_nonblocking(config_section, key, value)
if not updated:
logger.debug(f"deferred WebUI config update: section={section_name}, key={key}")
return updated
def _delete_runtime_config(section_name, key):
"""删除 WebUI 配置项;后台任务占用配置时延后执行。"""
config_section = _RUNTIME_CONFIG_SECTIONS[section_name]
deleted = config.delete_config_nonblocking(config_section, key)
if not deleted:
logger.debug(f"deferred WebUI config delete: section={section_name}, key={key}")
return deleted
def _save_runtime_config():
"""请求保存 WebUI 配置;后台任务占用配置时立即返回。"""
saved = config.try_save_config()
if not saved:
logger.debug("deferred WebUI config save until active task completes")
return saved
def _saved_ui_choice(key, options, default, section=None):
"""读取一个持久化选择,并把旧配置或手工编辑的非法值降级为默认值。"""
options = list(options)
section = config.ui if section is None else section
saved = section.get(key, default)
numeric_default = isinstance(default, (int, float)) and not isinstance(
default, bool
)
# bool 是 int 的子类,``True == 1``。手工把数值选项写成 TOML
# 布尔值时必须拒绝,不能让它伪装成第一个数值 option。
if numeric_default and isinstance(saved, bool):
return default
for option in options:
if saved == option:
# 返回 options 中的真实值,顺便把 TOML 1.0 等价归一化为
# 整数选项 1,避免下游参数类型随配置写法漂移。
return option
# TOML 中的数值通常保留原类型;仍兼容用户手工写成字符串的情况。
if numeric_default and isinstance(saved, str):
try:
converted = type(default)(saved)
except (TypeError, ValueError):
converted = None
for option in options:
if converted == option:
return option
return default
def _saved_ui_number(key, default, minimum, maximum, number_type=float):
"""读取并限幅持久化数值,避免非法配置破坏 Streamlit slider。"""
try:
saved = config.ui.get(key, default)
if isinstance(saved, bool):
raise ValueError("boolean is not a numeric setting")
value = number_type(saved)
if isinstance(value, float) and not math.isfinite(value):
raise ValueError("non-finite value")
except (TypeError, ValueError, OverflowError):
value = default
return min(maximum, max(minimum, value))
def _saved_ui_bool(key, default):
"""兼容 TOML 布尔值和常见手工字符串,拒绝含义不明的旧值。"""
value = config.ui.get(key, default)
if isinstance(value, bool):
return value
if isinstance(value, str):
normalized = value.strip().lower()
if normalized in {"1", "true", "yes", "on"}:
return True
if normalized in {"0", "false", "no", "off"}:
return False
return default
def _saved_ui_color(key, default):
"""只把标准六位十六进制颜色传给 Streamlit color picker。"""
value = str(config.ui.get(key, default) or "").strip()
if re.fullmatch(r"#[0-9a-fA-F]{6}", value):
return value
return default
def _saved_ui_text(key, default="", max_length=None):
"""读取持久化文本并遵守对应 WebUI 控件的长度上限。"""
value = str(config.ui.get(key, default) or default)
if max_length is not None:
value = value[:max_length]
return value
def _run_llm_read_operation(operation_name, operation):
"""
使用稳定的当前 LLM 配置执行只读请求,并避免等待视频生成任务。
能立即取得配置锁时继续沿用原来的互斥保护;锁已被后台视频任务持有时,
全局配置在任务结束前不会发生变化,因此可以安全复制当前配置,并叠加页面
尚未落盘的 Provider、模型和密钥。这样新文案使用界面中的最新选择,同时
不会改变正在生成的视频任务。
"""
with config.try_runtime_config_lock() as lock_acquired:
# 配置层在复制全局值和叠加待更新值期间持有队列锁,因此快照只能看到
# 更新前或更新后的完整状态,不会混用两组 Provider 参数。
app_config_snapshot = config.snapshot_config_with_pending(config.app)
if lock_acquired:
return operation(app_config_snapshot)
logger.info(
f"run read-only LLM operation with active task configuration: "
f"operation={operation_name}"
)
return operation(app_config_snapshot)
def _parse_chatterbox_voices(voices):
# Chatterbox 是自托管服务,音色列表由用户在 WebUI 中手动输入。
# 这里统一兼容 TOML 数组和输入框里的逗号分隔字符串,避免下拉框、
# 试听按钮和后续生成流程使用不同格式导致状态不一致。
if isinstance(voices, str):
return [v.strip() for v in voices.split(",") if v.strip()]
return [str(v).strip() for v in voices or [] if str(v).strip()]
def _sync_chatterbox_config_from_session_state():
# Streamlit 的按钮会触发整页 rerun,而 Chatterbox 配置输入框位于
# “试听语音合成”按钮之后。如果试听时只读取 config.chatterbox,可能拿不到
# 用户刚在输入框里填入的 base_url/model/voices。先从 session_state 同步一次,
# 可以保证按钮逻辑和输入框显示逻辑使用同一份最新配置。
_set_runtime_config(
"chatterbox",
"base_url",
(
st.session_state.get(
"chatterbox_base_url_input",
config.chatterbox.get("base_url") or DEFAULT_CHATTERBOX_BASE_URL,
)
or ""
).strip(),
)
_set_runtime_config(
"chatterbox",
"api_key",
st.session_state.get(
"chatterbox_api_key_input", config.chatterbox.get("api_key", "")
),
)
_set_runtime_config(
"chatterbox",
"model_id",
(
st.session_state.get(
"chatterbox_model_input",
config.chatterbox.get("model_id") or DEFAULT_CHATTERBOX_MODEL,
)
or DEFAULT_CHATTERBOX_MODEL
).strip(),
)
_set_runtime_config(
"chatterbox",
"voices",
_parse_chatterbox_voices(
st.session_state.get(
"chatterbox_voices_input",
config.chatterbox.get("voices") or DEFAULT_CHATTERBOX_VOICES,
)
),
)
def _sync_kokoro_config_from_session_state():
# 音色目录先于设置输入框渲染,先同步浏览器状态,确保本次 rerun 就使用
# 新端点和手工音色配置,不必再操作一次控件。
_set_runtime_config(
"kokoro",
"base_url",
(
st.session_state.get(
"kokoro_base_url_input",
config.kokoro.get("base_url") or DEFAULT_KOKORO_BASE_URL,
)
or ""
).strip(),
)
_set_runtime_config(
"kokoro",
"api_key",
st.session_state.get(
"kokoro_api_key_input", config.kokoro.get("api_key", "")
),
)
_set_runtime_config(
"kokoro",
"model_id",
(
st.session_state.get(
"kokoro_model_input",
config.kokoro.get("model_id") or DEFAULT_KOKORO_MODEL,
)
or DEFAULT_KOKORO_MODEL
).strip(),
)
_set_runtime_config(
"kokoro",
"voices",
_parse_chatterbox_voices(
st.session_state.get(
"kokoro_voices_input",
config.kokoro.get("voices") or DEFAULT_KOKORO_VOICES,
)
),
)
def _get_kokoro_voice_options(saved_voice_name: str) -> list[str]:
"""会话内短缓存远端目录,断线时保留上次选择,不把故障当成用户改音色。"""
if config.kokoro.get("voices"):
return voice.get_kokoro_voices()
# 仅保留当前服务的一条缓存。更换端点/凭据立即重查,缓存不保存明文 Key;
# 30 秒内的其他 UI 操作不重复阻塞 5 秒等待一个已知离线的服务。
signature = (
(config.kokoro.get("base_url") or "").strip().rstrip("/"),
_credential_signature(config.kokoro.get("api_key", "")),
)
catalog = st.session_state.get("kokoro_voice_catalog", {})
if catalog.get("signature") != signature:
catalog = {"signature": signature, "voices": [], "checked_at": None}
now = time.monotonic()
if catalog["checked_at"] is None or now - catalog["checked_at"] >= 30:
fetched = voice.get_kokoro_voices(fallback=False)
catalog.update(checked_at=now, available=bool(fetched))
if fetched:
catalog["voices"] = fetched
st.session_state["kokoro_voice_catalog"] = catalog
options = list(catalog["voices"])
if not catalog["available"]:
st.warning(tr("Kokoro Voices Unavailable"))
# 首次打开时可能没有缓存,仍保留配置文件中的真实选择;恢复连接后
# 只有成功返回的新目录才能判定某个旧音色确实已被服务器删除。
if voice.is_kokoro_voice(saved_voice_name) and saved_voice_name not in options:
options.insert(0, saved_voice_name)
return options or [f"kokoro:{voice.KOKORO_DEFAULT_VOICE}"]
def _detect_audio_mime(audio_file: str, audio_bytes: bytes) -> str:
# 有些 OpenAI-compatible TTS 服务,例如 travisvn/chatterbox-tts-api,
# 即使请求 response_format=mp3,也会返回 WAV 内容。WebUI 试听如果固定
# 使用 audio/mp3,浏览器可能无法播放,因此这里按文件头识别真实格式。
header = audio_bytes[:12]
if header.startswith(b"RIFF") and header[8:12] == b"WAVE":
return "audio/wav"
if header.startswith(b"ID3") or header[:2] in (
b"\xff\xfb",
b"\xff\xf3",
b"\xff\xf2",
):
return "audio/mp3"
if header.startswith(b"OggS"):
return "audio/ogg"
ext = os.path.splitext(audio_file)[1].lower()
return {
".wav": "audio/wav",
".m4a": "audio/mp4",
".aac": "audio/aac",
".ogg": "audio/ogg",
".flac": "audio/flac",
}.get(ext, "audio/mp3")
def _build_uploaded_file_path(uploaded_file, target_dir, allowed_extensions, prefix):
"""为浏览器上传文件生成受控的服务端保存路径。"""
original_name = os.path.basename(str(uploaded_file.name or ""))
extension = os.path.splitext(original_name)[1].lower()
if extension not in allowed_extensions:
logger.warning(
f"reject unsupported uploaded file extension: {original_name or ''}"
)
raise ValueError("unsupported uploaded file type")
normalized_target_dir = os.path.realpath(target_dir)
os.makedirs(normalized_target_dir, exist_ok=True)
# 不复用浏览器传入的文件名,避免路径分隔符、控制字符或同名覆盖。UUID 只用于
# 服务端落盘,不改变用户在上传控件中看到的原始名称。
file_path = os.path.realpath(
os.path.join(normalized_target_dir, f"{prefix}-{uuid4().hex}{extension}")
)
if os.path.commonpath([normalized_target_dir, file_path]) != normalized_target_dir:
logger.warning(f"invalid uploaded file path: {file_path}")
raise ValueError("invalid uploaded file path")
return file_path
def _save_uploaded_local_materials(uploaded_files):
"""Validate a WebUI material batch and undo earlier files on failure."""
local_videos_dir = utils.storage_dir("local_videos", create=True)
materials = []
persisted = []
saved_paths = []
try:
for uploaded_file in uploaded_files:
stored_name = material_upload_service.save_material_upload(
uploaded_file.name, uploaded_file
)
file_path = os.path.join(local_videos_dir, stored_name)
saved_paths.append(file_path)
material_info = MaterialInfo()
material_info.provider = "local"
material_info.url = file_path
materials.append(material_info)
persisted.append(
{
"provider": material_info.provider,
"url": material_info.url,
"duration": material_info.duration,
}
)
except Exception:
for file_path in saved_paths:
try:
os.remove(file_path)
except OSError as exc:
logger.warning(
f"failed to remove local material after batch error: "
f"path={file_path}, error={exc}"
)
raise
return materials, persisted
def _initialize_session_state():
"""集中初始化跨 rerun 保留的页面状态。"""
if not st.session_state.get("cross_post_recovery_checked"):
# WebUI 可以不经过 FastAPI 独立运行,因此也需要在首次会话初始化时处理
# 进程重启留下的发布状态。恢复失败时不写标记,后续 rerun 会再次尝试。
recovered = tm.recover_interrupted_cross_posts()
if recovered is not None:
st.session_state["cross_post_recovery_checked"] = True
saved_ui_language = config.ui.get("language", "")
browser_locale = st.context.locale
initial_ui_language = utils.resolve_ui_language(
saved_language=saved_ui_language,
browser_locale=browser_locale,
supported_languages=locales.keys(),
)
defaults = {
"video_subject": "",
"video_script": "",
"video_terms": "",
"paragraph_number_input": _saved_ui_number(
"paragraph_number",
1,
llm.MIN_SCRIPT_PARAGRAPH_NUMBER,
llm.MAX_SCRIPT_PARAGRAPH_NUMBER,
int,
),
"video_script_prompt": _saved_ui_text(
"video_script_prompt",
max_length=llm.MAX_SCRIPT_PROMPT_LENGTH,
),
"custom_system_prompt": _saved_ui_text(
"custom_system_prompt",
llm.DEFAULT_SCRIPT_SYSTEM_PROMPT,
llm.MAX_SCRIPT_SYSTEM_PROMPT_LENGTH,
),
"match_materials_to_script": bool(
config.app.get("match_materials_to_script", False)
),
"custom_bgm_file_input": _saved_ui_text("custom_bgm_file"),
"sonilo_bgm_prompt_input": _saved_ui_text(
"sonilo_bgm_prompt",
max_length=sonilo_service.MAX_PROMPT_LENGTH,
),
"elevenlabs_music_prompt_input": _saved_ui_text(
"elevenlabs_music_prompt",
max_length=elevenlabs_music_service.MAX_PROMPT_LENGTH,
),
"subtitle_enabled_checkbox": _saved_ui_bool("subtitle_enabled", True),
"stroke_color_picker": _saved_ui_color("stroke_color", "#000000"),
"stroke_width_slider": _saved_ui_number(
"stroke_width", 1.5, 0.0, 10.0
),
"loomloom_candidate_count": _saved_ui_number(
"loomloom_candidate_count",
3,
1,
loomloom.MAX_SCRIPT_CANDIDATES,
int,
),
"loomloom_script_duration_seconds": _saved_ui_number(
"loomloom_script_duration_seconds", 60, 10, 600, int
),
"ui_language": initial_ui_language,
# 已落盘的本地素材允许用户只修改文案后继续复用。
"local_video_materials": [],
# 生成按钮回调先登记任务,使顶部入口能立即显示运行中数量。
"active_generation_tasks": {},
# 最近一次从当前页面提交的任务。生成改为后台执行后,页面 Fragment
# 通过这个 ID 查询状态;刷新时不再依赖正在执行的旧页面脚本。
"current_generation_task_id": "",
# LoomLoom 询价与执行必须跨 Streamlit rerun 保留完全相同的输入和
# clientRequestId,避免网络重试产生重复付费任务。
"loomloom_script_batch": None,
"loomloom_script_quote": None,
"loomloom_script_input_signature": "",
"loomloom_client_request_id": "",
"loomloom_run_id": "",
"loomloom_run_status": "",
"loomloom_run_error": "",
"loomloom_poll_failure_count": 0,
"loomloom_poll_retry_after": 0.0,
"loomloom_poll_paused": False,
"loomloom_script_candidates": (),
"loomloom_candidate_errors": (),
"loomloom_selected_candidate": 0,
"loomloom_video_batch": None,
"loomloom_video_quote": None,
"loomloom_video_input_signature": "",
"loomloom_video_client_request_id": "",
"loomloom_video_confirm_charge": False,
"loomloom_video_quote_error_signature": "",
"loomloom_video_quote_error": "",
"loomloom_video_capability": None,
"loomloom_video_capability_fingerprint": "",
"loomloom_video_capability_load_attempt": "",
"loomloom_video_capability_error": "",
"loomloom_video_model_id": "",
# 文案或完整配音刚生成时,在视频数量控件创建前消费这个摘要并自动
# 填入推荐素材数;消费后即清空,避免覆盖用户后续手动调整。
"loomloom_video_scene_autofill_digest": "",
"wavespeed_confirm_charge": False,
"volcengine_seedance_confirm_charge": False,
"infistar_confirm_charge": False,
"jalapeno_confirm_charge": False,
"ofox_confirm_charge": False,
"metaso_minimax_confirm_charge": False,
"muapi_confirm_charge": False,
# AI 视频按素材段计费,默认只生成一段,用户确认效果后再主动增加数量。
"loomloom_video_scene_count": _saved_ui_number(
"loomloom_video_scene_count",
1,
1,
loomloom.MAX_VIDEO_SCENES,
int,
),
}
for key, value in defaults.items():
st.session_state.setdefault(key, value)
_initialize_session_state()
def tr(key):
loc = locales.get(st.session_state["ui_language"], {})
value = loc.get("Translation", {}).get(key)
if value is not None:
return value
# 新功能优先维护中英文。其它语言缺少单项翻译时统一回退英文,避免在多个
# locale 中复制相同英文后长期失去同步;英文也没有该键时才显示原始 key。
return locales.get("en", {}).get("Translation", {}).get(key, key)
# -----------------------------------------------------------------------------
# 任务管理:历史扫描、运行状态、参数恢复与列表交互
# -----------------------------------------------------------------------------
def _format_task_time(timestamp):
if not timestamp:
return "-"
return datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M")
def _format_task_subject(subject, max_length=30):
subject = str(subject or "").replace("\n", " ").strip()
if len(subject) <= max_length:
return subject or "-"
return f"{subject[:max_length]}..."
def _safe_load_task_script(task_path):
script_file = os.path.join(task_path, "script.json")
if not os.path.isfile(script_file):
return {}
try:
with open(script_file, "r", encoding="utf-8") as f:
payload = json.load(f)
if not isinstance(payload, dict):
logger.warning(f"task script data is not an object: {script_file}")
return {}
return payload
except Exception as e:
logger.warning(f"failed to read task script data: {script_file}, {e}")
return {}
def _find_final_task_video(task_path: str) -> str:
"""
返回任务目录中序号最小的最终成片。
合成流程还会产生 combined、temp-clip 和 MoviePy 临时文件,这些文件不能
表示任务已成功完成,因此这里只接受 ``final-<序号>.<扩展名>``。
"""
try:
files = os.listdir(task_path)
except OSError:
return ""
candidates = []
for file_name in files:
match = _FINAL_VIDEO_PATTERN.fullmatch(file_name)
if match and os.path.isfile(os.path.join(task_path, file_name)):
candidates.append((int(match.group("index")), file_name))
if not candidates:
return ""
_, file_name = min(candidates, key=lambda item: item[0])
return os.path.join(task_path, file_name)
def _build_restore_upload_requirements(params: Mapping) -> dict:
"""
记录历史任务中无法由 Streamlit 自动恢复的上传文件依赖。
浏览器不允许程序重新填充 file_uploader,因此恢复任务时需要单独记录本地
素材和自定义音频依赖,并在用户重新生成前检查是否已经主动补充或替换。
"""
return {
"local_materials": params.get("video_source") == "local",
"custom_audio": bool(params.get("custom_audio_file")),
"original_voice_name": params.get("voice_name") or "",
}
def _get_unmet_restore_upload_requirements(
requirements: Mapping | None,
*,
video_source: str,
voice_name: str,
has_local_materials: bool,
has_custom_audio: bool,
voice_mode: str | None = None,
) -> set[str]:
"""返回当前表单仍未满足的历史上传文件依赖。"""
requirements = requirements or {}
unmet = set()
if (
requirements.get("local_materials")
and video_source == "local"
and not has_local_materials
):
unmet.add("local_materials")
if requirements.get("custom_audio") and not has_custom_audio:
if voice_mode is not None:
# 新版 WebUI 使用显式配音方式。用户切换到自动配音或无配音,表示
# 已主动替换历史上传音频;只有继续选择上传模式时才要求重新上传。
if voice_mode == VOICE_MODE_UPLOAD:
unmet.add("custom_audio")
elif voice_name == requirements.get("original_voice_name", ""):
# 保留旧调用方按音色判断的兼容行为,避免影响 API 和已有测试工具。
unmet.add("custom_audio")
return unmet
def _queue_task_restore(task_id):
# 任务列表运行在 fragment 中,不能直接修改已经创建的主表单控件状态。
# 这里只记录候选任务并触发整页 rerun,确认和参数恢复由主页面统一处理。
st.session_state["task_restore_candidate_id"] = task_id
st.session_state["task_manager_popover_nonce"] = (
st.session_state.get("task_manager_popover_nonce", 0) + 1
)
st.rerun(scope="app")
def _normalize_task_state(state):
if state in (
const.TASK_STATE_COMPLETE,
const.TASK_STATE_FAILED,
const.TASK_STATE_PROCESSING,
):
return state
try:
return int(state)
except (TypeError, ValueError):
return state
def _active_generation_tasks():
tasks = st.session_state.setdefault("active_generation_tasks", {})
if not isinstance(tasks, dict):
tasks = {}
st.session_state["active_generation_tasks"] = tasks
return tasks
def _add_active_generation_task(task_id, subject=None):
tasks = _active_generation_tasks()
task = tasks.setdefault(task_id, {})
task["subject"] = subject or task.get("subject") or task_id
task["mtime"] = task.get("mtime") or datetime.now().timestamp()
def _remove_active_generation_task(task_id):
tasks = _active_generation_tasks()
if task_id in tasks:
del tasks[task_id]
if st.session_state.get("pending_generation_task_id") == task_id:
del st.session_state["pending_generation_task_id"]
def _prepare_generation_task():
# st.button 的 on_click 会在页面脚本重新执行前触发。这里提前生成任务 ID,
# 顶部任务管理入口就能在同一次 rerun 中显示“生成中”数量。
current_id = _current_generation_task_id()
if current_id and webui_task.is_generation_active(current_id):
st.session_state["pending_generation_task_id"] = current_id
return
task_id = str(uuid4())
st.session_state["pending_generation_task_id"] = task_id
subject = st.session_state.get("video_subject") or st.session_state.get(
"video_script"
)
_add_active_generation_task(task_id, subject=subject)
def _task_state_label(state, has_video):
normalized_state = _normalize_task_state(state)
if normalized_state == const.TASK_STATE_COMPLETE:
return tr("Task Status Complete")
if normalized_state == const.TASK_STATE_FAILED:
return tr("Task Status Failed")
if normalized_state == const.TASK_STATE_PROCESSING:
return tr("Task Status Processing")
if has_video:
return tr("Task Status Complete")
return tr("Task Status History")
def _task_state_filter_key(task):
normalized_state = _normalize_task_state(task.get("state"))
if normalized_state == const.TASK_STATE_PROCESSING:
return "processing"
if normalized_state == const.TASK_STATE_FAILED:
return "failed"
if normalized_state == const.TASK_STATE_COMPLETE or task["video_file"]:
return "complete"
return "history"
def _scan_history_tasks(limit=30):
tasks_root = utils.task_dir()
if not os.path.isdir(tasks_root):
return []
# 任务管理 fragment 每两秒刷新一次。先只读取低成本的目录元数据并截取最近
# 的任务,再解析 script.json 和视频列表,避免历史任务很多时反复扫描全部内容。
task_entries = []
try:
with os.scandir(tasks_root) as entries:
for entry in entries:
try:
if entry.name.startswith(".") or not entry.is_dir(
follow_symlinks=False
):
continue
task_entries.append(
(
entry.stat(follow_symlinks=False).st_mtime,
entry.name,
entry.path,
)
)
except OSError as e:
# 单个任务目录可能正在被删除,不应因此让整个任务面板失效。
logger.debug(f"skip unavailable task directory: {entry.path}, {e}")
except OSError as e:
logger.warning(f"failed to scan task directory: {tasks_root}, {e}")
return []
task_entries.sort(key=lambda item: item[0], reverse=True)
tasks = []
for mtime, name, task_path in task_entries[:limit]:
script_data = _safe_load_task_script(task_path)
params_data = script_data.get("params", {})
if not isinstance(params_data, dict):
params_data = {}
script_text = script_data.get("script", "")
if not isinstance(script_text, str):
script_text = ""
video_file = _find_final_task_video(task_path)
subject = (
params_data.get("video_subject")
or script_text[:40]
or name
)
tasks.append(
{
"task_id": name,
"subject": subject,
"state": const.TASK_STATE_COMPLETE if video_file else None,
"progress": 100 if video_file else 0,
"mtime": mtime,
"task_path": task_path,
"video_file": video_file,
"source": "history",
}
)
return tasks
def _collect_task_summaries(limit=20):
history_tasks = {task["task_id"]: task for task in _scan_history_tasks(limit=50)}
active_tasks = _active_generation_tasks()
try:
runtime_tasks, _ = sm.state.get_all_tasks(1, 50)
except Exception as e:
logger.warning(f"failed to load runtime tasks: {e}")
runtime_tasks = []
# The paginated state view can omit this session's newer tasks after 50
# older records. Read those active IDs directly so a completed or failed
# task cannot remain labelled as processing forever.
runtime_ids = {task.get("task_id") for task in runtime_tasks}
for task_id in active_tasks:
if task_id in runtime_ids:
continue
try:
task = sm.state.get_task(task_id)
except Exception as e:
logger.warning(f"failed to load active task {task_id}: {e}")
continue
if task:
runtime_tasks.append(task)
for task in runtime_tasks:
task_id = task.get("task_id", "")
if not task_id:
continue
task_path = os.path.join(utils.task_dir(), task_id)
history_task = history_tasks.get(task_id, {})
video_files = task.get("videos") or []
video_file = (
video_files[0] if video_files else history_task.get("video_file", "")
)
subject = (
task.get("video_subject")
or history_task.get("subject")
or (task.get("script", "")[:40] if task.get("script") else "")
or task_id
)
task_mtime = active_tasks.get(task_id, {}).get("mtime") or history_task.get(
"mtime", 0
)
if os.path.isdir(task_path):
try:
task_mtime = os.path.getmtime(task_path)
except OSError:
# Another session can delete this directory between isdir and
# getmtime. Keep rendering the persisted task state.
pass
history_tasks[task_id] = {
"task_id": task_id,
"subject": subject,
"state": task.get("state"),
"cross_post_state": task.get("cross_post_state"),
"progress": int(task.get("progress", 0) or 0),
"mtime": task_mtime,
"task_path": task_path,
"video_file": video_file,
"source": "runtime",
}
for task_id, active_task in active_tasks.items():
history_task = history_tasks.get(task_id, {})
if history_task and _task_state_filter_key(history_task) in {
"complete",
"failed",
}:
# 会话中的 active 标记只负责覆盖任务刚提交到状态存储前的极短窗口。
# 后台任务结束后必须以真实终态为准,不能把失败任务重新显示为生成中。
continue
task_path = os.path.join(utils.task_dir(), task_id)
history_tasks[task_id] = {
"task_id": task_id,
"subject": active_task.get("subject")
or history_task.get("subject")
or task_id,
"state": const.TASK_STATE_PROCESSING,
"progress": history_task.get("progress", 0),
"mtime": active_task.get("mtime")
or history_task.get("mtime", datetime.now().timestamp()),
"task_path": task_path,
"video_file": history_task.get("video_file", ""),
"source": "active",
}
tasks = list(history_tasks.values())
return sorted(tasks, key=lambda item: item["mtime"], reverse=True)[:limit]
def _is_headless_server():
# Docker 或无桌面的服务器部署中,WebUI 进程接触不到用户的桌面环境:
# xdg-open / webbrowser 只会在容器内静默失败。此时应改为浏览器内预览
# 视频、以路径提示代替打开目录。macOS/Windows 桌面部署不受影响。
if sys.platform == "darwin" or sys.platform.startswith("win"):
return False
return not (os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY"))
def _open_task_path(task_path):
tasks_root = os.path.abspath(utils.task_dir())
normalized_path = os.path.abspath(task_path)
if not normalized_path.startswith(tasks_root + os.sep):
logger.warning(f"invalid task folder path: {normalized_path}")
return
if not os.path.isdir(normalized_path):
return
if _is_headless_server():
# storage 目录通常以卷挂载映射回宿主机,提示相对路径即可定位文件。
rel_path = os.path.relpath(normalized_path, os.path.dirname(tasks_root))
st.toast(f"{tr('Open Task Folder')}: ./storage/{rel_path}", icon="📂")
return
webbrowser.open(f"file://{normalized_path}")
def _open_task_video(video_file):
tasks_root = os.path.abspath(utils.task_dir())
normalized_file = os.path.abspath(video_file)
# 视频路径来自任务目录扫描或运行期状态。这里仍然限制只能打开任务目录
# 内的文件,避免 UI 操作被异常路径扩展成任意本地文件打开能力。
if not normalized_file.startswith(tasks_root + os.sep):
logger.warning(f"invalid task video path: {normalized_file}")
return
if not os.path.isfile(normalized_file):
logger.warning(f"task video does not exist: {normalized_file}")
return
if _is_headless_server():
# 无桌面环境时在任务面板内嵌播放器预览,代替调用系统播放器。
st.session_state["task_preview_video_file"] = normalized_file
return
try:
if sys.platform == "darwin":
subprocess.Popen(["open", normalized_file])
elif sys.platform.startswith("win"):
os.startfile(normalized_file) # type: ignore[attr-defined]
else:
subprocess.Popen(["xdg-open", normalized_file])
except Exception as e:
logger.error(f"failed to open task video: {normalized_file}, {e}")
def _delete_task(task_id, task_path, task_state=None):
# 页面展示的状态可能落后于后台任务。删除前同时检查传入状态、当前会话的
# 活跃任务和最新状态,避免任务刚开始或已产出中间视频时被误删。
current_task = None
try:
current_task = sm.state.get_task(task_id)
except Exception as e:
logger.exception(f"failed to verify task state before deletion: {task_id}, {e}")
return False
task_snapshot = dict(current_task or {})
task_snapshot.setdefault("state", task_state)
if task_id in _active_generation_tasks():
task_snapshot["state"] = const.TASK_STATE_PROCESSING
if tm.is_task_busy(task_snapshot):
logger.warning(f"refused to delete running task: {task_id}")
return False
tasks_root = os.path.abspath(utils.task_dir())
normalized_path = os.path.abspath(task_path)
# 删除任务会移除任务状态和本地生成文件。这里必须限定在 storage/tasks
# 下,避免异常 task_path 造成误删其它本地目录。
if not normalized_path.startswith(tasks_root + os.sep):
logger.warning(f"invalid task folder path for deletion: {normalized_path}")
return False
try:
if os.path.isdir(normalized_path):
shutil.rmtree(normalized_path)
# Keep the persisted task available for retry when file removal fails
# (for example, a video is still locked by another process on Windows).
if hasattr(sm.state, "delete_task"):
sm.state.delete_task(task_id)
logger.info(f"deleted task: {task_id}")
return True
except Exception as e:
logger.exception(f"failed to delete task: {task_id}, {e}")
return False
def _count_processing_tasks(tasks):
# 顶部任务管理入口只需要展示“生成中”任务数量。
# 这里复用内部状态 key 判断,避免依赖多语言展示文案导致不同语言下统计不一致。
processing_task_ids = {
task["task_id"]
for task in tasks
if _task_state_filter_key(task) == "processing"
}
return len(processing_task_ids)
def _task_manager_label(processing_count):
label = tr("Task Manager")
if processing_count <= 0:
return label
return f"{label} · {processing_count}"
def _build_video_download_name(subject, index, total):
"""根据视频主题生成跨平台安全的下载文件名。"""
safe_subject = _DOWNLOAD_FILENAME_INVALID_PATTERN.sub(" ", str(subject or ""))
safe_subject = re.sub(r"\s+", " ", safe_subject).strip(" .")[:80].rstrip(" .")
if not safe_subject:
safe_subject = "video"
# Win32 在识别设备名时会忽略扩展名前的尾随空格和句点。与背景音乐上传
# 的现有规则保持一致,避免 ``CON .topic`` 绕过保留名保护。
windows_basename = safe_subject.split(".", 1)[0].rstrip(" .").upper()
if windows_basename in _WINDOWS_RESERVED_FILENAMES:
safe_subject = f"_{safe_subject}"
suffix = f"-{index}" if total > 1 else ""
return f"{safe_subject}{suffix}.mp4"
def _render_task_table(filtered_tasks, key_prefix):
with st.container(key=f"task_table_header_{key_prefix}"):
header_cols = st.columns([1.1, 1.7, 3.0, 0.8, 1.6], vertical_alignment="center")
header_cols[0].caption(tr("Task Status"))
header_cols[1].caption(tr("Task Updated At"))
header_cols[2].caption(tr("Task Subject"))
header_cols[3].caption(tr("Task Progress"))
header_cols[4].caption(tr("Task Actions"))
if not filtered_tasks:
st.info(tr("No Tasks Match Filter"))
return
visible_tasks = filtered_tasks[:12]
list_height = min(390, max(96, len(visible_tasks) * 58))
with st.container(height=list_height, border=False):
for task in visible_tasks:
task_id = task["task_id"]
has_video = bool(task["video_file"] and os.path.isfile(task["video_file"]))
is_processing = _task_state_filter_key(task) == "processing"
is_busy = is_processing or tm.is_task_busy(task)
has_restore_data = os.path.isfile(
os.path.join(task["task_path"], "script.json")
)
safe_task_key = "".join(ch if ch.isalnum() else "_" for ch in task_id)[:40]
# 使用 Streamlit 原生 bordered container + columns 保留每行操作。
# 相比自定义 HTML/CSS 表格,这种方式对 Streamlit 版本变更更稳;
# 相比 dataframe,又能保留播放、打开目录、删除等行内动作。
with st.container(
key=f"task_row_{key_prefix}_{safe_task_key}", border=True
):
row_cols = st.columns(
[1.1, 1.7, 3.0, 0.8, 1.6],
vertical_alignment="center",
)
row_cols[0].write(_task_state_label(task["state"], has_video))
row_cols[1].write(_format_task_time(task["mtime"]))
row_cols[2].write(_format_task_subject(task["subject"]))
row_cols[3].write(f"{task['progress']}%")
action_cols = row_cols[4].columns(
4,
vertical_alignment="center",
gap="small",
)
with action_cols[0]:
play_label = tr("Play")
if st.button(
play_label,
key=f"play_task_{key_prefix}_{task_id}",
use_container_width=True,
icon=":material/play_arrow:",
help=play_label,
disabled=not has_video,
):
_open_task_video(task["video_file"])
with action_cols[1]:
open_label = tr("Open Task Folder")
if st.button(
open_label,
key=f"open_task_{key_prefix}_{task_id}",
use_container_width=True,
icon=":material/folder_open:",
help=open_label,
):
_open_task_path(task["task_path"])
with action_cols[2]:
restore_label = tr("Regenerate Task")
if st.button(
restore_label,
key=f"restore_task_{key_prefix}_{task_id}",
use_container_width=True,
icon=":material/replay:",
help=restore_label,
disabled=is_processing or not has_restore_data,
):
_queue_task_restore(task_id)
with action_cols[3]:
delete_label = tr("Delete Task")
delete_help = (
f"{delete_label} ({tr('Task Status Processing')})"
if is_busy
else delete_label
)
if st.button(
delete_label,
key=f"delete_task_{key_prefix}_{task_id}",
use_container_width=True,
icon=":material/delete:",
help=delete_help,
disabled=is_busy,
):
if _delete_task(task_id, task["task_path"], task["state"]):
st.toast(tr("Task Deleted"))
st.rerun()
else:
st.error(tr("Task Delete Failed"))
def _render_task_manager_panel(tasks=None):
tasks = tasks if tasks is not None else _collect_task_summaries()
if not tasks:
st.info(tr("No Tasks Yet"))
return
# Streamlit 1.59 支持有状态 Tabs 的惰性渲染。切换时只重新构建当前列表,
# 避免定时 Fragment 每两秒重复创建四套任务行和操作按钮。
status_tabs = [
("all", tr("All Tasks")),
("processing", tr("Task Status Processing")),
("complete", tr("Task Status Complete")),
("failed", tr("Task Status Failed")),
]
tabs = st.tabs(
[label for _, label in status_tabs],
key="task_manager_status_tabs",
on_change="rerun",
)
for (status_key, _), tab in zip(status_tabs, tabs):
if not tab.open:
continue
with tab:
filtered_tasks = [
task
for task in tasks
if status_key == "all" or _task_state_filter_key(task) == status_key
]
_render_task_table(filtered_tasks, status_key)
_render_task_video_preview()
def _render_task_video_preview():
# 无桌面部署下“播放”按钮的浏览器内回退:在任务面板底部渲染播放器。
preview_file = st.session_state.get("task_preview_video_file")
if not preview_file:
return
tasks_root = os.path.abspath(utils.task_dir())
if not (
preview_file.startswith(tasks_root + os.sep) and os.path.isfile(preview_file)
):
st.session_state.pop("task_preview_video_file", None)
return
st.divider()
preview_cols = st.columns([5, 1], vertical_alignment="center")
task_name = os.path.basename(os.path.dirname(preview_file))
preview_cols[0].caption(f"{os.path.basename(preview_file)} · {task_name}")
closed = preview_cols[1].button(
"✕",
key="close_task_video_preview",
use_container_width=True,
help=tr("Close"),
)
if closed:
st.session_state.pop("task_preview_video_file", None)
return
st.video(preview_file)
@st.fragment(run_every="2s")
def _render_task_manager_entry():
# 任务可能由当前页面或其它页面触发生成。入口单独用 fragment 定时刷新,
# 只更新任务数量和 popover 内容,不打断主页面表单输入。
task_summaries = _collect_task_summaries()
processing_task_count = _count_processing_tasks(task_summaries)
with st.container(key="task_manager_entry", width="content"):
with st.popover(
_task_manager_label(processing_task_count),
width="content",
key=(
"task_manager_popover_"
f"{st.session_state.get('task_manager_popover_nonce', 0)}"
),
):
_render_task_manager_panel(task_summaries)
def _load_task_restore_payload(task_id):
tasks_root = os.path.realpath(utils.task_dir())
task_path = os.path.realpath(os.path.join(tasks_root, str(task_id)))
try:
if os.path.commonpath([tasks_root, task_path]) != tasks_root:
raise ValueError("task path is outside the task directory")
except ValueError as e:
logger.warning(f"invalid task restore path: {task_id}, {e}")
return None
script_data = _safe_load_task_script(task_path)
raw_params = script_data.get("params")
if not isinstance(raw_params, dict):
logger.warning(f"task has no restorable parameters: {task_id}")
return None
params_input = dict(raw_params)
if script_data.get("script"):
params_input["video_script"] = script_data["script"]
if script_data.get("search_terms"):
params_input["video_terms"] = script_data["search_terms"]
try:
params = VideoParams.model_validate(params_input).model_dump(mode="json")
except Exception as e:
logger.warning(f"failed to validate task restore parameters: {task_id}, {e}")
return None
return {
"task_id": str(task_id),
"subject": params.get("video_subject") or script_data.get("script") or task_id,
"params": params,
}
def _infer_tts_server_from_voice(voice_name):
if voice.is_no_voice(voice_name):
return voice.NO_VOICE_NAME
if voice.is_siliconflow_voice(voice_name):
return "siliconflow"
if voice.is_gemini_voice(voice_name):
return "gemini-tts"
if voice.is_mimo_voice(voice_name):
return "mimo-tts"
if voice.is_minimax_voice(voice_name):
return "minimax-tts"
if voice.is_elevenlabs_voice(voice_name):
return "elevenlabs"
if voice.is_chatterbox_voice(voice_name):
return "chatterbox"
if voice.is_kokoro_voice(voice_name):
return "kokoro"
if voice.is_fish_audio_voice(voice_name):
return "fish_audio"
if voice.is_voxcpm_voice(voice_name):
return "voxcpm"
if voice.is_azure_v2_voice(voice_name):
return "azure-tts-v2"
return "azure-tts-v1"
def _set_stable_widget_value(key, value):
if value is not None:
st.session_state[localized_widget_key(key)] = value
def _apply_pending_task_restore():
payload = st.session_state.pop("task_restore_payload", None)
if not payload:
return False
_apply_restored_params(payload["params"])
st.session_state["task_restore_succeeded"] = True
logger.info(f"restored task configuration: {payload['task_id']}")
return True
def _apply_restored_params(params):
"""
把一份完整的生成参数写回页面控件状态。
历史任务恢复和设置预设导入使用同一份参数模型,因此共用同一个实现,避免
新增字段时只更新其中一条路径。调用方必须在渲染任何控件之前执行,否则
Streamlit 会拒绝修改已经实例化的控件状态。
"""
video_terms = params.get("video_terms") or ""
if isinstance(video_terms, list):
video_terms = ", ".join(str(term) for term in video_terms)
# 文案与高级脚本设置。
st.session_state["video_subject"] = params.get("video_subject") or ""
st.session_state["video_script"] = params.get("video_script") or ""
st.session_state["video_terms"] = str(video_terms)
_set_stable_widget_value(
"script_language_select", params.get("video_language") or ""
)
st.session_state["paragraph_number_input"] = params.get("paragraph_number", 1)
st.session_state["video_script_prompt"] = params.get("video_script_prompt") or ""
st.session_state["custom_system_prompt"] = (
params.get("custom_system_prompt") or llm.DEFAULT_SCRIPT_SYSTEM_PROMPT
)
# 视频设置。素材上传控件不能由服务端写入,因此本地素材需要用户重新选择。
video_source = params.get("video_source") or "pexels"
_set_stable_widget_value("video_source_select", video_source)
_set_stable_widget_value(
"video_concat_mode_select", params.get("video_concat_mode") or "random"
)
_set_stable_widget_value(
"video_transition_mode_select",
params.get("video_transition_mode") or VideoTransitionMode.none.value,
)
_set_stable_widget_value(
f"video_aspect_for_{video_source}",
params.get("video_aspect") or VideoAspect.portrait.value,
)
_set_stable_widget_value(
"video_fit_mode_select",
params.get("video_fit_mode") or VideoFitMode.cover.value,
)
_set_stable_widget_value(
"image_motion_select", params.get("image_motion") or ImageMotion.zoom.value
)
_set_stable_widget_value(
"video_clip_duration_select", params.get("video_clip_duration", 3)
)
_set_stable_widget_value(
"video_clip_speed_slider",
# API 可以写入超过 WebUI 范围的速度,任务生成阶段会安全归一化,但
# 历史记录仍可能保留原值。恢复任务前再次归一化,避免给 Streamlit
# slider 注入越界值、NaN 或无穷值导致控件状态异常。
utils.normalize_clip_speed(params.get("video_clip_speed", 1.0)),
)
_set_stable_widget_value("video_count_select", params.get("video_count", 1))
st.session_state["match_materials_to_script"] = bool(
params.get("match_materials_to_script", False)
)
# 音频设置。TTS server 未写入旧任务,根据历史 voice_name 推断。
voice_name = params.get("voice_name") or voice.NO_VOICE_NAME
tts_server = _infer_tts_server_from_voice(voice_name)
if params.get("custom_audio_file"):
voice_mode = VOICE_MODE_UPLOAD
elif voice.is_no_voice(voice_name):
voice_mode = VOICE_MODE_NONE
else:
voice_mode = VOICE_MODE_TTS
_set_stable_widget_value("voice_mode_control", voice_mode)
if tts_server != voice.NO_VOICE_NAME:
_set_stable_widget_value("tts_server_select", tts_server)
_set_stable_widget_value(f"speech_synthesis_select_{tts_server}", voice_name)
_set_stable_widget_value("voice_volume_select", params.get("voice_volume", 1.0))
_set_stable_widget_value("voice_rate_select", params.get("voice_rate", 1.0))
bgm_type = params.get("bgm_type") or ""
_set_stable_widget_value("bgm_type_select", bgm_type)
_set_stable_widget_value("bgm_volume_select", params.get("bgm_volume", 0.2))
if bgm_type == "preset" and params.get("bgm_file"):
# 预设歌曲控件使用文件名作为稳定业务值。历史任务可能保存绝对路径或
# 相对路径,统一取 basename 后即可匹配当前安全枚举出的歌曲列表。
_set_stable_widget_value(
"preset_song_select", os.path.basename(str(params["bgm_file"]))
)
st.session_state["custom_bgm_file_input"] = params.get("bgm_file") or ""
st.session_state["sonilo_bgm_prompt_input"] = (
params.get("video_music_prompt") or params.get("sonilo_bgm_prompt") or ""
)
st.session_state["elevenlabs_music_prompt_input"] = (
params.get("video_music_prompt") or ""
)
# 字幕设置。对旧任务中的越界数值做最小限幅,避免 Slider 无法初始化。
st.session_state["subtitle_enabled_checkbox"] = bool(
params.get("subtitle_enabled", True)
)
_set_stable_widget_value("font_name_select", params.get("font_name") or "")
_set_stable_widget_value(
"subtitle_position_select", params.get("subtitle_position") or "bottom"
)
_set_stable_widget_value(
"subtitle_display_mode_select", params.get("subtitle_display_mode") or "sentence"
)
_set_stable_widget_value(
"subtitle_animation_select", params.get("subtitle_animation") or "none"
)
custom_position = min(100.0, max(0.0, float(params.get("custom_position", 70.0))))
st.session_state["custom_position_input"] = str(custom_position)
st.session_state["font_color_picker"] = params.get("text_fore_color") or "#FFFFFF"
st.session_state["font_size_slider"] = min(
100, max(30, int(params.get("font_size", 60)))
)
st.session_state["stroke_color_picker"] = params.get("stroke_color") or "#000000"
st.session_state["stroke_width_slider"] = min(
10.0, max(0.0, float(params.get("stroke_width", 1.5)))
)
background_color = params.get("text_background_color")
background_enabled = bool(background_color)
st.session_state["subtitle_background_enabled_checkbox"] = background_enabled
if isinstance(background_color, str):
st.session_state["subtitle_background_color_picker"] = background_color
st.session_state["rounded_subtitle_background_checkbox"] = bool(
params.get("rounded_subtitle_background", False) and background_enabled
)
st.session_state.pop("local_video_materials_uploader", None)
# 历史任务只保存素材路径,不能保证这些文件在当前环境仍然存在。
# 同时清空当前页面已缓存的上传素材,避免恢复后误用另一个任务的文件。
st.session_state["local_video_materials"] = []
st.session_state.pop("custom_audio_file_uploader", None)
st.session_state.pop("voxcpm_reference_audio_uploader", None)
st.session_state.pop(VOXCPM_REFERENCE_AUDIO_SESSION_KEY, None)
st.session_state.pop(VOXCPM_REFERENCE_AUDIO_ERROR_SESSION_KEY, None)
st.session_state.pop("voxcpm_prompt_audio_uploader", None)
st.session_state.pop(VOXCPM_PROMPT_AUDIO_SESSION_KEY, None)
st.session_state.pop(VOXCPM_PROMPT_AUDIO_ERROR_SESSION_KEY, None)
st.session_state.pop(VOXCPM_PROMPT_TEXT_SESSION_KEY, None)
st.session_state.pop(VOXCPM_HIGH_FIDELITY_SESSION_KEY, None)
st.session_state.pop(VOXCPM_SEPARATE_PROMPT_AUDIO_SESSION_KEY, None)
st.session_state.pop(VOXCPM_PROMPT_EXAMPLE_MODE_SESSION_KEY, None)
st.session_state.pop("custom_bgm_uploader", None)
st.session_state.pop("custom_bgm_validation", None)
st.session_state["task_restore_upload_requirements"] = (
_build_restore_upload_requirements(params)
)
return True
def _dismiss_task_restore_dialog():
st.session_state.pop("task_restore_candidate_id", None)
@st.dialog(
tr("Regenerate Task"),
width="small",
on_dismiss=_dismiss_task_restore_dialog,
)
def _render_task_restore_dialog(task_id):
payload = _load_task_restore_payload(task_id)
if payload is None:
st.error(tr("Task Restore Failed"))
if st.button(tr("Cancel"), key="cancel_invalid_task_restore"):
st.session_state.pop("task_restore_candidate_id", None)
st.rerun(scope="app")
return
st.write(tr("Regenerate Task Confirmation"))
st.caption(_format_task_subject(payload["subject"], max_length=80))
cancel_col, load_col = st.columns(2)
if cancel_col.button(
tr("Cancel"),
key="cancel_task_restore",
use_container_width=True,
):
st.session_state.pop("task_restore_candidate_id", None)
st.rerun(scope="app")
if load_col.button(
tr("Load Task Configuration"),
key="confirm_task_restore",
type="primary",
use_container_width=True,
):
st.session_state["task_restore_payload"] = payload
st.session_state.pop("task_restore_candidate_id", None)
st.rerun(scope="app")
def _dismiss_settings_dialog():
"""关闭设置弹窗,并确保下一次整页 rerun 不会再次自动打开。"""
st.session_state["settings_dialog_open"] = False
def _open_settings_dialog(target_tab=None):
"""打开设置弹窗,并可直接定位到指定业务标签页。"""
st.session_state["settings_dialog_open"] = True
if not target_tab:
target_tab = "general"
if target_tab:
# 这里只保存稳定的业务 ID,不保存翻译文本;真正创建 tabs 前再根据
# 当前界面语言解析 label,避免用户切换语言后旧文案成为非法选项。
st.session_state["settings_dialog_target_tab"] = target_tab
def _open_material_settings_dialog():
"""供视频来源组件回调使用:直接打开素材服务设置。"""
_open_settings_dialog("material")
def _render_brand(available_update: str | None = None):
"""渲染项目名称、当前版本和可选的更新入口。"""
update_link = ""
if available_update:
update_label = html.escape(
tr("Update Available").format(version=available_update)
)
# Streamlit 会继续用 Markdown 解析传入的 HTML。这里保持链接为单行,
# 避免多行字符串的缩进被识别成代码块,导致页面直接显示 HTML 源码。
update_link = (
''
f"{update_label}"
)
st.markdown(
f"""
""",
unsafe_allow_html=True,
)
@st.fragment(run_every="1s")
def _render_pending_version_check():
"""检查未完成时只刷新品牌区域,避免阻塞或反复执行整页表单。"""
snapshot = version_checker.poll_available_update(config.project_version)
if snapshot.complete:
# 检查完成后刷新一次整页,让顶部栏改为静态渲染并停止 fragment 轮询。
# 该刷新发生在后台请求完成之后,不会延迟初始页面的其它内容。
st.rerun(scope="app")
_render_brand()
def _render_top_bar():
"""渲染品牌、任务管理、设置和语言切换组成的页面顶部栏。"""
# 顶部栏分为品牌区和操作区两个独立区域。窄屏下由 Streamlit
# 将两个区域整体换行,操作区内部再根据剩余宽度自动换行。
with st.container(key="top_bar"):
brand_col, actions_col = st.columns(
[3.5, 2.0],
vertical_alignment="center",
gap="small",
)
with brand_col:
update_snapshot = version_checker.poll_available_update(config.project_version)
if update_snapshot.complete:
_render_brand(update_snapshot.available_version)
else:
_render_pending_version_check()
with actions_col:
with st.container(
key="top_bar_actions",
horizontal=True,
horizontal_alignment="right",
vertical_alignment="center",
gap="small",
width="stretch",
):
_render_task_manager_entry()
st.button(
tr("Settings"),
key="open_settings_dialog_button",
type="secondary",
icon=":material/settings:",
width="content",
on_click=_open_settings_dialog,
)
language_codes = list(locales.keys())
selected_index = 0
for i, code in enumerate(language_codes):
if code == st.session_state.get("ui_language", ""):
selected_index = i
selected_language_code = st.selectbox(
"Language / 语言",
options=language_codes,
index=selected_index,
format_func=lambda code: locales[code].get("Language", code),
key="top_language_code_selector",
label_visibility="collapsed",
width=180,
)
if selected_language_code:
previous_language = st.session_state.get("ui_language", "")
if selected_language_code != previous_language:
logger.info(
"UI language changed by user: "
f"previous_language={previous_language or ''}, "
f"selected_language={selected_language_code}"
)
st.session_state["ui_language"] = selected_language_code
# 浏览器自动识别只影响当前会话;只有用户主动切换下拉框时才
# 写入 config.toml,后续新会话将优先使用该明确选择。
_set_runtime_config("ui", "language", selected_language_code)
_save_runtime_config()
# 切换语言会先重跑顶部栏,正文控件尚未渲染就触发 rerun。
# 显式保留本次创作内容,防止 Streamlit 清理旧控件状态后,
# 新语言页面把已经输入的主题、文案和关键词重置为空。
for content_key in ("video_subject", "video_script", "video_terms"):
if content_key in st.session_state:
st.session_state[content_key] = st.session_state[content_key]
# 切换语言后强制刷新,避免 selectbox 继续展示旧语言文案。
st.rerun()
support_locales = [
"ca-ES",
"zh-CN",
"zh-HK",
"zh-TW",
"de-DE",
"en-US",
"es-ES",
"fr-FR",
"id-ID",
"it-IT",
"ru-RU",
"vi-VN",
"th-TH",
"tr-TR",
]
# -----------------------------------------------------------------------------
# 通用 UI 组件、资源缓存与日志
# -----------------------------------------------------------------------------
@st.cache_data(ttl=30, show_spinner=False)
def get_all_fonts():
# 字体目录很少变化,但 Streamlit 每次控件交互都会 rerun 页面。短周期缓存
# 可以避免连续重复 os.walk,同时保证新增字体后最多 30 秒即可被发现。
fonts = []
for root, dirs, files in os.walk(font_dir):
for file in files:
if file.endswith(".ttf") or file.endswith(".ttc"):
# Keep each option tied to its file, including nested fonts.
relative_path = os.path.relpath(os.path.join(root, file), font_dir)
fonts.append(relative_path.replace(os.sep, "/"))
fonts.sort()
return fonts
@st.cache_data(ttl=30, show_spinner=False)
def get_all_songs():
# 背景音乐与字体使用相同的短周期策略,不做永久缓存,兼顾 rerun 性能和
# 用户运行期间手动添加音乐文件的场景。
songs = []
for root, dirs, files in os.walk(song_dir):
for file in files:
if file.endswith(".mp3"):
songs.append(file)
return songs
def open_task_folder(task_id):
try:
# task_id 应始终是服务端生成的 UUID。这里先做格式校验,避免异常值
# 通过路径拼接访问任务目录之外的位置,也避免后续打开目录时触发
# 平台 shell 对特殊字符的解释。
normalized_task_id = str(UUID(str(task_id)))
tasks_root = os.path.abspath(utils.task_dir())
path = os.path.abspath(os.path.join(tasks_root, normalized_task_id))
# 即使 UUID 校验通过,也再次确认最终路径仍在任务根目录内,避免
# 未来调用方调整 task_id 来源时引入路径穿越风险。
if not path.startswith(tasks_root + os.sep):
logger.warning(f"invalid task folder path: {path}")
return
if os.path.isdir(path):
webbrowser.open(f"file://{path}")
except Exception as e:
logger.exception(f"failed to open task folder: task_id={task_id}, error={e}")
@st.cache_resource
def init_log():
# 基础日志 Handler 属于进程级资源,而不是页面会话状态。Streamlit 每次组件
# 交互都会 rerun 页面脚本,代码热重载也可能让缓存失效。日志初始化只能
# 精确替换终端 Handler,不能清空正在生成任务使用的 WebUI 临时 Handler。
_lvl = get_logging_options()[0]
return configure_terminal_logger(
sys.stdout,
level=_lvl,
colorize=True,
)
init_log()
def tr_optional(key, fallback_language=""):
loc = locales.get(st.session_state["ui_language"], {})
value = loc.get("Translation", {}).get(key, "")
if not value and fallback_language:
fallback_loc = locales.get(fallback_language, {})
value = fallback_loc.get("Translation", {}).get(key, "")
return value if value else ""
def render_onboarding_tour():
# 引导只覆盖三个稳定入口,不尝试控制 Dialog、Tabs 或业务表单。这样既能让
# 新用户理解完整流程,也不会把引导状态与 Streamlit 的动态组件生命周期耦合。
steps = [
Tour.bind(
"open_settings_dialog_button",
title=tr("Onboarding Model Settings Title"),
desc=tr("Onboarding Model Settings Description"),
side="bottom",
align="end",
),
Tour.bind(
"main_settings_grid",
title=tr("Onboarding Creation Settings Title"),
desc=tr("Onboarding Creation Settings Description"),
side="top",
align="center",
),
Tour.bind(
"generate_video_button",
title=tr("Onboarding Generate Video Title"),
desc=tr("Onboarding Generate Video Description"),
side="top",
align="center",
),
]
# streamlit-tour 1.1.0 没有在 Python 构造参数中暴露导航文案,但底层
# Driver.js 支持在每一步的 popover 配置中覆盖按钮文本。这里统一注入本地化
# 文案,并对内容做 HTML 转义,因为组件会通过 innerHTML 渲染这些字段。
previous_text = html.escape(tr("Onboarding Previous"))
next_text = html.escape(tr("Onboarding Next"))
done_text = html.escape(tr("Onboarding Done"))
for index, step in enumerate(steps):
step.popover["prevBtnText"] = f"← {previous_text}"
# Driver.js 会在合并单步配置时覆盖已经替换过变量的进度模板,因此直接
# 写入当前步骤和总步骤数,避免页面显示未解析的 {{current}} 占位符。
step.popover["progressText"] = f"{index + 1} / {len(steps)}"
if index == len(steps) - 1:
step.popover["doneBtnText"] = done_text
else:
step.popover["nextBtnText"] = f"{next_text} →"
tour = Tour(
steps=steps,
key=ONBOARDING_TOUR_KEY,
show_progress=True,
animate=True,
overlay_opacity=0.55,
one_time_tour=True,
)
# 每个 Streamlit 会话只主动启动一次。是否已经完成则由组件通过浏览器
# localStorage 判断,避免页面 rerun 或普通控件交互反复弹出引导。
auto_start_key = f"{ONBOARDING_TOUR_KEY}-auto-started"
if not st.session_state.get(auto_start_key, False):
st.session_state[auto_start_key] = True
tour.start()
def _render_generation_logs(task_id):
"""渲染后台任务日志快照,不从工作线程访问 Streamlit 会话状态。"""
if config.ui.get("hide_log", False):
return
log_records = webui_task.get_task_logs(task_id)
if not log_records:
return
st.code("\n".join(log_records))
def _current_generation_task_id():
"""URL 只保存本地任务 ID,让刷新/重连恢复原任务,不保存配置或凭据。"""
task_id = st.session_state.get("current_generation_task_id", "")
if not task_id:
try:
task_id = str(UUID(st.query_params.get("generation_task", "")))
except (ValueError, TypeError, AttributeError):
return ""
st.session_state["current_generation_task_id"] = task_id
return task_id
def _generation_is_busy(task_id):
if not task_id:
return False
if webui_task.is_generation_active(task_id):
return True
try:
task = sm.state.get_task(task_id)
except Exception:
# 查询异常时不能把“状态未知”当作已完成而放开重复提交。
return True
return _normalize_task_state((task or {}).get("state")) == const.TASK_STATE_PROCESSING
def _render_generation_activity(task):
activity = task.get("generation_activity") or {}
phase = activity.get("phase", "")
labels = {
"queued": "Generation Waiting In Queue",
"preparing": "Generation Preparing",
"script": "Generation Writing Script",
"terms": "Generation Preparing Terms",
"audio": "Generation Preparing Audio",
"subtitle": "Generation Preparing Subtitles",
"materials": "Generation Preparing Materials",
"submitting": "Generation Submitting",
"checking": "Generation Checking",
"remote_queued": "Generation Remote Queued",
"generating": "Generation Remote Running",
"generated": "Generation Remote Completed",
"downloading": "Generation Downloading",
"verifying": "Generation Verifying",
"rendering": "Generation Rendering",
}
st.info(tr(labels.get(phase, "Generating Video")))
now = time.time()
started = (task.get("generation_submitted_at") or task.get("generation_started_at")
or activity.get("started_at") or now)
elapsed = max(0, int(now - float(started)))
st.caption(tr("Generation Elapsed").format(time=f"{elapsed // 60:02d}:{elapsed % 60:02d}"))
if phase in {"submitting", "remote_queued", "generating", "checking"}:
st.caption(tr("Generation Long Wait Hint"))
if activity.get("retrying"):
st.warning(tr("Generation Poll Retry Hint"))
if activity.get("confirmed_at"):
age = max(0, int(now - float(activity["confirmed_at"])))
st.caption(tr("Generation Last Checked").format(seconds=age))
if phase == "downloading" and activity.get("downloaded_bytes"):
downloaded = activity["downloaded_bytes"] / (1024 * 1024)
st.caption(tr("Generation Downloaded").format(size=f"{downloaded:.1f}"))
if activity.get("remote_id"):
st.caption(tr("Generation Remote Task").format(task_id=activity["remote_id"]))
def _render_generation_task_snapshot(task_id, task):
"""根据状态存储中的快照渲染进度、失败原因或最终成片。"""
if not task:
st.info(tr("Generating Video"))
_render_generation_logs(task_id)
return
state = _normalize_task_state(task.get("state"))
progress = max(0, min(100, int(task.get("progress", 0) or 0)))
if state == const.TASK_STATE_PROCESSING:
_render_generation_activity(task)
st.progress(
progress,
text=f"{tr('Task Progress')}: {progress}%",
)
_render_generation_logs(task_id)
return
video_files = task.get("videos") or []
if state == const.TASK_STATE_FAILED:
error = str(task.get("error") or "").strip()
error_key = task.get("error_key")
error_params = task.get("error_params") or {}
if (
isinstance(error_key, str)
and error_key.startswith("codex.")
and isinstance(error_params, Mapping)
):
error = tr(error_key).format(**error_params)
message = tr("Video Generation Failed")
st.error(f"{message}: {error}" if error else message)
activity = task.get("generation_activity") or {}
if activity.get("remote_id"):
st.warning(tr("Generation Existing Task Hint").format(task_id=activity["remote_id"]))
if not video_files:
_render_generation_logs(task_id)
return
elif state != const.TASK_STATE_COMPLETE or not video_files:
st.error(tr("Video Generation Failed"))
_render_generation_logs(task_id)
return
if state == const.TASK_STATE_COMPLETE:
st.success(tr("Video Generation Completed"))
for warning in task.get("warnings") or []:
if isinstance(warning, Mapping) and warning.get("code") == "batch_materials_reused":
st.warning(
tr("Batch Material Reuse Warning").format(
index=warning.get("video_index", ""), count=warning.get("count", 0)
)
)
elif isinstance(warning, Mapping) and warning.get("code") == "sonilo_bgm_failed":
st.warning(
tr("Sonilo BGM Fallback Warning").format(
index=warning.get("video_index", "")
)
)
elif (
isinstance(warning, Mapping)
and warning.get("code") == "elevenlabs_bgm_failed"
):
st.warning(
tr("ElevenLabs BGM Fallback Warning").format(
index=warning.get("video_index", "")
)
)
else:
st.warning(str(warning))
available_videos = [
(index, url)
for index, url in enumerate(video_files)
if os.path.isfile(url)
]
for index, url in enumerate(video_files):
if os.path.isfile(url):
continue
logger.warning(
f"generated video is unavailable: "
f"task_id={task_id}, video_file={url}"
)
try:
if not available_videos:
st.warning(tr("Generated Video Files Unavailable"))
else:
player_cols = st.columns(len(available_videos) * 2 + 1)
for player_index, (video_index, url) in enumerate(available_videos):
with player_cols[player_index * 2 + 1]:
st.video(url)
download_label = tr("Download Video")
if len(video_files) > 1:
download_label = f"{download_label} {video_index + 1}"
download_name = _build_video_download_name(
task.get("video_subject"),
video_index + 1,
len(video_files),
)
with open(url, "rb") as video_file:
st.download_button(
download_label,
data=video_file,
file_name=download_name,
mime=mimetypes.guess_type(url)[0] or "video/mp4",
key=f"download_generated_video_{task_id}_{video_index}",
icon=":material/download:",
on_click="ignore",
use_container_width=True,
)
except Exception as exc:
logger.exception(
f"failed to render generated video preview: task_id={task_id}, "
f"video_files={video_files}, error={exc}"
)
_render_generation_logs(task_id)
if (
state == const.TASK_STATE_COMPLETE
and st.session_state.get("handled_generation_task_id") != task_id
):
# Fragment 可能重复渲染同一个完成任务。无论是否开启自动打开目录,
# 每个任务都只处理一次完成事件,避免重复弹出资源管理器或重复写入日志。
st.session_state["handled_generation_task_id"] = task_id
if config.ui.get("open_task_folder_on_completion", True):
open_task_folder(task_id)
logger.info(f"{tr('Video Generation Completed')}: task_id={task_id}")
@st.fragment(run_every=webui_task.TASK_LOG_REFRESH_INTERVAL_SECONDS)
def _render_running_generation_task(task_id):
"""只在任务运行期间轮询;结束后切回静态结果,停止不必要的定时刷新。"""
try:
task = sm.state.get_task(task_id)
except Exception as exc:
logger.exception(
f"failed to query WebUI generation task: task_id={task_id}, error={exc}"
)
st.warning(tr("Generation Status Unavailable"))
return
if not task and not webui_task.is_generation_active(task_id):
st.warning(tr("Generation Task Unavailable"))
return
state = _normalize_task_state((task or {}).get("state"))
if state in {const.TASK_STATE_COMPLETE, const.TASK_STATE_FAILED}:
_remove_active_generation_task(task_id)
# 完整页面脚本现在没有耗时生成逻辑,可以安全 rerun 并把结果改为静态
# 渲染。这样任务结束后不会让浏览器永久保留一个两秒轮询的 Fragment。
st.rerun(scope="app")
_render_generation_task_snapshot(task_id, task)
if not webui_task.is_generation_active(task_id):
# 持久状态可能属于重启前或另一服务进程,不能断言远端任务已停止。
# 只允许解除当前页面的跟踪,不取消任务、不改写其真实状态。
st.warning(tr("Generation Untracked Worker Hint"))
if st.button(tr("Stop Tracking Generation"), key=f"stop_tracking_generation_{task_id}"):
_remove_active_generation_task(task_id)
st.session_state["current_generation_task_id"] = ""
st.query_params.pop("generation_task", None)
st.rerun(scope="app")
def _render_current_generation_task():
"""在生成按钮下方恢复当前页面最近提交任务的可查询 UI。"""
task_id = _current_generation_task_id()
if not task_id:
return
try:
task = sm.state.get_task(task_id)
except Exception as exc:
logger.exception(
f"failed to query current WebUI task: task_id={task_id}, error={exc}"
)
# 即使首次读取失败,也要注册 Fragment,让状态后端恢复后自动继续。
_render_running_generation_task(task_id)
return
if not task and not webui_task.is_generation_active(task_id):
st.warning(tr("Generation Task Unavailable"))
return
state = _normalize_task_state((task or {}).get("state"))
if state in {const.TASK_STATE_COMPLETE, const.TASK_STATE_FAILED}:
_remove_active_generation_task(task_id)
_render_generation_task_snapshot(task_id, task)
return
_render_running_generation_task(task_id)
def get_llm_provider_tips(provider_id, **kwargs):
# LLM provider 说明文案统一使用 `llm_provider_tips.` 规则。
# 这样新增 provider 时只需要在 locale 中补文案;没有文案时不展示提示块,
# 避免 Main.py 里继续堆叠大量中英文硬编码说明。
provider = get_llm_provider(provider_id)
if provider is None:
return ""
# Provider 配置说明目前统一维护中文和英文两套规范模板;其它界面语言
# 统一使用英文,避免在 locale 中复制英文后长期不同步。后续某个语种完成
# 全量翻译后,再将它加入这里的独立维护范围。
ui_language = st.session_state.get("ui_language", "en")
tips_language = ui_language if ui_language in {"zh", "en"} else "en"
tips = (
locales.get(tips_language, {}).get("Translation", {}).get(provider.tips_key, "")
)
if not tips:
return tips
service_endpoint = provider.preferred_service_endpoint(
prefer_international=tips_language == "en"
)
api_key_url = (
service_endpoint.api_key_url
if service_endpoint
else provider.effective_api_key_url()
)
format_context = {
"api_key_url": api_key_url,
"default_model": provider.default_model,
"default_base_url": (
service_endpoint.base_url
if service_endpoint
else provider.effective_default_base_url
),
"model_docs_url": (
service_endpoint.model_docs_url
if service_endpoint and service_endpoint.model_docs_url
else provider.effective_model_docs_url(
prefer_international=tips_language == "en"
)
),
**{
f"default_{field.config_suffix}": field.default_value
for field in provider.extra_fields
},
**kwargs,
}
try:
return tips.format(**format_context)
except Exception as e:
logger.warning(f"format llm provider tips failed: {provider_id}, {e}")
return tips
def format_llm_connection_error(provider_id, base_url, error):
"""为可明确定位的鉴权错误补充配置检查建议,同时保留原始响应。"""
error_text = str(error or "").strip()
if provider_id in {"openai_codex", "openai_chatgpt"}:
if provider_id == "openai_chatgpt":
from app.services.chatgpt_oauth import ERROR_MESSAGES
message_prefix = "chatgpt.oauth"
else:
from app.services.codex_oauth import ERROR_MESSAGES
message_prefix = "codex.oauth"
for key, message in ERROR_MESSAGES.items():
if error_text == message:
return tr(f"{message_prefix}.{key}")
status_match = re.fullmatch(
r"OpenAI - (?:Codex|ChatGPT \(Native OAuth\)) returned HTTP (\d+)\. Check the selected model and your account access\.",
error_text,
)
if status_match:
return tr(f"{message_prefix}.http_error").format(status=status_match[1])
normalized_error = error_text.lower()
authentication_markers = (
"401",
"authentication",
"invalid api key",
"invalid_api_key",
"unauthorized",
)
provider = get_llm_provider(provider_id)
if provider is None or not provider.service_endpoints or not any(
marker in normalized_error for marker in authentication_markers
):
return error_text
message = tr_optional(
provider.authentication_error_key,
fallback_language="en",
)
if not message:
return error_text
return message.format(base_url=base_url or "-", error=error_text)
def get_llm_provider_label(provider):
return tr_optional(provider.label_key) or provider.default_label
def get_tts_provider_tips(provider_id):
# TTS 配置说明与 LLM Provider 采用相同维护策略:只维护中英文,
# 其它界面语言统一回退英文,避免复制后长期不同步。
ui_language = st.session_state.get("ui_language", "en")
tips_language = ui_language if ui_language in {"zh", "en"} else "en"
return (
locales.get(tips_language, {})
.get("Translation", {})
.get(f"tts_provider_tips.{provider_id}", "")
)
def localized_widget_key(name, *parts):
# 部分 Streamlit selectbox 使用稳定 key 记住选择状态,但展示文本来自 locale。
# 语言切换时把语言也放进 key,可以强制重建控件,避免选中项仍显示旧语言。
language = st.session_state.get("ui_language", config.ui.get("language", ""))
suffix_parts = [name, language, *[str(part) for part in parts if part]]
return "_".join(suffix_parts)
def stable_selectbox(label, options, default_value, key, format_func=None, **kwargs):
# Streamlit 1.59 对 selectbox 的状态复用更敏感:如果控件没有固定 key,
# 或者真实选项只是一组临时下标,页面 rerun 后容易被重新计算的 index 覆盖,
# 表现为用户第一次选择不生效、需要再选一次。这个 helper 统一用稳定业务值
# 作为真实选项,并在 session_state 里保存该值;展示文案只通过 format_func
# 转换,避免翻译文案、选项顺序或上游配置变化影响选择状态。
options = list(options)
if not options:
raise ValueError(f"selectbox options cannot be empty: {key}")
if default_value not in options:
default_value = options[0]
widget_key = localized_widget_key(key)
selected_value = st.session_state.get(widget_key)
accepts_custom_value = bool(kwargs.get("accept_new_options"))
has_valid_custom_value = (
accepts_custom_value
and isinstance(selected_value, str)
and bool(selected_value.strip())
)
if selected_value not in options and not has_valid_custom_value:
# 如果上游选项发生变化(例如切换 TTS provider 后声音列表变了),
# 旧值已经不合法。控件创建前直接初始化 session_state,之后只让 key
# 管理状态,不再同时传入 index。这样可以避免 Streamlit 在 rerun 时
# 用重新计算的 index 覆盖用户刚选择的值,导致第一次选择不生效。
st.session_state[widget_key] = default_value
if format_func is None:
format_func = str
return st.selectbox(
label,
options=options,
format_func=format_func,
key=widget_key,
**kwargs,
)
# Streamlit 原生 selectbox 暂不支持 HTML optgroup。这里使用 1.59 自带的
# Components v2 封装原生