wehub-resource-sync
3222 lines
133 KiB
Python
3222 lines
133 KiB
Python
import html
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
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 作为独立入口运行时，需要将项目根目录加入模块搜索路径。
|
||
root_dir = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
|
||
if root_dir not in sys.path:
|
||
sys.path.append(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 (
|
||
MaterialInfo,
|
||
VideoAspect,
|
||
VideoConcatMode,
|
||
VideoParams,
|
||
VideoTransitionMode,
|
||
)
|
||
from app.services import cache_manager, llm, video, voice
|
||
from app.services import state as sm
|
||
from app.services import task as tm
|
||
from app.utils import utils
|
||
|
||
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"<style>{style_file.read_text(encoding='utf-8')}</style>"
|
||
st.markdown(streamlit_style, unsafe_allow_html=True)
|
||
# 定义资源目录
|
||
font_dir = os.path.join(root_dir, "resource", "fonts")
|
||
song_dir = os.path.join(root_dir, "resource", "songs")
|
||
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"]
|
||
ONBOARDING_TOUR_KEY = "mpt-onboarding-v1"
|
||
VOICE_MODE_TTS = "tts"
|
||
VOICE_MODE_UPLOAD = "upload"
|
||
VOICE_MODE_NONE = "none"
|
||
# “默认”是 WebUI 专用哨兵，不会写入 config.toml，也不会传给 FFmpeg。
|
||
# 后端在 video_codec 未配置时继续采用稳定的 libx264；单独保留该哨兵可以区分
|
||
# “跟随项目默认策略”和“用户明确固定 libx264”，便于未来安全调整默认策略。
|
||
DEFAULT_VIDEO_CODEC_OPTION = "__default__"
|
||
DEFAULT_SUBTITLE_SETTINGS = {
|
||
"subtitle_enabled": True,
|
||
"font_name": "MicrosoftYaHeiBold.ttc",
|
||
"subtitle_position": "bottom",
|
||
"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 = {
|
||
".mp4",
|
||
".mov",
|
||
".avi",
|
||
".flv",
|
||
".mkv",
|
||
".jpg",
|
||
".jpeg",
|
||
".png",
|
||
}
|
||
CUSTOM_AUDIO_EXTENSIONS = {".mp3", ".wav", ".m4a", ".aac", ".flac", ".ogg"}
|
||
_FINAL_VIDEO_PATTERN = re.compile(
|
||
r"^final-(?P<index>\d+)\.(?P<extension>mp4|mov|mkv|webm)$",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# 启动配置、会话状态与本地化
|
||
# -----------------------------------------------------------------------------
|
||
|
||
|
||
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 同步一次，
|
||
# 可以保证按钮逻辑和输入框显示逻辑使用同一份最新配置。
|
||
config.chatterbox["base_url"] = (
|
||
st.session_state.get(
|
||
"chatterbox_base_url_input",
|
||
config.chatterbox.get("base_url") or DEFAULT_CHATTERBOX_BASE_URL,
|
||
)
|
||
or ""
|
||
).strip()
|
||
config.chatterbox["api_key"] = st.session_state.get(
|
||
"chatterbox_api_key_input", config.chatterbox.get("api_key", "")
|
||
)
|
||
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()
|
||
config.chatterbox["voices"] = _parse_chatterbox_voices(
|
||
st.session_state.get(
|
||
"chatterbox_voices_input",
|
||
config.chatterbox.get("voices") or DEFAULT_CHATTERBOX_VOICES,
|
||
)
|
||
)
|
||
|
||
|
||
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 '<empty>'}"
|
||
)
|
||
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 _initialize_session_state():
|
||
"""集中初始化跨 rerun 保留的页面状态。"""
|
||
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(),
|
||
)
|
||
|
||
if "ui_language" not in st.session_state:
|
||
# 语言初始化只在当前浏览器会话首次运行时记录一次。Accept-Language 可以
|
||
# 帮助判断浏览器首选语言顺序，而 st.context.locale 对应浏览器实际返回的
|
||
# navigator.language；两者一起记录能定位“系统是中文但页面显示英文”。
|
||
accept_language = st.context.headers.get("Accept-Language", "")
|
||
logger.info(
|
||
"initialize UI language: "
|
||
f"saved_language={saved_ui_language or '<empty>'}, "
|
||
f"browser_locale={browser_locale or '<empty>'}, "
|
||
f"accept_language={accept_language or '<empty>'}, "
|
||
f"resolved_language={initial_ui_language}, "
|
||
f"supported_languages={','.join(locales.keys())}"
|
||
)
|
||
|
||
defaults = {
|
||
"video_subject": "",
|
||
"video_script": "",
|
||
"video_terms": "",
|
||
"video_script_prompt": "",
|
||
"custom_system_prompt": llm.DEFAULT_SCRIPT_SYSTEM_PROMPT,
|
||
"match_materials_to_script": bool(
|
||
config.app.get("match_materials_to_script", False)
|
||
),
|
||
"ui_language": initial_ui_language,
|
||
# 已落盘的本地素材允许用户只修改文案后继续复用。
|
||
"local_video_materials": [],
|
||
# 控件交互会触发 rerun，日志必须保存在会话中才能持续显示。
|
||
"generation_log_records": [],
|
||
# 生成按钮回调先登记任务，使顶部入口能立即显示运行中数量。
|
||
"active_generation_tasks": {},
|
||
}
|
||
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"], {})
|
||
return loc.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:
|
||
return json.load(f)
|
||
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:
|
||
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 中显示“生成中”数量。
|
||
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 script_data else {}
|
||
video_file = _find_final_task_video(task_path)
|
||
subject = (
|
||
params_data.get("video_subject")
|
||
or script_data.get("script", "")[: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)}
|
||
|
||
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 = []
|
||
|
||
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
|
||
)
|
||
|
||
history_tasks[task_id] = {
|
||
"task_id": task_id,
|
||
"subject": subject,
|
||
"state": task.get("state"),
|
||
"progress": int(task.get("progress", 0) or 0),
|
||
"mtime": os.path.getmtime(task_path)
|
||
if os.path.isdir(task_path)
|
||
else history_task.get("mtime", 0),
|
||
"task_path": task_path,
|
||
"video_file": video_file,
|
||
"source": "runtime",
|
||
}
|
||
|
||
for task_id, active_task in _active_generation_tasks().items():
|
||
history_task = history_tasks.get(task_id, {})
|
||
if history_task and _task_state_filter_key(history_task) == "complete":
|
||
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 _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 os.path.isdir(normalized_path):
|
||
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
|
||
|
||
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
|
||
|
||
current_state = current_task.get("state") if current_task else None
|
||
is_processing = (
|
||
any(
|
||
_normalize_task_state(state) == const.TASK_STATE_PROCESSING
|
||
for state in (task_state, current_state)
|
||
)
|
||
or task_id in _active_generation_tasks()
|
||
)
|
||
if is_processing:
|
||
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 hasattr(sm.state, "delete_task"):
|
||
sm.state.delete_task(task_id)
|
||
if os.path.isdir(normalized_path):
|
||
shutil.rmtree(normalized_path)
|
||
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 _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"
|
||
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_processing
|
||
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_processing,
|
||
):
|
||
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)
|
||
|
||
|
||
@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_elevenlabs_voice(voice_name):
|
||
return "elevenlabs"
|
||
if voice.is_chatterbox_voice(voice_name):
|
||
return "chatterbox"
|
||
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
|
||
|
||
params = payload["params"]
|
||
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_clip_duration_select", params.get("video_clip_duration", 3)
|
||
)
|
||
_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))
|
||
st.session_state["custom_bgm_file_input"] = params.get("bgm_file") 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"
|
||
)
|
||
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["task_restore_upload_requirements"] = (
|
||
_build_restore_upload_requirements(params)
|
||
)
|
||
|
||
st.session_state["task_restore_succeeded"] = True
|
||
logger.info(f"restored task configuration: {payload['task_id']}")
|
||
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 _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:
|
||
st.markdown(
|
||
f"""
|
||
<h1 class="mpt-brand">
|
||
<span class="mpt-brand__name">MoneyPrinterTurbo</span>
|
||
<a class="mpt-brand__version"
|
||
href="https://github.com/harry0703/MoneyPrinterTurbo"
|
||
target="_blank"
|
||
rel="noopener noreferrer"
|
||
aria-label="Open MoneyPrinterTurbo on GitHub"
|
||
title="Open project on GitHub">v{config.project_version}</a>
|
||
</h1>
|
||
""",
|
||
unsafe_allow_html=True,
|
||
)
|
||
|
||
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()
|
||
|
||
if st.button(
|
||
tr("Settings"),
|
||
key="open_settings_dialog_button",
|
||
type="secondary",
|
||
icon=":material/settings:",
|
||
width="content",
|
||
):
|
||
st.session_state["settings_dialog_open"] = True
|
||
|
||
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 '<empty>'}, "
|
||
f"selected_language={selected_language_code}"
|
||
)
|
||
st.session_state["ui_language"] = selected_language_code
|
||
# 浏览器自动识别只影响当前会话；只有用户主动切换下拉框时才
|
||
# 写入 config.toml，后续新会话将优先使用该明确选择。
|
||
config.ui["language"] = selected_language_code
|
||
config.save_config()
|
||
# 切换语言后强制刷新，避免 selectbox 继续展示旧语言文案。
|
||
st.rerun()
|
||
|
||
|
||
support_locales = [
|
||
"zh-CN",
|
||
"zh-HK",
|
||
"zh-TW",
|
||
"de-DE",
|
||
"en-US",
|
||
"es-ES",
|
||
"fr-FR",
|
||
"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"):
|
||
fonts.append(file)
|
||
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(os.path.join(root_dir, "storage", "tasks"))
|
||
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 页面脚本；如果每次都执行 logger.remove()，会把其它会话或
|
||
# 正在生成任务使用的临时 Handler 一并删除。cache_resource 保证每个进程只
|
||
# 初始化一次，页面级生成日志仍通过各自的 handler id 单独添加和移除。
|
||
logger.remove()
|
||
_lvl = "DEBUG"
|
||
|
||
def format_record(record):
|
||
# 日志统一展示项目相对路径，并移除消息中可能重复出现的项目根目录。
|
||
file_path = record["file"].path
|
||
relative_path = os.path.relpath(file_path, root_dir)
|
||
record["file"].path = f"./{relative_path}"
|
||
record["message"] = record["message"].replace(root_dir, ".")
|
||
|
||
_format = (
|
||
"<green>{time:%Y-%m-%d %H:%M:%S}</> | "
|
||
+ "<level>{level}</> | "
|
||
+ '"{file.path}:{line}":<blue> {function}</> '
|
||
+ "- <level>{message}</>"
|
||
+ "\n"
|
||
)
|
||
return _format
|
||
|
||
return logger.add(
|
||
sys.stdout,
|
||
level=_lvl,
|
||
format=format_record,
|
||
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(container):
|
||
if config.ui.get("hide_log", False):
|
||
container.empty()
|
||
return
|
||
|
||
log_records = st.session_state.get("generation_log_records", [])
|
||
if not log_records:
|
||
container.empty()
|
||
return
|
||
|
||
with container:
|
||
st.code("\n".join(log_records))
|
||
|
||
|
||
def remove_logger_handler_safely(handler_id):
|
||
try:
|
||
logger.remove(handler_id)
|
||
except ValueError:
|
||
# Streamlit 交互可能触发 rerun，Loguru handler 在 finally 执行前
|
||
# 已经被其它初始化流程移除。这里忽略缺失 handler，避免生成成功后
|
||
# 因清理日志监听器失败而把页面打成异常。
|
||
logger.debug(f"log handler already removed: {handler_id}")
|
||
|
||
|
||
def get_llm_provider_tips(provider_id, **kwargs):
|
||
# LLM provider 说明文案统一使用 `llm_provider_tips.<provider_id>` 规则。
|
||
# 这样新增 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
|
||
|
||
format_context = {
|
||
"api_key_url": provider.api_key_url,
|
||
"default_model": provider.default_model,
|
||
"default_base_url": provider.default_base_url,
|
||
**{
|
||
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 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)
|
||
if selected_value not in options:
|
||
# 如果上游选项发生变化（例如切换 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,
|
||
)
|
||
|
||
|
||
def sync_script_order_concat_mode():
|
||
"""在文案顺序匹配开启时固定使用顺序拼接，并在关闭后恢复原选择。"""
|
||
widget_key = localized_widget_key("video_concat_mode_select")
|
||
previous_key = "video_concat_mode_before_script_order_match"
|
||
match_script_order = bool(st.session_state.get("match_materials_to_script", False))
|
||
|
||
if match_script_order:
|
||
current_mode = st.session_state.get(widget_key, VideoConcatMode.random.value)
|
||
if current_mode != VideoConcatMode.sequential.value:
|
||
st.session_state[previous_key] = current_mode
|
||
st.session_state[widget_key] = VideoConcatMode.sequential.value
|
||
return
|
||
|
||
previous_mode = st.session_state.pop(previous_key, None)
|
||
if previous_mode in {
|
||
VideoConcatMode.sequential.value,
|
||
VideoConcatMode.random.value,
|
||
}:
|
||
st.session_state[widget_key] = previous_mode
|
||
|
||
|
||
def reset_script_system_prompt():
|
||
"""将高级脚本设置中的系统提示词恢复为当前版本的默认内容。"""
|
||
st.session_state["custom_system_prompt"] = llm.DEFAULT_SCRIPT_SYSTEM_PROMPT
|
||
|
||
|
||
def reset_subtitle_settings():
|
||
"""恢复 WebUI 字幕控件和持久化配置中的默认值。"""
|
||
defaults = DEFAULT_SUBTITLE_SETTINGS
|
||
st.session_state["subtitle_enabled_checkbox"] = defaults["subtitle_enabled"]
|
||
_set_stable_widget_value("font_name_select", defaults["font_name"])
|
||
_set_stable_widget_value("subtitle_position_select", defaults["subtitle_position"])
|
||
st.session_state["custom_position_input"] = str(defaults["custom_position"])
|
||
st.session_state["font_color_picker"] = defaults["text_fore_color"]
|
||
st.session_state["font_size_slider"] = defaults["font_size"]
|
||
st.session_state["stroke_color_picker"] = defaults["stroke_color"]
|
||
st.session_state["stroke_width_slider"] = defaults["stroke_width"]
|
||
st.session_state["subtitle_background_enabled_checkbox"] = defaults[
|
||
"subtitle_background_enabled"
|
||
]
|
||
st.session_state["subtitle_background_color_picker"] = defaults[
|
||
"subtitle_background_color"
|
||
]
|
||
st.session_state["rounded_subtitle_background_checkbox"] = defaults[
|
||
"rounded_subtitle_background"
|
||
]
|
||
|
||
# 同步会持久化的 UI 选项，确保恢复后刷新页面仍保持默认设置。
|
||
for key in (
|
||
"font_name",
|
||
"subtitle_position",
|
||
"custom_position",
|
||
"text_fore_color",
|
||
"font_size",
|
||
"subtitle_background_enabled",
|
||
"subtitle_background_color",
|
||
"rounded_subtitle_background",
|
||
):
|
||
config.ui[key] = defaults[key]
|
||
|
||
|
||
@st.dialog(tr("Final Prompt Preview"), width="large")
|
||
def render_script_prompt_preview(prompt):
|
||
"""展示将要发送给大模型的完整脚本生成提示词。"""
|
||
st.code(prompt, language="markdown", wrap_lines=True)
|
||
|
||
|
||
def stable_segmented_control(
|
||
label, options, default_value, key, format_func=None, **kwargs
|
||
):
|
||
"""使用稳定业务值创建单选分段控件，避免语言切换后状态被展示文案覆盖。"""
|
||
options = list(options)
|
||
if not options:
|
||
raise ValueError(f"segmented control options cannot be empty: {key}")
|
||
|
||
if default_value not in options:
|
||
default_value = options[0]
|
||
|
||
widget_key = localized_widget_key(key)
|
||
if st.session_state.get(widget_key) not in options:
|
||
st.session_state[widget_key] = default_value
|
||
|
||
return st.segmented_control(
|
||
label,
|
||
options=options,
|
||
selection_mode="single",
|
||
required=True,
|
||
format_func=format_func or str,
|
||
key=widget_key,
|
||
**kwargs,
|
||
)
|
||
|
||
|
||
@st.cache_data(ttl=300, show_spinner=False)
|
||
def get_groq_model_ids(api_key: str, base_url: str) -> list[str]:
|
||
if not api_key:
|
||
return []
|
||
|
||
normalized_base_url = (
|
||
(base_url or "https://api.groq.com/openai/v1").strip().rstrip("/")
|
||
)
|
||
models_url = f"{normalized_base_url}/models"
|
||
|
||
try:
|
||
response = requests.get(
|
||
models_url,
|
||
headers={"Authorization": f"Bearer {api_key}"},
|
||
timeout=10,
|
||
)
|
||
response.raise_for_status()
|
||
payload = response.json()
|
||
data = payload.get("data", [])
|
||
|
||
model_ids = []
|
||
for item in data:
|
||
if isinstance(item, dict):
|
||
model_id = item.get("id")
|
||
if isinstance(model_id, str) and model_id.strip():
|
||
model_ids.append(model_id.strip())
|
||
|
||
return sorted(set(model_ids))
|
||
except Exception as e:
|
||
logger.warning(f"failed to fetch groq models: {e}")
|
||
return []
|
||
|
||
|
||
def _get_material_api_keys(config_key):
|
||
"""将配置中的素材 API Key 统一转换为 WebUI 可编辑字符串。"""
|
||
api_keys = config.app.get(config_key, [])
|
||
if isinstance(api_keys, str):
|
||
api_keys = [api_keys]
|
||
return ", ".join(api_keys)
|
||
|
||
|
||
def _save_material_api_keys(config_key, value):
|
||
"""保存逗号分隔的素材 API Key，并允许用户显式清空旧配置。"""
|
||
normalized_value = value.replace(" ", "")
|
||
config.app[config_key] = normalized_value.split(",") if normalized_value else []
|
||
|
||
|
||
def _format_file_size(size_bytes):
|
||
"""将字节数格式化为适合设置页展示的紧凑容量文本。"""
|
||
size = float(max(0, size_bytes))
|
||
units = ("B", "KB", "MB", "GB", "TB")
|
||
for unit in units:
|
||
if size < 1024 or unit == units[-1]:
|
||
return f"{size:.0f} {unit}" if unit in ("B", "KB") else f"{size:.2f} {unit}"
|
||
size /= 1024
|
||
return f"{size_bytes} B"
|
||
|
||
|
||
@st.cache_data(ttl=30, show_spinner=False)
|
||
def _get_video_cache_stats(max_age_days=None):
|
||
"""
|
||
短周期缓存目录统计，避免设置弹窗内普通控件交互反复扫描大量文件。
|
||
|
||
缓存键包含清理天数，因此切换范围只会为每个范围扫描一次；主动刷新或清理
|
||
完成后会显式清空，最多 30 秒的缓存不会影响实际删除时的二次扫描。
|
||
"""
|
||
return cache_manager.get_video_cache_stats(max_age_days=max_age_days)
|
||
|
||
|
||
def _render_cache_management_settings(panel):
|
||
"""渲染默认在线视频素材缓存的统计、预览和安全清理操作。"""
|
||
with panel:
|
||
cleanup_message = st.session_state.pop("video_cache_cleanup_message", None)
|
||
if cleanup_message:
|
||
message_type, message = cleanup_message
|
||
if message_type == "success":
|
||
st.success(message)
|
||
else:
|
||
st.warning(message)
|
||
|
||
st.caption(tr("Video Cache Directory"))
|
||
st.code(cache_manager.video_cache_dir(), language="text")
|
||
|
||
total_stats = _get_video_cache_stats()
|
||
metric_count, metric_size, metric_oldest = st.columns(3)
|
||
metric_count.metric(tr("Cache File Count"), total_stats.file_count)
|
||
metric_size.metric(
|
||
tr("Cache Total Size"), _format_file_size(total_stats.total_size)
|
||
)
|
||
oldest_text = (
|
||
datetime.fromtimestamp(total_stats.oldest_mtime).strftime("%Y-%m-%d")
|
||
if total_stats.oldest_mtime is not None
|
||
else "-"
|
||
)
|
||
metric_oldest.metric(tr("Oldest Cache Date"), oldest_text)
|
||
|
||
st.caption(tr("Video Cache Management Help"))
|
||
cleanup_options = (30, 7, 90, None)
|
||
cleanup_labels = {
|
||
30: tr("Cache Older Than 30 Days"),
|
||
7: tr("Cache Older Than 7 Days"),
|
||
90: tr("Cache Older Than 90 Days"),
|
||
None: tr("All Video Cache"),
|
||
}
|
||
max_age_days = st.selectbox(
|
||
tr("Cache Cleanup Range"),
|
||
options=cleanup_options,
|
||
format_func=lambda value: cleanup_labels[value],
|
||
key="video_cache_cleanup_range",
|
||
)
|
||
cleanup_preview = _get_video_cache_stats(max_age_days=max_age_days)
|
||
st.info(
|
||
tr("Cache Cleanup Preview").format(
|
||
count=cleanup_preview.file_count,
|
||
size=_format_file_size(cleanup_preview.total_size),
|
||
)
|
||
)
|
||
|
||
confirm_nonce = st.session_state.get("video_cache_cleanup_confirm_nonce", 0)
|
||
confirmed = st.checkbox(
|
||
tr("Confirm Cache Cleanup"),
|
||
key=f"video_cache_cleanup_confirm_{confirm_nonce}",
|
||
)
|
||
refresh_col, open_col, cleanup_col = st.columns(3)
|
||
if refresh_col.button(
|
||
tr("Refresh Cache Stats"),
|
||
key="refresh_video_cache_stats",
|
||
use_container_width=True,
|
||
icon=":material/refresh:",
|
||
):
|
||
_get_video_cache_stats.clear()
|
||
st.rerun(scope="fragment")
|
||
|
||
if open_col.button(
|
||
tr("Open Cache Directory"),
|
||
key="open_video_cache_directory",
|
||
use_container_width=True,
|
||
icon=":material/folder_open:",
|
||
):
|
||
webbrowser.open(Path(cache_manager.video_cache_dir()).as_uri())
|
||
|
||
cleanup_disabled = not confirmed or cleanup_preview.file_count == 0
|
||
if cleanup_col.button(
|
||
tr("Clean Cache Now"),
|
||
key="clean_video_cache_now",
|
||
type="primary",
|
||
disabled=cleanup_disabled,
|
||
use_container_width=True,
|
||
icon=":material/delete_sweep:",
|
||
):
|
||
result = cache_manager.clean_video_cache(max_age_days=max_age_days)
|
||
message_key = (
|
||
"Cache Cleanup Completed With Failures"
|
||
if result.failed_count
|
||
else "Cache Cleanup Completed"
|
||
)
|
||
st.session_state["video_cache_cleanup_message"] = (
|
||
"warning" if result.failed_count else "success",
|
||
tr(message_key).format(
|
||
count=result.deleted_count,
|
||
size=_format_file_size(result.deleted_size),
|
||
failed=result.failed_count,
|
||
),
|
||
)
|
||
# Streamlit 不允许在控件实例化后修改同名 session_state。通过递增
|
||
# nonce 让下一次 fragment rerun 创建未勾选的新控件，避免清理完成后
|
||
# 危险确认状态被继续保留。
|
||
st.session_state["video_cache_cleanup_confirm_nonce"] = confirm_nonce + 1
|
||
_get_video_cache_stats.clear()
|
||
st.rerun(scope="fragment")
|
||
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# 设置与提示词弹窗
|
||
# -----------------------------------------------------------------------------
|
||
|
||
|
||
# 设置属于低频操作，使用中等尺寸 Dialog 避免长期占用主页面纵向空间，
|
||
# 同时控制阅读行宽，避免弹窗在宽屏设备上显得过于松散。
|
||
# Dialog 继承 fragment 行为，内部控件交互只重绘弹窗；函数末尾单独保存配置，
|
||
# 关闭时通过回调触发整页同步，确保生成流程读取最新 Provider 和界面设置。
|
||
@st.dialog(
|
||
tr("Settings"),
|
||
width="medium",
|
||
on_dismiss=_dismiss_settings_dialog,
|
||
)
|
||
def _render_settings_dialog():
|
||
with st.container():
|
||
# 历史 hide_config 只用于隐藏旧基础设置面板。改为固定设置入口后，该值
|
||
# 不再有用户可见意义，统一迁移为 false，避免旧配置影响后续版本。
|
||
config.app["hide_config"] = False
|
||
(
|
||
middle_config_panel,
|
||
right_config_panel,
|
||
cache_config_panel,
|
||
left_config_panel,
|
||
) = st.tabs(
|
||
[
|
||
tr("LLM Settings Tab"),
|
||
tr("Material API Tab"),
|
||
tr("Cache Management Tab"),
|
||
tr("Interface Settings Tab"),
|
||
]
|
||
)
|
||
|
||
# 左侧面板 - 日志设置
|
||
with left_config_panel:
|
||
hide_log = st.checkbox(
|
||
tr("Hide Log"),
|
||
value=config.ui.get("hide_log", False),
|
||
key="hide_log_checkbox",
|
||
)
|
||
config.ui["hide_log"] = hide_log
|
||
|
||
_render_cache_management_settings(cache_config_panel)
|
||
|
||
# 中间面板 - LLM 设置
|
||
|
||
with middle_config_panel:
|
||
# 下拉顺序、默认 label 和稳定 provider id 全部来自 Registry；locale
|
||
# 只覆盖展示文案，不再让 Main.py 维护第二份 Provider 列表。
|
||
llm_provider_ids = [
|
||
provider.provider_id for provider in LLM_PROVIDER_REGISTRY
|
||
]
|
||
llm_provider_labels = {
|
||
provider.provider_id: get_llm_provider_label(provider)
|
||
for provider in LLM_PROVIDER_REGISTRY
|
||
}
|
||
saved_llm_provider = config.app.get(
|
||
"llm_provider", DEFAULT_LLM_PROVIDER_ID
|
||
).lower()
|
||
if saved_llm_provider not in llm_provider_ids:
|
||
saved_llm_provider = DEFAULT_LLM_PROVIDER_ID
|
||
|
||
llm_provider = stable_selectbox(
|
||
tr("LLM Provider"),
|
||
options=llm_provider_ids,
|
||
default_value=saved_llm_provider,
|
||
key="llm_provider_select",
|
||
format_func=lambda provider_id: llm_provider_labels[provider_id],
|
||
)
|
||
# 配置表单和 Provider 说明并排展示，减少长说明在窄列中的换行，
|
||
# 同时充分利用基础设置面板的横向空间。
|
||
llm_form_panel, llm_help_panel = st.columns(
|
||
[0.9, 1.1],
|
||
gap="large",
|
||
vertical_alignment="top",
|
||
)
|
||
llm_helper = llm_help_panel.container()
|
||
config.app["llm_provider"] = llm_provider
|
||
llm_provider_spec = get_llm_provider(llm_provider)
|
||
if llm_provider_spec is None:
|
||
# 正常情况下下拉选项全部来自 Registry，不会进入该分支；保留
|
||
# 明确错误用于诊断损坏的 session state 或后续接入遗漏。
|
||
raise RuntimeError(f"unsupported llm provider: {llm_provider}")
|
||
|
||
llm_api_key = config.app.get(llm_provider_spec.config_key("api_key"), "")
|
||
llm_base_url = (
|
||
config.app.get(llm_provider_spec.config_key("base_url"), "")
|
||
or llm_provider_spec.default_base_url
|
||
)
|
||
llm_default_base_url = llm_provider_spec.default_base_url
|
||
llm_model_name = llm_provider_spec.resolve_model_name(
|
||
config.app.get(llm_provider_spec.config_key("model_name"), "")
|
||
)
|
||
|
||
provider_tip_context = {}
|
||
if llm_provider == "ollama":
|
||
llm_default_base_url = config.get_default_ollama_base_url()
|
||
if not llm_base_url:
|
||
llm_base_url = llm_default_base_url
|
||
docker_hint = ""
|
||
if config.is_running_in_container():
|
||
docker_hint = tr_optional(
|
||
"llm_provider_tips.ollama.docker_hint",
|
||
fallback_language="en",
|
||
)
|
||
provider_tip_context["docker_hint"] = docker_hint
|
||
|
||
tips = get_llm_provider_tips(llm_provider, **provider_tip_context)
|
||
if tips:
|
||
with llm_helper:
|
||
st.info(tips)
|
||
|
||
st_llm_api_key = llm_api_key
|
||
if llm_provider_spec.show_api_key:
|
||
st_llm_api_key = llm_form_panel.text_input(
|
||
tr("API Key"),
|
||
value=llm_api_key,
|
||
type="password",
|
||
key=f"{llm_provider}_api_key_input",
|
||
)
|
||
|
||
st_llm_base_url = llm_base_url
|
||
if llm_provider_spec.show_base_url:
|
||
st_llm_base_url = llm_form_panel.text_input(
|
||
tr("Base Url"),
|
||
value=llm_base_url,
|
||
key=f"{llm_provider}_base_url_input",
|
||
)
|
||
st_llm_model_name = ""
|
||
if llm_provider == "groq":
|
||
effective_api_key = st_llm_api_key or llm_api_key
|
||
effective_base_url = st_llm_base_url or llm_base_url
|
||
groq_models = get_groq_model_ids(
|
||
api_key=effective_api_key,
|
||
base_url=effective_base_url,
|
||
)
|
||
|
||
if groq_models:
|
||
selected_index = 0
|
||
if llm_model_name in groq_models:
|
||
selected_index = groq_models.index(llm_model_name)
|
||
|
||
st_llm_model_name = llm_form_panel.selectbox(
|
||
tr("Model Name"),
|
||
options=groq_models,
|
||
index=selected_index,
|
||
key="groq_model_name_select",
|
||
)
|
||
else:
|
||
st_llm_model_name = llm_form_panel.text_input(
|
||
tr("Model Name"),
|
||
value=llm_model_name,
|
||
key="groq_model_name_input",
|
||
)
|
||
if effective_api_key:
|
||
llm_form_panel.caption(tr("Groq Model List Load Failed"))
|
||
else:
|
||
llm_form_panel.caption(
|
||
tr("Groq API Key Required for Model List")
|
||
)
|
||
else:
|
||
st_llm_model_name = llm_form_panel.text_input(
|
||
tr("Model Name"),
|
||
value=llm_model_name,
|
||
key=f"{llm_provider}_model_name_input",
|
||
)
|
||
# 输入框展示 Registry 默认值，但配置只保存真实的用户覆盖值。
|
||
# 这样默认模型、Base URL 更新后，未自定义的用户能够自动跟随。
|
||
config.app[llm_provider_spec.config_key("api_key")] = st_llm_api_key
|
||
config.app[llm_provider_spec.config_key("base_url")] = (
|
||
normalize_provider_override(
|
||
st_llm_base_url,
|
||
llm_default_base_url,
|
||
)
|
||
)
|
||
config.app[llm_provider_spec.config_key("model_name")] = (
|
||
normalize_provider_override(
|
||
st_llm_model_name,
|
||
llm_provider_spec.default_model,
|
||
)
|
||
)
|
||
|
||
# Provider 专用字段也由 Registry 声明。例如 Cloudflare AI Gateway
|
||
# 需要 Account ID；以后新增类似字段时无需再在 Main.py 增加判断。
|
||
for field in llm_provider_spec.extra_fields:
|
||
field_config_key = llm_provider_spec.config_key(field.config_suffix)
|
||
field_value = llm_form_panel.text_input(
|
||
tr(field.label_key),
|
||
value=(config.app.get(field_config_key, "") or field.default_value),
|
||
type="password" if field.secret else "default",
|
||
key=f"{llm_provider}_{field.config_suffix}_input",
|
||
)
|
||
config.app[field_config_key] = normalize_provider_override(
|
||
field_value,
|
||
field.default_value,
|
||
)
|
||
|
||
if llm_form_panel.button(
|
||
tr("Test LLM Connection"),
|
||
key="test_llm_connection_button",
|
||
use_container_width=True,
|
||
type="secondary",
|
||
icon=":material/network_check:",
|
||
):
|
||
with llm_form_panel.spinner(tr("Testing LLM Connection")):
|
||
with config.runtime_config_lock():
|
||
connection_ok, connection_error, connection_elapsed = (
|
||
llm.test_connection()
|
||
)
|
||
|
||
if connection_ok:
|
||
llm_form_panel.success(
|
||
tr("LLM Connection Test Succeeded").format(
|
||
provider=llm_provider_labels[llm_provider],
|
||
model=st_llm_model_name or "-",
|
||
elapsed=f"{connection_elapsed:.2f}",
|
||
)
|
||
)
|
||
else:
|
||
llm_form_panel.error(
|
||
tr("LLM Connection Test Failed").format(error=connection_error)
|
||
)
|
||
|
||
# 右侧面板 - API 密钥设置
|
||
with right_config_panel:
|
||
pexels_api_key = _get_material_api_keys("pexels_api_keys")
|
||
pexels_api_key = st.text_input(
|
||
tr("Pexels API Key"),
|
||
value=pexels_api_key,
|
||
type="password",
|
||
key="pexels_api_keys_input",
|
||
)
|
||
_save_material_api_keys("pexels_api_keys", pexels_api_key)
|
||
|
||
pixabay_api_key = _get_material_api_keys("pixabay_api_keys")
|
||
pixabay_api_key = st.text_input(
|
||
tr("Pixabay API Key"),
|
||
value=pixabay_api_key,
|
||
type="password",
|
||
key="pixabay_api_keys_input",
|
||
)
|
||
_save_material_api_keys("pixabay_api_keys", pixabay_api_key)
|
||
|
||
coverr_api_key = _get_material_api_keys("coverr_api_keys")
|
||
coverr_api_key = st.text_input(
|
||
tr("Coverr API Key"),
|
||
value=coverr_api_key,
|
||
type="password",
|
||
key="coverr_api_keys_input",
|
||
)
|
||
_save_material_api_keys("coverr_api_keys", coverr_api_key)
|
||
|
||
config.save_config()
|
||
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# 主生成表单：文案、视频、音频与字幕面板
|
||
# -----------------------------------------------------------------------------
|
||
|
||
|
||
def _render_script_settings(panel, params):
|
||
"""渲染文案设置并更新生成参数。"""
|
||
with panel:
|
||
with st.container(border=True):
|
||
st.write(tr("Video Script Settings"))
|
||
params.video_subject = st.text_input(
|
||
tr("Video Subject"),
|
||
placeholder=tr("Video Subject Placeholder"),
|
||
key="video_subject",
|
||
).strip()
|
||
|
||
video_languages = [
|
||
(tr("Auto Detect"), ""),
|
||
]
|
||
for code in support_locales:
|
||
video_languages.append((code, code))
|
||
|
||
selected_language_code = stable_selectbox(
|
||
tr("Script Language"),
|
||
options=[value for _, value in video_languages],
|
||
default_value="",
|
||
key="script_language_select",
|
||
format_func=lambda value: dict(
|
||
(v, label) for label, v in video_languages
|
||
)[value],
|
||
)
|
||
params.video_language = selected_language_code
|
||
|
||
# 使用带 key 的局部容器限定折叠入口样式，保持 expander 的原生交互，
|
||
# 同时避免样式误伤页面顶部的“基础设置”等其他折叠区域。
|
||
with st.container(key="advanced_settings_script"):
|
||
with st.expander(tr("Advanced Script Settings"), expanded=False):
|
||
st.session_state.setdefault("paragraph_number_input", 1)
|
||
params.paragraph_number = st.slider(
|
||
tr("Script Paragraph Number"),
|
||
min_value=llm.MIN_SCRIPT_PARAGRAPH_NUMBER,
|
||
max_value=llm.MAX_SCRIPT_PARAGRAPH_NUMBER,
|
||
key="paragraph_number_input",
|
||
)
|
||
params.video_script_prompt = st.text_area(
|
||
tr("Custom Script Requirements"),
|
||
height=100,
|
||
max_chars=llm.MAX_SCRIPT_PROMPT_LENGTH,
|
||
placeholder=tr("Custom Script Requirements Placeholder"),
|
||
key="video_script_prompt",
|
||
).strip()
|
||
|
||
system_prompt = st.text_area(
|
||
tr("Custom System Prompt"),
|
||
height=240,
|
||
max_chars=llm.MAX_SCRIPT_SYSTEM_PROMPT_LENGTH,
|
||
key="custom_system_prompt",
|
||
).strip()
|
||
# 默认内容由服务层统一维护。界面虽然直接展示默认提示词，但只有
|
||
# 用户实际修改后才随任务传递，避免历史任务固化旧版本默认规则。
|
||
params.custom_system_prompt = (
|
||
""
|
||
if system_prompt == llm.DEFAULT_SCRIPT_SYSTEM_PROMPT.strip()
|
||
else system_prompt
|
||
)
|
||
|
||
restore_prompt_col, preview_prompt_col = st.columns(2)
|
||
if restore_prompt_col.button(
|
||
tr("Restore Default System Prompt"),
|
||
key="restore_default_system_prompt",
|
||
icon=":material/restart_alt:",
|
||
on_click=reset_script_system_prompt,
|
||
use_container_width=True,
|
||
):
|
||
st.toast(tr("Default System Prompt Restored"))
|
||
if preview_prompt_col.button(
|
||
tr("Preview Final Prompt"),
|
||
key="preview_final_script_prompt",
|
||
icon=":material/preview:",
|
||
use_container_width=True,
|
||
):
|
||
render_script_prompt_preview(
|
||
llm.build_script_prompt(
|
||
video_subject=params.video_subject,
|
||
language=params.video_language,
|
||
paragraph_number=params.paragraph_number,
|
||
video_script_prompt=params.video_script_prompt,
|
||
custom_system_prompt=params.custom_system_prompt,
|
||
)
|
||
)
|
||
|
||
if st.button(
|
||
tr("Generate Video Script and Keywords"),
|
||
key="auto_generate_script",
|
||
use_container_width=True,
|
||
type="secondary",
|
||
icon=":material/auto_awesome:",
|
||
):
|
||
if not params.video_subject:
|
||
# 视频主题是脚本生成的必要输入，提前拦截可以避免无意义的模型调用。
|
||
st.toast(tr("Please Enter the Video Subject First"))
|
||
st.warning(tr("Please Enter the Video Subject First"))
|
||
else:
|
||
with st.spinner(tr("Generating Video Script and Keywords")):
|
||
with config.runtime_config_lock():
|
||
script = llm.generate_script(
|
||
video_subject=params.video_subject,
|
||
language=params.video_language,
|
||
paragraph_number=params.paragraph_number,
|
||
video_script_prompt=params.video_script_prompt,
|
||
custom_system_prompt=params.custom_system_prompt,
|
||
)
|
||
terms = llm.generate_terms(
|
||
params.video_subject,
|
||
script,
|
||
amount=8 if params.match_materials_to_script else 5,
|
||
match_script_order=params.match_materials_to_script,
|
||
)
|
||
if "Error: " in script:
|
||
st.error(tr(script))
|
||
elif "Error: " in terms:
|
||
st.error(tr(terms))
|
||
else:
|
||
st.session_state["video_script"] = script
|
||
st.session_state["video_terms"] = ", ".join(terms)
|
||
params.video_script = st.text_area(
|
||
tr("Video Script"),
|
||
help=tr("Video Script Help"),
|
||
height=180,
|
||
key="video_script",
|
||
)
|
||
if st.button(
|
||
tr("Generate Video Keywords"),
|
||
key="auto_generate_terms",
|
||
use_container_width=True,
|
||
type="secondary",
|
||
icon=":material/auto_awesome:",
|
||
):
|
||
if not params.video_script:
|
||
# 视频关键词需要基于文案提取，文案为空时提前提示并跳过模型调用。
|
||
st.toast(tr("Please Enter the Video Subject"))
|
||
st.warning(tr("Please Enter the Video Subject"))
|
||
else:
|
||
with st.spinner(tr("Generating Video Keywords")):
|
||
with config.runtime_config_lock():
|
||
terms = llm.generate_terms(
|
||
params.video_subject,
|
||
params.video_script,
|
||
amount=8 if params.match_materials_to_script else 5,
|
||
match_script_order=params.match_materials_to_script,
|
||
)
|
||
if "Error: " in terms:
|
||
st.error(tr(terms))
|
||
else:
|
||
st.session_state["video_terms"] = ", ".join(terms)
|
||
|
||
params.video_terms = st.text_area(
|
||
tr("Video Keywords"),
|
||
help=tr("Video Keywords Help"),
|
||
key="video_terms",
|
||
)
|
||
|
||
|
||
def _render_video_settings(panel, params):
|
||
"""渲染视频设置并返回本次选择的本地素材。"""
|
||
uploaded_files = []
|
||
with panel:
|
||
with st.container(border=True):
|
||
st.write(tr("Video Settings"))
|
||
video_concat_modes = [
|
||
(tr("Sequential"), "sequential"),
|
||
(tr("Random"), "random"),
|
||
]
|
||
video_sources = [
|
||
(tr("Pexels"), "pexels"),
|
||
(tr("Pixabay"), "pixabay"),
|
||
(tr("Coverr"), "coverr"),
|
||
(tr("Local file"), "local"),
|
||
]
|
||
|
||
saved_video_source_name = config.app.get("video_source", "pexels")
|
||
|
||
params.video_source = stable_selectbox(
|
||
tr("Video Source"),
|
||
options=[value for _, value in video_sources],
|
||
default_value=saved_video_source_name,
|
||
key="video_source_select",
|
||
format_func=lambda value: dict(
|
||
(v, label) for label, v in video_sources
|
||
)[value],
|
||
)
|
||
config.app["video_source"] = params.video_source
|
||
|
||
if params.video_source == "local":
|
||
# Streamlit 的文件类型校验对扩展名大小写敏感，这里同时放行大小写两种形式。
|
||
local_file_types = sorted(
|
||
extension.removeprefix(".")
|
||
for extension in LOCAL_MATERIAL_EXTENSIONS
|
||
)
|
||
uploaded_files = st.file_uploader(
|
||
tr("Upload Local Files"),
|
||
type=local_file_types
|
||
+ [file_type.upper() for file_type in local_file_types],
|
||
accept_multiple_files=True,
|
||
key="local_video_materials_uploader",
|
||
)
|
||
|
||
# 文案顺序匹配会从关键词生成到最终合成全程保持叙事顺序，因此开启时
|
||
# 顺序拼接是唯一符合实际执行逻辑的选项。同步控件值可避免界面仍显示
|
||
# “随机拼接”，同时保留用户原选择，关闭后自动恢复。
|
||
sync_script_order_concat_mode()
|
||
selected_concat_mode = stable_selectbox(
|
||
tr("Video Concat Mode"),
|
||
options=[value for _, value in video_concat_modes],
|
||
default_value=VideoConcatMode.random.value,
|
||
key="video_concat_mode_select",
|
||
format_func=lambda value: dict(
|
||
(v, label) for label, v in video_concat_modes
|
||
)[value],
|
||
disabled=bool(st.session_state.get("match_materials_to_script", False)),
|
||
)
|
||
params.video_concat_mode = VideoConcatMode(selected_concat_mode)
|
||
|
||
params.match_materials_to_script = st.checkbox(
|
||
tr("Match Materials to Script Order"),
|
||
help=tr("Match Materials to Script Order Help"),
|
||
key="match_materials_to_script",
|
||
on_change=sync_script_order_concat_mode,
|
||
)
|
||
config.app["match_materials_to_script"] = params.match_materials_to_script
|
||
|
||
# 视频转场模式
|
||
video_transition_modes = [
|
||
(tr("None"), VideoTransitionMode.none.value),
|
||
(tr("Shuffle"), VideoTransitionMode.shuffle.value),
|
||
(tr("FadeIn"), VideoTransitionMode.fade_in.value),
|
||
(tr("FadeOut"), VideoTransitionMode.fade_out.value),
|
||
(tr("SlideIn"), VideoTransitionMode.slide_in.value),
|
||
(tr("SlideOut"), VideoTransitionMode.slide_out.value),
|
||
(tr("ZoomIn"), VideoTransitionMode.zoom_in.value),
|
||
(tr("ZoomOut"), VideoTransitionMode.zoom_out.value),
|
||
]
|
||
selected_transition_mode = stable_selectbox(
|
||
tr("Video Transition Mode"),
|
||
options=[value for _, value in video_transition_modes],
|
||
default_value=VideoTransitionMode.none.value,
|
||
key="video_transition_mode_select",
|
||
format_func=lambda value: dict(
|
||
(v, label) for label, v in video_transition_modes
|
||
)[value],
|
||
)
|
||
params.video_transition_mode = VideoTransitionMode(selected_transition_mode)
|
||
|
||
video_aspect_ratios = [
|
||
(tr("Portrait"), VideoAspect.portrait.value),
|
||
(tr("Landscape"), VideoAspect.landscape.value),
|
||
]
|
||
# Coverr 库 99% 是 16:9 横屏,默认竖屏会让画面被大量黑边包围。
|
||
# 用 source-specific widget key 让每个 source 各自记忆 aspect 选择:
|
||
# - 首次切到 coverr → 默认 Landscape(index=1)
|
||
# - 其他 source 沿用 Portrait(index=0)
|
||
# - 用户在某 source 下手动改过 aspect,session_state 会记住,
|
||
# 下次回到同一 source 时尊重用户选择,不会再被强制覆盖。
|
||
default_aspect_index = 1 if params.video_source == "coverr" else 0
|
||
selected_aspect_ratio = stable_selectbox(
|
||
tr("Video Ratio"),
|
||
options=[value for _, value in video_aspect_ratios],
|
||
default_value=video_aspect_ratios[default_aspect_index][1],
|
||
key=f"video_aspect_for_{params.video_source}",
|
||
format_func=lambda value: dict(
|
||
(v, label) for label, v in video_aspect_ratios
|
||
)[value],
|
||
)
|
||
params.video_aspect = VideoAspect(selected_aspect_ratio)
|
||
|
||
params.video_clip_duration = stable_selectbox(
|
||
tr("Clip Duration"),
|
||
options=[2, 3, 4, 5, 6, 7, 8, 9, 10],
|
||
default_value=3,
|
||
key="video_clip_duration_select",
|
||
help=tr("Clip Duration Help"),
|
||
)
|
||
params.video_count = stable_selectbox(
|
||
tr("Number of Videos Generated Simultaneously"),
|
||
options=[1, 2, 3, 4, 5],
|
||
default_value=1,
|
||
key="video_count_select",
|
||
)
|
||
|
||
video_codec_options = [
|
||
(tr("Default Video Encoder"), DEFAULT_VIDEO_CODEC_OPTION),
|
||
("libx264 (CPU)", "libx264"),
|
||
("NVIDIA NVENC (h264_nvenc)", "h264_nvenc"),
|
||
("AMD AMF (h264_amf)", "h264_amf"),
|
||
("Intel QSV (h264_qsv)", "h264_qsv"),
|
||
("Windows MediaFoundation (h264_mf)", "h264_mf"),
|
||
("macOS VideoToolbox (h264_videotoolbox)", "h264_videotoolbox"),
|
||
]
|
||
saved_video_codec = config.app.get(
|
||
"video_codec", DEFAULT_VIDEO_CODEC_OPTION
|
||
)
|
||
saved_video_codec_values = [item[1] for item in video_codec_options]
|
||
if saved_video_codec not in saved_video_codec_values:
|
||
# 旧版本或手工配置可能留下无效值。UI 回到“默认”而不是替用户
|
||
# 固定某个编码器，后端仍会按稳定策略解析为 libx264。
|
||
saved_video_codec = DEFAULT_VIDEO_CODEC_OPTION
|
||
selected_video_codec = stable_selectbox(
|
||
tr("Video Encoder"),
|
||
options=saved_video_codec_values,
|
||
default_value=saved_video_codec,
|
||
key="video_encoder_select",
|
||
format_func=lambda value: dict(
|
||
(v, label) for label, v in video_codec_options
|
||
)[value],
|
||
help=tr("Video Encoder Help"),
|
||
)
|
||
if selected_video_codec == DEFAULT_VIDEO_CODEC_OPTION:
|
||
# 默认模式不持久化具体编码器，让配置表达“跟随项目默认值”。
|
||
config.app.pop("video_codec", None)
|
||
else:
|
||
config.app["video_codec"] = selected_video_codec
|
||
return uploaded_files
|
||
|
||
|
||
def _render_voice_preview(params, friendly_names, selected_tts_server, voice_name):
|
||
"""使用当前音色、音量和语速生成一段试听音频。"""
|
||
if not friendly_names or not st.button(
|
||
tr("Play Voice"),
|
||
key="play_voice_button",
|
||
icon=":material/graphic_eq:",
|
||
):
|
||
return
|
||
|
||
if selected_tts_server == "chatterbox":
|
||
_sync_chatterbox_config_from_session_state()
|
||
play_content = params.video_subject or params.video_script
|
||
if not play_content:
|
||
# ElevenLabs 音色缺少明确语言字段时，根据展示名称中的越南语字符
|
||
# 选择试听文案，避免用不匹配的语言判断音色效果。
|
||
if voice.is_elevenlabs_voice(voice_name):
|
||
parts = voice_name.split(":", 2)
|
||
display = parts[2] if len(parts) >= 3 else ""
|
||
vietnamese_chars = set("àáâãèéêìíòóôõùúýăđơưÀÁÂÃÈÉÊÌÍÒÓÔÕÙÚÝĂĐƠƯ")
|
||
if any(char in vietnamese_chars for char in display):
|
||
play_content = "Xin chào, đây là đoạn âm thanh thử nghiệm giọng nói."
|
||
else:
|
||
play_content = tr("Voice Example")
|
||
else:
|
||
play_content = tr("Voice Example")
|
||
|
||
with st.spinner(tr("Synthesizing Voice")):
|
||
temp_dir = utils.storage_dir("temp", create=True)
|
||
audio_file = os.path.join(temp_dir, f"tmp-voice-{str(uuid4())}.mp3")
|
||
logger.info(
|
||
"generating voice preview: "
|
||
f"voice={voice_name}, rate={params.voice_rate}, "
|
||
f"volume={params.voice_volume}"
|
||
)
|
||
with config.runtime_config_lock():
|
||
sub_maker = voice.tts(
|
||
text=play_content,
|
||
voice_name=voice_name,
|
||
voice_rate=params.voice_rate,
|
||
voice_file=audio_file,
|
||
voice_volume=params.voice_volume,
|
||
)
|
||
# 首次试听失败后仍在同一个配置锁中重试，避免两个请求之间
|
||
# 被其它标签页切换 TTS Provider 或 API Key。
|
||
if not sub_maker:
|
||
play_content = "This is an example voice."
|
||
sub_maker = voice.tts(
|
||
text=play_content,
|
||
voice_name=voice_name,
|
||
voice_rate=params.voice_rate,
|
||
voice_file=audio_file,
|
||
voice_volume=params.voice_volume,
|
||
)
|
||
|
||
if not sub_maker or not os.path.exists(audio_file):
|
||
return
|
||
try:
|
||
with open(audio_file, "rb") as file:
|
||
audio_bytes = file.read()
|
||
if audio_bytes:
|
||
st.audio(
|
||
audio_bytes,
|
||
format=_detect_audio_mime(audio_file, audio_bytes),
|
||
)
|
||
else:
|
||
logger.error(f"voice preview audio file is empty: {audio_file}")
|
||
finally:
|
||
# 试听文件只服务当前 rerun，渲染为内存数据后立即清理。
|
||
if os.path.exists(audio_file):
|
||
os.remove(audio_file)
|
||
|
||
|
||
def _render_background_music_settings(params):
|
||
"""渲染背景音乐来源与音量设置。"""
|
||
st.divider()
|
||
bgm_options = [
|
||
(tr("No Background Music"), ""),
|
||
(tr("Random Background Music"), "random"),
|
||
(tr("Custom Background Music"), "custom"),
|
||
]
|
||
selected_bgm_type = stable_selectbox(
|
||
tr("Background Music Source"),
|
||
options=[value for _, value in bgm_options],
|
||
default_value="random",
|
||
key="bgm_type_select",
|
||
format_func=lambda value: dict((v, label) for label, v in bgm_options)[value],
|
||
)
|
||
params.bgm_type = selected_bgm_type
|
||
|
||
if params.bgm_type == "custom":
|
||
custom_bgm_file = st.text_input(
|
||
tr("Custom Background Music File"),
|
||
key="custom_bgm_file_input",
|
||
)
|
||
if custom_bgm_file:
|
||
# 文件名由服务层映射到 resource/songs 后校验，UI 不接受任意路径。
|
||
params.bgm_file = custom_bgm_file.strip()
|
||
params.bgm_volume = stable_selectbox(
|
||
tr("Background Music Volume"),
|
||
options=[0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0],
|
||
default_value=0.2,
|
||
key="bgm_volume_select",
|
||
format_func=lambda value: f"{int(value * 100)}%",
|
||
disabled=not params.bgm_type,
|
||
)
|
||
|
||
|
||
def _render_audio_settings(panel, params):
|
||
"""渲染音频设置并返回上传音频与当前配音模式。"""
|
||
with panel:
|
||
with st.container(border=True):
|
||
st.write(tr("Audio Settings"))
|
||
|
||
# 配音方式是音频设置的一级状态，负责明确区分自动配音、用户上传和无配音。
|
||
# 旧配置没有 voice_mode 时，根据原 tts_server 的无配音哨兵保持兼容。
|
||
saved_tts_server = config.ui.get("tts_server", "azure-tts-v1")
|
||
saved_voice_mode = config.ui.get("voice_mode")
|
||
if saved_voice_mode not in {
|
||
VOICE_MODE_TTS,
|
||
VOICE_MODE_UPLOAD,
|
||
VOICE_MODE_NONE,
|
||
}:
|
||
saved_voice_mode = (
|
||
VOICE_MODE_NONE
|
||
if saved_tts_server == voice.NO_VOICE_NAME
|
||
else VOICE_MODE_TTS
|
||
)
|
||
voice_mode_options = [VOICE_MODE_TTS, VOICE_MODE_UPLOAD, VOICE_MODE_NONE]
|
||
voice_mode_labels = {
|
||
VOICE_MODE_TTS: tr("Automatic Voiceover"),
|
||
VOICE_MODE_UPLOAD: tr("Upload Voiceover"),
|
||
VOICE_MODE_NONE: tr("No Voiceover"),
|
||
}
|
||
voice_mode = stable_segmented_control(
|
||
tr("Voiceover Mode"),
|
||
options=voice_mode_options,
|
||
default_value=saved_voice_mode,
|
||
key="voice_mode_control",
|
||
format_func=lambda value: voice_mode_labels[value],
|
||
width="stretch",
|
||
)
|
||
config.ui["voice_mode"] = voice_mode
|
||
tts_mode_enabled = voice_mode == VOICE_MODE_TTS
|
||
|
||
# Provider 下拉只负责选择自动配音服务；无配音已经由上方模式控制，
|
||
# 不再作为 TTS Provider 混入列表，避免两个入口表达同一状态。
|
||
tts_servers = [
|
||
("azure-tts-v1", "Azure TTS V1"),
|
||
("azure-tts-v2", "Azure TTS V2"),
|
||
("siliconflow", "SiliconFlow TTS"),
|
||
("gemini-tts", "Google Gemini TTS"),
|
||
("mimo-tts", "Xiaomi MiMo TTS"),
|
||
("elevenlabs", "ElevenLabs TTS"),
|
||
("chatterbox", "Chatterbox TTS"),
|
||
]
|
||
|
||
tts_server_values = [server_value for server_value, _ in tts_servers]
|
||
if saved_tts_server not in tts_server_values:
|
||
saved_tts_server = "azure-tts-v1"
|
||
|
||
if tts_mode_enabled:
|
||
selected_tts_server = stable_selectbox(
|
||
tr("Voiceover Service"),
|
||
options=tts_server_values,
|
||
default_value=saved_tts_server,
|
||
key="tts_server_select",
|
||
format_func=lambda value: dict(
|
||
(v, label) for v, label in tts_servers
|
||
)[value],
|
||
)
|
||
else:
|
||
# 非自动配音模式不渲染 TTS 控件，但保留上次选择，切回后可以继续使用。
|
||
selected_tts_server = saved_tts_server
|
||
|
||
config.ui["tts_server"] = selected_tts_server
|
||
|
||
# 服务说明紧跟 Provider 选择，先告诉用户需要准备什么，再进入音色和
|
||
# 凭证配置。没有说明的 Provider 不渲染空提示块。
|
||
if tts_mode_enabled:
|
||
provider_tips = get_tts_provider_tips(selected_tts_server)
|
||
if provider_tips:
|
||
st.info(provider_tips)
|
||
|
||
# 根据选择的TTS服务器获取声音列表
|
||
filtered_voices = []
|
||
saved_voice_name = config.ui.get("voice_name", "")
|
||
|
||
if not tts_mode_enabled:
|
||
# 上传音频和无配音模式不加载远程音色，减少无意义的网络请求和界面噪音。
|
||
filtered_voices = []
|
||
elif selected_tts_server == "siliconflow":
|
||
# 获取硅基流动的声音列表
|
||
filtered_voices = voice.get_siliconflow_voices()
|
||
elif selected_tts_server == "gemini-tts":
|
||
# 获取Gemini TTS的声音列表
|
||
filtered_voices = voice.get_gemini_voices()
|
||
elif selected_tts_server == "mimo-tts":
|
||
# 获取 Xiaomi MiMo TTS 的预置音色列表
|
||
filtered_voices = voice.get_mimo_voices()
|
||
elif selected_tts_server == "elevenlabs":
|
||
# Read from session_state first so the API key is available before
|
||
# the Play Voice button runs (which is earlier in the script than
|
||
# the API key text_input widget).
|
||
saved_elevenlabs_api_key = st.session_state.get(
|
||
"elevenlabs_api_key_input",
|
||
config.elevenlabs.get("api_key", ""),
|
||
)
|
||
if saved_elevenlabs_api_key:
|
||
config.elevenlabs["api_key"] = saved_elevenlabs_api_key
|
||
cache_key = f"elevenlabs_voices_{saved_elevenlabs_api_key}"
|
||
if cache_key not in st.session_state:
|
||
st.session_state[cache_key] = voice.get_elevenlabs_voices(
|
||
saved_elevenlabs_api_key
|
||
)
|
||
filtered_voices = st.session_state[cache_key]
|
||
elif selected_tts_server == "chatterbox":
|
||
# 自托管 Chatterbox 服务的预置音色（来自 [chatterbox] voices 配置）
|
||
_sync_chatterbox_config_from_session_state()
|
||
filtered_voices = voice.get_chatterbox_voices()
|
||
else:
|
||
# 获取Azure的声音列表
|
||
all_voices = voice.get_all_azure_voices(filter_locals=None)
|
||
|
||
# 根据选择的TTS服务器筛选声音
|
||
for v in all_voices:
|
||
if selected_tts_server == "azure-tts-v2":
|
||
# V2版本的声音名称中包含"v2"
|
||
if "V2" in v:
|
||
filtered_voices.append(v)
|
||
else:
|
||
# V1版本的声音名称中不包含"v2"
|
||
if "V2" not in v:
|
||
filtered_voices.append(v)
|
||
|
||
def _friendly(v):
|
||
if voice.is_no_voice(v):
|
||
return tr("No Voice Selected")
|
||
if voice.is_elevenlabs_voice(v):
|
||
parts = v.split(":", 2)
|
||
return parts[2] if len(parts) >= 3 else v
|
||
if voice.is_chatterbox_voice(v):
|
||
name = v.split(":", 1)[1] if ":" in v else v
|
||
return name.replace("-Female", "").replace("-Male", "")
|
||
return (
|
||
v.replace("Female", tr("Female"))
|
||
.replace("Male", tr("Male"))
|
||
.replace("Neural", "")
|
||
)
|
||
|
||
friendly_names = {v: _friendly(v) for v in filtered_voices}
|
||
|
||
saved_voice_name_index = 0
|
||
|
||
# 检查保存的声音是否在当前筛选的声音列表中
|
||
if saved_voice_name in friendly_names:
|
||
saved_voice_name_index = list(friendly_names.keys()).index(
|
||
saved_voice_name
|
||
)
|
||
else:
|
||
# 如果不在，则根据当前UI语言选择一个默认声音
|
||
for i, v in enumerate(filtered_voices):
|
||
if v.lower().startswith(st.session_state["ui_language"].lower()):
|
||
saved_voice_name_index = i
|
||
break
|
||
|
||
# 如果没有找到匹配的声音，使用第一个声音
|
||
if saved_voice_name_index >= len(friendly_names) and friendly_names:
|
||
saved_voice_name_index = 0
|
||
|
||
# 确保有声音可选
|
||
if tts_mode_enabled and friendly_names:
|
||
voice_name = stable_selectbox(
|
||
tr("Voiceover Voice"),
|
||
options=list(friendly_names.keys()),
|
||
default_value=list(friendly_names.keys())[saved_voice_name_index],
|
||
key=f"speech_synthesis_select_{selected_tts_server}",
|
||
format_func=lambda value: friendly_names[value],
|
||
)
|
||
|
||
params.voice_name = voice_name
|
||
if not voice.is_no_voice(voice_name):
|
||
# 占位 sentinel 仅用于非自动模式的禁用展示，不覆盖用户上一次
|
||
# 真正选择的音色，切回自动配音后可以恢复原设置。
|
||
config.ui["voice_name"] = voice_name
|
||
elif tts_mode_enabled:
|
||
# 如果没有声音可选，显示提示信息
|
||
st.warning(
|
||
tr(
|
||
"No voices available for the selected TTS server. Please select another server."
|
||
)
|
||
)
|
||
voice_name = ""
|
||
params.voice_name = ""
|
||
config.ui["voice_name"] = ""
|
||
else:
|
||
# 非自动配音模式不显示音色控件，只复用保存值维持参数结构稳定。
|
||
voice_name = saved_voice_name or voice.NO_VOICE_NAME
|
||
params.voice_name = voice_name
|
||
|
||
# 当选择V2版本或者声音是V2声音时，显示服务区域和API key输入框
|
||
if tts_mode_enabled and (
|
||
selected_tts_server == "azure-tts-v2"
|
||
or (voice_name and voice.is_azure_v2_voice(voice_name))
|
||
):
|
||
saved_azure_speech_region = config.azure.get("speech_region", "")
|
||
saved_azure_speech_key = config.azure.get("speech_key", "")
|
||
azure_speech_region = st.text_input(
|
||
tr("Speech Region"),
|
||
value=saved_azure_speech_region,
|
||
key="azure_speech_region_input",
|
||
)
|
||
azure_speech_key = st.text_input(
|
||
tr("Speech Key"),
|
||
value=saved_azure_speech_key,
|
||
type="password",
|
||
key="azure_speech_key_input",
|
||
)
|
||
config.azure["speech_region"] = azure_speech_region
|
||
config.azure["speech_key"] = azure_speech_key
|
||
|
||
if tts_mode_enabled and selected_tts_server == "gemini-tts":
|
||
# Gemini TTS 与 Gemini LLM 共用同一份密钥；在音频面板提供直接入口，
|
||
# 用户无需先切换 LLM Provider 才能完成语音配置。
|
||
gemini_tts_api_key = st.text_input(
|
||
f"Google Gemini {tr('API Key')}",
|
||
value=config.app.get("gemini_api_key", ""),
|
||
type="password",
|
||
key="gemini_tts_api_key_input",
|
||
)
|
||
config.app["gemini_api_key"] = gemini_tts_api_key
|
||
|
||
# 当选择硅基流动时，显示API key输入框和说明信息
|
||
if tts_mode_enabled and (
|
||
selected_tts_server == "siliconflow"
|
||
or (voice_name and voice.is_siliconflow_voice(voice_name))
|
||
):
|
||
saved_siliconflow_api_key = config.siliconflow.get("api_key", "")
|
||
|
||
siliconflow_api_key = st.text_input(
|
||
tr("SiliconFlow API Key"),
|
||
value=saved_siliconflow_api_key,
|
||
type="password",
|
||
key="siliconflow_api_key_input",
|
||
)
|
||
|
||
config.siliconflow["api_key"] = siliconflow_api_key
|
||
|
||
# 当选择 Xiaomi MiMo TTS 时，复用 MiMo LLM provider 的 API Key。
|
||
# 这样用户如果同时使用 MiMo 生成文案和语音，只需要维护一份密钥。
|
||
if tts_mode_enabled and (
|
||
selected_tts_server == "mimo-tts"
|
||
or (voice_name and voice.is_mimo_voice(voice_name))
|
||
):
|
||
saved_mimo_api_key = config.app.get("mimo_api_key", "")
|
||
|
||
mimo_api_key = st.text_input(
|
||
tr("MiMo API Key"),
|
||
value=saved_mimo_api_key,
|
||
type="password",
|
||
key="mimo_tts_api_key_input",
|
||
)
|
||
|
||
config.app["mimo_api_key"] = mimo_api_key
|
||
|
||
# ElevenLabs API key section
|
||
if tts_mode_enabled and (
|
||
selected_tts_server == "elevenlabs"
|
||
or (voice_name and voice.is_elevenlabs_voice(voice_name))
|
||
):
|
||
saved_elevenlabs_api_key = config.elevenlabs.get("api_key", "")
|
||
|
||
elevenlabs_api_key = st.text_input(
|
||
tr("ElevenLabs API Key"),
|
||
value=saved_elevenlabs_api_key,
|
||
type="password",
|
||
key="elevenlabs_api_key_input",
|
||
)
|
||
|
||
_elevenlabs_models = [
|
||
"eleven_multilingual_v2",
|
||
"eleven_flash_v2_5",
|
||
"eleven_v3",
|
||
]
|
||
saved_elevenlabs_model = config.elevenlabs.get(
|
||
"model_id", "eleven_multilingual_v2"
|
||
)
|
||
if saved_elevenlabs_model not in _elevenlabs_models:
|
||
saved_elevenlabs_model = "eleven_multilingual_v2"
|
||
elevenlabs_model = stable_selectbox(
|
||
tr("ElevenLabs Model"),
|
||
options=_elevenlabs_models,
|
||
default_value=saved_elevenlabs_model,
|
||
key="elevenlabs_model_select",
|
||
)
|
||
config.elevenlabs["model_id"] = elevenlabs_model
|
||
|
||
if elevenlabs_api_key != saved_elevenlabs_api_key:
|
||
for k in list(st.session_state.keys()):
|
||
if k.startswith("elevenlabs_voices_"):
|
||
del st.session_state[k]
|
||
|
||
config.elevenlabs["api_key"] = elevenlabs_api_key
|
||
|
||
# Chatterbox API settings section (self-hosted, OpenAI-compatible)
|
||
if tts_mode_enabled and (
|
||
selected_tts_server == "chatterbox"
|
||
or (voice_name and voice.is_chatterbox_voice(voice_name))
|
||
):
|
||
chatterbox_base_url = st.text_input(
|
||
tr("Chatterbox Base URL"),
|
||
value=config.chatterbox.get("base_url")
|
||
or DEFAULT_CHATTERBOX_BASE_URL,
|
||
key="chatterbox_base_url_input",
|
||
placeholder=tr("Chatterbox Base URL Placeholder"),
|
||
)
|
||
config.chatterbox["base_url"] = (chatterbox_base_url or "").strip()
|
||
|
||
chatterbox_api_key = st.text_input(
|
||
tr("Chatterbox API Key"),
|
||
value=config.chatterbox.get("api_key", ""),
|
||
type="password",
|
||
key="chatterbox_api_key_input",
|
||
)
|
||
config.chatterbox["api_key"] = chatterbox_api_key
|
||
|
||
chatterbox_model = st.text_input(
|
||
tr("Chatterbox Model"),
|
||
value=config.chatterbox.get("model_id") or DEFAULT_CHATTERBOX_MODEL,
|
||
key="chatterbox_model_input",
|
||
)
|
||
config.chatterbox["model_id"] = (
|
||
chatterbox_model or DEFAULT_CHATTERBOX_MODEL
|
||
).strip()
|
||
|
||
_saved_chatterbox_voices = (
|
||
_parse_chatterbox_voices(config.chatterbox.get("voices"))
|
||
or DEFAULT_CHATTERBOX_VOICES
|
||
)
|
||
if isinstance(_saved_chatterbox_voices, list):
|
||
_saved_chatterbox_voices = ", ".join(_saved_chatterbox_voices)
|
||
chatterbox_voices = st.text_input(
|
||
tr("Chatterbox Voices"),
|
||
value=str(_saved_chatterbox_voices or ""),
|
||
key="chatterbox_voices_input",
|
||
placeholder=tr("Chatterbox Voices Placeholder"),
|
||
)
|
||
config.chatterbox["voices"] = _parse_chatterbox_voices(
|
||
chatterbox_voices
|
||
)
|
||
|
||
# 三种模式只渲染当前任务真正需要的控件。自动配音可调音量和语速；
|
||
# 上传音频只需要文件和音量；无配音不再展示无效设置。
|
||
params.voice_name = (
|
||
voice.NO_VOICE_NAME if voice_mode == VOICE_MODE_NONE else voice_name
|
||
)
|
||
params.voice_volume = 1.0
|
||
params.voice_rate = 1.0
|
||
uploaded_audio_file = None
|
||
|
||
if tts_mode_enabled:
|
||
voice_control_cols = st.columns(2)
|
||
with voice_control_cols[0]:
|
||
params.voice_volume = stable_selectbox(
|
||
tr("Voiceover Volume"),
|
||
options=[0.6, 0.8, 1.0, 1.2, 1.5, 2.0, 3.0, 4.0, 5.0],
|
||
default_value=1.0,
|
||
key="voice_volume_select",
|
||
format_func=lambda value: f"{int(value * 100)}%",
|
||
help=tr("Voiceover Volume Help"),
|
||
)
|
||
|
||
with voice_control_cols[1]:
|
||
params.voice_rate = stable_selectbox(
|
||
tr("Voiceover Speed"),
|
||
options=[0.8, 0.9, 1.0, 1.1, 1.2, 1.3, 1.5, 1.8, 2.0],
|
||
default_value=1.0,
|
||
key="voice_rate_select",
|
||
format_func=lambda value: f"{value:.1f}×",
|
||
help=tr("Voiceover Speed Help"),
|
||
)
|
||
|
||
# 试听必须位于音量和语速控件之后，确保调用使用当前控件值。
|
||
_render_voice_preview(
|
||
params,
|
||
friendly_names,
|
||
selected_tts_server,
|
||
voice_name,
|
||
)
|
||
elif voice_mode == VOICE_MODE_UPLOAD:
|
||
custom_audio_file_types = sorted(
|
||
extension.removeprefix(".") for extension in CUSTOM_AUDIO_EXTENSIONS
|
||
)
|
||
uploaded_audio_file = st.file_uploader(
|
||
tr("Upload Voiceover File"),
|
||
type=custom_audio_file_types
|
||
+ [file_type.upper() for file_type in custom_audio_file_types],
|
||
accept_multiple_files=False,
|
||
key="custom_audio_file_uploader",
|
||
help=tr("Upload Voiceover File Help"),
|
||
)
|
||
params.voice_volume = stable_selectbox(
|
||
tr("Voiceover Volume"),
|
||
options=[0.6, 0.8, 1.0, 1.2, 1.5, 2.0, 3.0, 4.0, 5.0],
|
||
default_value=1.0,
|
||
key="voice_volume_select",
|
||
format_func=lambda value: f"{int(value * 100)}%",
|
||
help=tr("Voiceover Volume Help"),
|
||
)
|
||
if uploaded_audio_file:
|
||
st.audio(uploaded_audio_file, format="audio/mp3")
|
||
st.info(
|
||
tr(
|
||
"Custom audio will be used directly. TTS synthesis will be skipped for this task."
|
||
)
|
||
)
|
||
_render_background_music_settings(params)
|
||
return uploaded_audio_file, voice_mode
|
||
|
||
|
||
def _render_subtitle_settings(panel, params):
|
||
"""渲染字幕设置并更新生成参数。"""
|
||
with panel:
|
||
with st.container(border=True):
|
||
st.write(tr("Subtitle Settings"))
|
||
st.session_state.setdefault(
|
||
"subtitle_enabled_checkbox",
|
||
DEFAULT_SUBTITLE_SETTINGS["subtitle_enabled"],
|
||
)
|
||
params.subtitle_enabled = st.checkbox(
|
||
tr("Enable Subtitles"),
|
||
key="subtitle_enabled_checkbox",
|
||
)
|
||
subtitle_settings_disabled = not params.subtitle_enabled
|
||
font_names = get_all_fonts()
|
||
saved_font_name = config.ui.get(
|
||
"font_name", DEFAULT_SUBTITLE_SETTINGS["font_name"]
|
||
)
|
||
saved_font_name_index = 0
|
||
if saved_font_name in font_names:
|
||
saved_font_name_index = font_names.index(saved_font_name)
|
||
params.font_name = stable_selectbox(
|
||
tr("Font"),
|
||
options=font_names,
|
||
default_value=font_names[saved_font_name_index] if font_names else "",
|
||
key="font_name_select",
|
||
disabled=subtitle_settings_disabled,
|
||
)
|
||
config.ui["font_name"] = params.font_name
|
||
|
||
subtitle_positions = [
|
||
(tr("Top"), "top"),
|
||
(tr("Center"), "center"),
|
||
(tr("Bottom"), "bottom"),
|
||
(tr("Custom"), "custom"),
|
||
]
|
||
saved_subtitle_position = config.ui.get(
|
||
"subtitle_position", DEFAULT_SUBTITLE_SETTINGS["subtitle_position"]
|
||
)
|
||
saved_position_index = 2
|
||
for i, (_, pos_value) in enumerate(subtitle_positions):
|
||
if pos_value == saved_subtitle_position:
|
||
saved_position_index = i
|
||
break
|
||
selected_subtitle_position = stable_selectbox(
|
||
tr("Position"),
|
||
options=[value for _, value in subtitle_positions],
|
||
default_value=subtitle_positions[saved_position_index][1],
|
||
key="subtitle_position_select",
|
||
format_func=lambda value: dict(
|
||
(v, label) for label, v in subtitle_positions
|
||
)[value],
|
||
disabled=subtitle_settings_disabled,
|
||
)
|
||
params.subtitle_position = selected_subtitle_position
|
||
config.ui["subtitle_position"] = params.subtitle_position
|
||
|
||
if params.subtitle_position == "custom":
|
||
saved_custom_position = config.ui.get(
|
||
"custom_position", DEFAULT_SUBTITLE_SETTINGS["custom_position"]
|
||
)
|
||
st.session_state.setdefault(
|
||
"custom_position_input", str(saved_custom_position)
|
||
)
|
||
custom_position = st.text_input(
|
||
tr("Custom Position (% from top)"),
|
||
key="custom_position_input",
|
||
disabled=subtitle_settings_disabled,
|
||
)
|
||
try:
|
||
params.custom_position = float(custom_position)
|
||
if params.custom_position < 0 or params.custom_position > 100:
|
||
st.error(tr("Please enter a value between 0 and 100"))
|
||
else:
|
||
config.ui["custom_position"] = params.custom_position
|
||
except ValueError:
|
||
st.error(tr("Please enter a valid number"))
|
||
|
||
# 非中文语言的颜色标签通常比中文更长。为颜色选择器保留适当宽度，
|
||
# 避免标签换行，同时仍给字号滑块保留足够的可操作空间。
|
||
font_cols = st.columns([0.42, 0.58])
|
||
with font_cols[0]:
|
||
saved_text_fore_color = config.ui.get(
|
||
"text_fore_color", DEFAULT_SUBTITLE_SETTINGS["text_fore_color"]
|
||
)
|
||
st.session_state.setdefault("font_color_picker", saved_text_fore_color)
|
||
params.text_fore_color = st.color_picker(
|
||
tr("Font Color"),
|
||
key="font_color_picker",
|
||
disabled=subtitle_settings_disabled,
|
||
)
|
||
config.ui["text_fore_color"] = params.text_fore_color
|
||
|
||
with font_cols[1]:
|
||
saved_font_size = config.ui.get(
|
||
"font_size", DEFAULT_SUBTITLE_SETTINGS["font_size"]
|
||
)
|
||
st.session_state.setdefault("font_size_slider", saved_font_size)
|
||
params.font_size = st.slider(
|
||
tr("Font Size"),
|
||
30,
|
||
100,
|
||
key="font_size_slider",
|
||
disabled=subtitle_settings_disabled,
|
||
)
|
||
config.ui["font_size"] = params.font_size
|
||
|
||
stroke_cols = st.columns([0.42, 0.58])
|
||
with stroke_cols[0]:
|
||
st.session_state.setdefault(
|
||
"stroke_color_picker", DEFAULT_SUBTITLE_SETTINGS["stroke_color"]
|
||
)
|
||
params.stroke_color = st.color_picker(
|
||
tr("Stroke Color"),
|
||
key="stroke_color_picker",
|
||
disabled=subtitle_settings_disabled,
|
||
)
|
||
with stroke_cols[1]:
|
||
st.session_state.setdefault(
|
||
"stroke_width_slider", DEFAULT_SUBTITLE_SETTINGS["stroke_width"]
|
||
)
|
||
params.stroke_width = st.slider(
|
||
tr("Stroke Width"),
|
||
0.0,
|
||
10.0,
|
||
key="stroke_width_slider",
|
||
disabled=subtitle_settings_disabled,
|
||
)
|
||
|
||
# 背景开关的本地化名称普遍比颜色标签更长，因此让开关占据略多空间。
|
||
subtitle_bg_cols = st.columns([0.55, 0.45])
|
||
saved_subtitle_background_enabled = config.ui.get(
|
||
"subtitle_background_enabled",
|
||
DEFAULT_SUBTITLE_SETTINGS["subtitle_background_enabled"],
|
||
)
|
||
st.session_state.setdefault(
|
||
"subtitle_background_enabled_checkbox",
|
||
saved_subtitle_background_enabled,
|
||
)
|
||
with subtitle_bg_cols[0]:
|
||
subtitle_background_enabled = st.checkbox(
|
||
tr("Enable Subtitle Background"),
|
||
key="subtitle_background_enabled_checkbox",
|
||
disabled=subtitle_settings_disabled,
|
||
)
|
||
config.ui["subtitle_background_enabled"] = subtitle_background_enabled
|
||
|
||
# 背景颜色和圆角样式都从属于字幕背景开关。子控件始终保留在页面中，
|
||
# 父开关关闭时统一禁用，避免一个控件消失而另一个控件禁用造成布局跳动。
|
||
# 颜色值仍保存在 UI 配置中，重新启用背景后可以恢复用户之前的选择；
|
||
# 传给生成服务的参数则设为 False，确保关闭状态不会实际渲染背景。
|
||
saved_subtitle_background_color = config.ui.get(
|
||
"subtitle_background_color",
|
||
DEFAULT_SUBTITLE_SETTINGS["subtitle_background_color"],
|
||
)
|
||
st.session_state.setdefault(
|
||
"subtitle_background_color_picker",
|
||
saved_subtitle_background_color,
|
||
)
|
||
with subtitle_bg_cols[1]:
|
||
selected_subtitle_background_color = st.color_picker(
|
||
tr("Subtitle Background Color"),
|
||
key="subtitle_background_color_picker",
|
||
disabled=subtitle_settings_disabled
|
||
or not subtitle_background_enabled,
|
||
)
|
||
config.ui["subtitle_background_color"] = selected_subtitle_background_color
|
||
params.text_background_color = (
|
||
selected_subtitle_background_color
|
||
if subtitle_background_enabled
|
||
else False
|
||
)
|
||
|
||
saved_rounded_subtitle_background = config.ui.get(
|
||
"rounded_subtitle_background",
|
||
DEFAULT_SUBTITLE_SETTINGS["rounded_subtitle_background"],
|
||
)
|
||
# 背景关闭时，圆角背景没有可渲染的底色。这里禁用控件但保留原配置，
|
||
# 用户下次重新开启字幕背景后，可以继续使用之前保存的圆角偏好。
|
||
rounded_background_disabled = (
|
||
subtitle_settings_disabled or not subtitle_background_enabled
|
||
)
|
||
st.session_state.setdefault(
|
||
"rounded_subtitle_background_checkbox",
|
||
saved_rounded_subtitle_background,
|
||
)
|
||
selected_rounded_subtitle_background = st.checkbox(
|
||
tr("Rounded Subtitle Background"),
|
||
help=tr("Rounded Subtitle Background Help"),
|
||
disabled=rounded_background_disabled,
|
||
key="rounded_subtitle_background_checkbox",
|
||
)
|
||
params.rounded_subtitle_background = (
|
||
selected_rounded_subtitle_background
|
||
if subtitle_background_enabled
|
||
else False
|
||
)
|
||
if not subtitle_settings_disabled and subtitle_background_enabled:
|
||
config.ui["rounded_subtitle_background"] = (
|
||
selected_rounded_subtitle_background
|
||
)
|
||
|
||
if video.subtitle_colors_are_indistinguishable(params):
|
||
# 同色配置仍然是合法的用户选择，因此只在字幕设置区域就近提示，
|
||
# 不阻止生成。用户可以根据实际视觉需求决定是否继续。
|
||
st.warning(tr("Subtitle Colors Are Indistinguishable"))
|
||
|
||
subtitle_preview_text = params.video_script or params.video_subject
|
||
selected_font_path = os.path.join(font_dir, params.font_name)
|
||
if (
|
||
params.subtitle_enabled
|
||
and subtitle_preview_text
|
||
and not video.subtitle_font_supports_text(
|
||
selected_font_path, subtitle_preview_text
|
||
)
|
||
):
|
||
st.warning(tr("Subtitle Font Does Not Support Text"))
|
||
|
||
if st.button(
|
||
tr("Restore Default Subtitle Settings"),
|
||
key="restore_default_subtitle_settings",
|
||
icon=":material/restart_alt:",
|
||
on_click=reset_subtitle_settings,
|
||
use_container_width=True,
|
||
):
|
||
st.toast(tr("Default Subtitle Settings Restored"))
|
||
|
||
|
||
def _render_generation_controls(
|
||
params, uploaded_files, uploaded_audio_file, voice_mode
|
||
):
|
||
"""校验生成依赖、执行任务，并渲染日志与成片结果。"""
|
||
restore_upload_requirements = st.session_state.get(
|
||
"task_restore_upload_requirements", {}
|
||
)
|
||
has_local_materials = bool(
|
||
uploaded_files or st.session_state.get("local_video_materials", [])
|
||
)
|
||
has_custom_audio = bool(uploaded_audio_file)
|
||
unmet_restore_requirements = _get_unmet_restore_upload_requirements(
|
||
restore_upload_requirements,
|
||
video_source=params.video_source,
|
||
voice_name=params.voice_name or "",
|
||
has_local_materials=has_local_materials,
|
||
has_custom_audio=has_custom_audio,
|
||
voice_mode=voice_mode,
|
||
)
|
||
if "local_materials" in unmet_restore_requirements:
|
||
st.warning(tr("Task Restore Local Materials Warning"))
|
||
if "custom_audio" in unmet_restore_requirements:
|
||
st.warning(tr("Task Restore Custom Audio Warning"))
|
||
if restore_upload_requirements and not unmet_restore_requirements:
|
||
# 用户已重新上传文件，或主动切换了素材来源/音色。此时历史任务的上传依赖
|
||
# 已经得到明确处理，清除标记，避免后续普通生成继续显示旧提示。
|
||
st.session_state.pop("task_restore_upload_requirements", None)
|
||
|
||
start_button = st.button(
|
||
tr("Generate Video"),
|
||
use_container_width=True,
|
||
type="primary",
|
||
key="generate_video_button",
|
||
on_click=_prepare_generation_task,
|
||
)
|
||
render_onboarding_tour()
|
||
log_container = st.empty()
|
||
if start_button:
|
||
st.session_state["generation_log_records"] = []
|
||
config.save_config()
|
||
task_id = st.session_state.get("pending_generation_task_id") or str(uuid4())
|
||
_add_active_generation_task(
|
||
task_id,
|
||
subject=params.video_subject or params.video_script or task_id,
|
||
)
|
||
if not params.video_subject and not params.video_script:
|
||
_remove_active_generation_task(task_id)
|
||
st.error(tr("Video Script and Subject Cannot Both Be Empty"))
|
||
st.stop()
|
||
|
||
if params.video_source not in ["pexels", "pixabay", "coverr", "local"]:
|
||
_remove_active_generation_task(task_id)
|
||
st.error(tr("Please Select a Valid Video Source"))
|
||
st.stop()
|
||
|
||
if params.video_source == "pexels" and not config.app.get(
|
||
"pexels_api_keys", ""
|
||
):
|
||
_remove_active_generation_task(task_id)
|
||
st.error(tr("Please Enter the Pexels API Key"))
|
||
st.stop()
|
||
|
||
if params.video_source == "pixabay" and not config.app.get(
|
||
"pixabay_api_keys", ""
|
||
):
|
||
_remove_active_generation_task(task_id)
|
||
st.error(tr("Please Enter the Pixabay API Key"))
|
||
st.stop()
|
||
|
||
if params.video_source == "coverr" and not config.app.get(
|
||
"coverr_api_keys", ""
|
||
):
|
||
_remove_active_generation_task(task_id)
|
||
st.error(tr("Please Enter the Coverr API Key"))
|
||
st.stop()
|
||
|
||
if params.video_source == "local" and not has_local_materials:
|
||
# 本地素材为空时继续执行会先产生 TTS/字幕，最后才在素材预处理阶段失败。
|
||
# 在任务启动前拦截，可以避免无意义的 API 调用和中间文件。
|
||
_remove_active_generation_task(task_id)
|
||
st.error(tr("Please Upload Local Materials First"))
|
||
st.stop()
|
||
|
||
if voice_mode == VOICE_MODE_UPLOAD and not uploaded_audio_file:
|
||
# 上传音频是用户显式选择的配音方式，缺少文件时不能静默退回 TTS。
|
||
# 在任务启动前拦截，避免产生与用户选择不一致的成片。
|
||
_remove_active_generation_task(task_id)
|
||
st.error(tr("Please Upload Voiceover File First"))
|
||
st.stop()
|
||
|
||
if "custom_audio" in unmet_restore_requirements:
|
||
# 历史自定义音频不能自动回填。用户尚未重新上传且也没有主动更换音色时，
|
||
# 必须阻止静默退回 TTS，否则重新生成的结果会与原任务语音不一致。
|
||
_remove_active_generation_task(task_id)
|
||
st.error(tr("Task Restore Custom Audio Warning"))
|
||
st.stop()
|
||
|
||
if uploaded_audio_file:
|
||
task_dir = utils.task_dir(task_id)
|
||
try:
|
||
custom_audio_path = _build_uploaded_file_path(
|
||
uploaded_audio_file,
|
||
task_dir,
|
||
CUSTOM_AUDIO_EXTENSIONS,
|
||
"custom-audio",
|
||
)
|
||
except ValueError:
|
||
_remove_active_generation_task(task_id)
|
||
st.error(tr("Unsupported Upload File Type"))
|
||
st.stop()
|
||
with open(custom_audio_path, "wb") as f:
|
||
f.write(uploaded_audio_file.getbuffer())
|
||
params.custom_audio_file = custom_audio_path
|
||
|
||
if uploaded_files:
|
||
local_videos_dir = utils.storage_dir("local_videos", create=True)
|
||
# 每次重新上传时都以本次选择的素材为准，避免旧素材不断重复追加。
|
||
params.video_materials = []
|
||
persisted_local_materials = []
|
||
for file in uploaded_files:
|
||
try:
|
||
file_path = _build_uploaded_file_path(
|
||
file,
|
||
local_videos_dir,
|
||
LOCAL_MATERIAL_EXTENSIONS,
|
||
"material",
|
||
)
|
||
except ValueError:
|
||
_remove_active_generation_task(task_id)
|
||
st.error(tr("Unsupported Upload File Type"))
|
||
st.stop()
|
||
with open(file_path, "wb") as f:
|
||
f.write(file.getbuffer())
|
||
m = MaterialInfo()
|
||
m.provider = "local"
|
||
m.url = file_path
|
||
params.video_materials.append(m)
|
||
persisted_local_materials.append(
|
||
{
|
||
"provider": m.provider,
|
||
"url": m.url,
|
||
"duration": m.duration,
|
||
}
|
||
)
|
||
# 将已上传并保存到本地的视频素材写入会话，供后续只改文案时直接复用。
|
||
st.session_state["local_video_materials"] = persisted_local_materials
|
||
elif (
|
||
params.video_source == "local" and st.session_state["local_video_materials"]
|
||
):
|
||
# 当用户没有重新上传文件时，复用最近一次已经保存到磁盘的本地素材列表。
|
||
params.video_materials = []
|
||
for material in st.session_state["local_video_materials"]:
|
||
m = MaterialInfo()
|
||
m.provider = material.get("provider", "local")
|
||
m.url = material.get("url", "")
|
||
m.duration = material.get("duration", 0)
|
||
if m.url:
|
||
params.video_materials.append(m)
|
||
|
||
def log_received(msg):
|
||
if config.ui["hide_log"]:
|
||
return
|
||
records = st.session_state.setdefault("generation_log_records", [])
|
||
records.append(str(msg).rstrip())
|
||
# 日志用于 WebUI 诊断，不需要无限增长；限制数量避免长任务反复
|
||
# rerun 后页面负担过重。
|
||
if len(records) > 1000:
|
||
del records[:-1000]
|
||
render_generation_logs(log_container)
|
||
|
||
log_handler_id = logger.add(log_received)
|
||
try:
|
||
st.toast(tr("Generating Video"))
|
||
logger.info(tr("Start Generating Video"))
|
||
logger.info(utils.to_json(params))
|
||
|
||
with config.runtime_config_lock():
|
||
result = tm.start(task_id=task_id, params=params)
|
||
if not result or "videos" not in result:
|
||
st.error(tr("Video Generation Failed"))
|
||
logger.error(tr("Video Generation Failed"))
|
||
st.stop()
|
||
|
||
video_files = result.get("videos", [])
|
||
st.success(tr("Video Generation Completed"))
|
||
try:
|
||
if video_files:
|
||
player_cols = st.columns(len(video_files) * 2 + 1)
|
||
for i, url in enumerate(video_files):
|
||
player_cols[i * 2 + 1].video(url)
|
||
except Exception as e:
|
||
logger.exception(
|
||
f"failed to render generated video preview: task_id={task_id}, "
|
||
f"video_files={video_files}, error={e}"
|
||
)
|
||
|
||
open_task_folder(task_id)
|
||
logger.info(tr("Video Generation Completed"))
|
||
finally:
|
||
_remove_active_generation_task(task_id)
|
||
remove_logger_handler_safely(log_handler_id)
|
||
|
||
render_generation_logs(log_container)
|
||
|
||
|
||
def _render_application():
|
||
"""按固定顺序渲染顶部栏、弹窗、生成表单和任务结果。"""
|
||
_render_top_bar()
|
||
|
||
if st.session_state.get("settings_dialog_open", False):
|
||
_render_settings_dialog()
|
||
|
||
restore_applied = _apply_pending_task_restore()
|
||
restore_candidate_id = st.session_state.get("task_restore_candidate_id")
|
||
if restore_candidate_id:
|
||
_render_task_restore_dialog(restore_candidate_id)
|
||
restore_succeeded = st.session_state.pop("task_restore_succeeded", False)
|
||
if restore_applied or restore_succeeded:
|
||
st.success(tr("Task Configuration Loaded"))
|
||
|
||
with st.container(key="main_settings_grid"):
|
||
panel = st.columns(4)
|
||
left_panel = panel[0]
|
||
middle_panel = panel[1]
|
||
audio_panel = panel[2]
|
||
right_panel = panel[3]
|
||
|
||
params = VideoParams(video_subject="")
|
||
params.match_materials_to_script = bool(
|
||
st.session_state.get("match_materials_to_script", False)
|
||
)
|
||
_render_script_settings(left_panel, params)
|
||
|
||
uploaded_files = _render_video_settings(middle_panel, params)
|
||
uploaded_audio_file, voice_mode = _render_audio_settings(audio_panel, params)
|
||
|
||
_render_subtitle_settings(right_panel, params)
|
||
|
||
_render_generation_controls(params, uploaded_files, uploaded_audio_file, voice_mode)
|
||
|
||
config.save_config()
|
||
|
||
|
||
_render_application()
|