ath-maas--pixelle-video
wehub-resource-sync
646 行
29 KiB
Python
646 行
29 KiB
Python
# Copyright (C) 2025 AIDC-AI
|
||
#
|
||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||
# you may not use this file except in compliance with the License.
|
||
# You may obtain a copy of the License at
|
||
# http://www.apache.org/licenses/LICENSE-2.0
|
||
# Unless required by applicable law or agreed to in writing, software
|
||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||
# See the License for the specific language governing permissions and
|
||
# limitations under the License.
|
||
|
||
"""
|
||
Asset-Based Pipeline UI
|
||
|
||
Implements the UI for generating videos from user-provided assets.
|
||
"""
|
||
|
||
import os
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import streamlit as st
|
||
from loguru import logger
|
||
|
||
from web.i18n import tr, get_language
|
||
from web.pipelines.base import PipelineUI, register_pipeline_ui
|
||
from web.pipelines.api_workflows import (
|
||
list_api_media_workflows,
|
||
render_api_video_controls,
|
||
workflow_select_help,
|
||
workflow_source_help,
|
||
workflow_source_label,
|
||
)
|
||
from web.components.content_input import render_bgm_section, render_version_info
|
||
from web.utils.async_helpers import run_async
|
||
from web.utils.streamlit_helpers import check_and_warn_selfhost_workflow
|
||
from pixelle_video.config import config_manager
|
||
from pixelle_video.models.progress import ProgressEvent
|
||
|
||
|
||
class AssetBasedPipelineUI(PipelineUI):
|
||
"""
|
||
UI for the Asset-Based Video Generation Pipeline.
|
||
Generates videos from user-provided assets (images/videos).
|
||
"""
|
||
name = "custom_media"
|
||
icon = "🎨"
|
||
|
||
@property
|
||
def display_name(self):
|
||
return tr("pipeline.custom_media.name")
|
||
|
||
@property
|
||
def description(self):
|
||
return tr("pipeline.custom_media.description")
|
||
|
||
def render(self, pixelle_video: Any):
|
||
# Three-column layout
|
||
left_col, middle_col, right_col = st.columns([1, 1, 1])
|
||
|
||
# ====================================================================
|
||
# Left Column: Asset Upload & Video Info
|
||
# ====================================================================
|
||
with left_col:
|
||
asset_params = self._render_asset_input()
|
||
bgm_params = render_bgm_section(key_prefix="asset_")
|
||
render_version_info()
|
||
|
||
# ====================================================================
|
||
# Middle Column: Video Configuration
|
||
# ====================================================================
|
||
with middle_col:
|
||
config_params = self._render_video_config(pixelle_video, asset_params)
|
||
|
||
# ====================================================================
|
||
# Right Column: Output Preview
|
||
# ====================================================================
|
||
with right_col:
|
||
# Combine all parameters
|
||
video_params = {
|
||
"pipeline": self.name,
|
||
**asset_params,
|
||
**bgm_params,
|
||
**config_params
|
||
}
|
||
|
||
self._render_output_preview(pixelle_video, video_params)
|
||
|
||
def _render_asset_input(self) -> dict:
|
||
"""Render asset upload section"""
|
||
with st.container(border=True):
|
||
st.markdown(f"**{tr('asset_based.section.assets')}**")
|
||
|
||
with st.expander(tr("help.feature_description"), expanded=False):
|
||
st.markdown(f"**{tr('help.what')}**")
|
||
st.markdown(tr("asset_based.assets.what"))
|
||
st.markdown(f"**{tr('help.how')}**")
|
||
st.markdown(tr("asset_based.assets.how"))
|
||
|
||
# File uploader for multiple files
|
||
uploaded_files = st.file_uploader(
|
||
tr("asset_based.assets.upload"),
|
||
type=["jpg", "jpeg", "png", "gif", "webp", "mp4", "mov", "avi", "mkv", "webm"],
|
||
accept_multiple_files=True,
|
||
help=tr("asset_based.assets.upload_help"),
|
||
key="asset_files"
|
||
)
|
||
|
||
# Save uploaded files to temp directory with unique session ID
|
||
asset_paths = []
|
||
if uploaded_files:
|
||
import uuid
|
||
session_id = str(uuid.uuid4()).replace('-', '')[:12]
|
||
temp_dir = Path(f"temp/assets_{session_id}")
|
||
temp_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
for uploaded_file in uploaded_files:
|
||
file_path = temp_dir / uploaded_file.name
|
||
with open(file_path, "wb") as f:
|
||
f.write(uploaded_file.getbuffer())
|
||
asset_paths.append(str(file_path.absolute()))
|
||
|
||
st.success(tr("asset_based.assets.count", count=len(asset_paths)))
|
||
|
||
# Preview uploaded assets
|
||
with st.expander(tr("asset_based.assets.preview"), expanded=True):
|
||
# Show in a grid (3 columns)
|
||
cols = st.columns(3)
|
||
for i, (file, path) in enumerate(zip(uploaded_files, asset_paths)):
|
||
with cols[i % 3]:
|
||
# Check if image or video
|
||
ext = Path(path).suffix.lower()
|
||
if ext in [".jpg", ".jpeg", ".png", ".gif", ".webp"]:
|
||
st.image(file, caption=file.name, use_container_width=True)
|
||
elif ext in [".mp4", ".mov", ".avi", ".mkv", ".webm"]:
|
||
st.video(file)
|
||
st.caption(file.name)
|
||
else:
|
||
st.info(tr("asset_based.assets.empty_hint"))
|
||
|
||
# Video title & intent
|
||
with st.container(border=True):
|
||
st.markdown(f"**{tr('asset_based.section.video_info')}**")
|
||
|
||
video_title = st.text_input(
|
||
tr("asset_based.video_title"),
|
||
placeholder=tr("asset_based.video_title_placeholder"),
|
||
help=tr("asset_based.video_title_help"),
|
||
key="asset_video_title"
|
||
)
|
||
|
||
intent = st.text_area(
|
||
tr("asset_based.intent"),
|
||
placeholder=tr("asset_based.intent_placeholder"),
|
||
help=tr("asset_based.intent_help"),
|
||
height=100,
|
||
key="asset_intent"
|
||
)
|
||
|
||
return {
|
||
"assets": asset_paths,
|
||
"video_title": video_title,
|
||
"intent": intent if intent else None
|
||
}
|
||
|
||
def _render_video_config(self, pixelle_video: Any, asset_params: dict | None = None) -> dict:
|
||
"""Render video configuration section"""
|
||
# Duration configuration
|
||
with st.container(border=True):
|
||
st.markdown(f"**{tr('video.title')}**")
|
||
|
||
# Duration slider
|
||
duration = st.slider(
|
||
tr("asset_based.duration"),
|
||
min_value=15,
|
||
max_value=120,
|
||
value=30,
|
||
step=5,
|
||
help=tr("asset_based.duration_help"),
|
||
key="asset_duration"
|
||
)
|
||
st.caption(tr("asset_based.duration_label", seconds=duration))
|
||
|
||
# Workflow source selection
|
||
with st.container(border=True):
|
||
st.markdown(f"**{tr('asset_based.section.source')}**")
|
||
|
||
with st.expander(tr("help.feature_description"), expanded=False):
|
||
st.markdown(f"**{tr('help.what')}**")
|
||
st.markdown(tr("asset_based.source.what"))
|
||
st.markdown(f"**{tr('help.how')}**")
|
||
st.markdown(tr("asset_based.source.how"))
|
||
|
||
source_options = {
|
||
"runninghub": tr("asset_based.source.runninghub"),
|
||
"selfhost": tr("asset_based.source.selfhost"),
|
||
"api": "API 调用" if get_language() == "zh_CN" else "API call",
|
||
}
|
||
|
||
# Check if RunningHub API key is configured
|
||
comfyui_config = config_manager.get_comfyui_config()
|
||
api_asset_analysis = getattr(pixelle_video, "api_asset_analysis", None)
|
||
api_vlm_models = (
|
||
api_asset_analysis.list_models(configured_only=True)
|
||
if api_asset_analysis is not None
|
||
else []
|
||
)
|
||
has_runninghub = bool(comfyui_config.get("runninghub_api_key"))
|
||
has_selfhost = bool(comfyui_config.get("comfyui_url"))
|
||
has_api_analysis = bool(api_vlm_models)
|
||
|
||
asset_paths = (asset_params or {}).get("assets") or []
|
||
image_exts = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
|
||
video_exts = {".mp4", ".mov", ".avi", ".mkv", ".webm"}
|
||
has_image_assets = any(Path(path).suffix.lower() in image_exts for path in asset_paths)
|
||
has_video_assets = any(Path(path).suffix.lower() in video_exts for path in asset_paths)
|
||
|
||
def analysis_source_available(source_name: str) -> bool:
|
||
source_dir = Path("workflows") / source_name
|
||
image_available = (source_dir / "analyse_image.json").exists()
|
||
video_available = (source_dir / "analyse_video.json").exists()
|
||
if has_image_assets and not image_available:
|
||
return False
|
||
if has_video_assets and not video_available:
|
||
return False
|
||
return image_available or video_available
|
||
|
||
# Prefer API VLM when configured, so API media workflows do not depend on RunningHub.
|
||
source_keys = []
|
||
if analysis_source_available("runninghub"):
|
||
source_keys.append("runninghub")
|
||
if analysis_source_available("selfhost"):
|
||
source_keys.append("selfhost")
|
||
if has_api_analysis:
|
||
source_keys.append("api")
|
||
if not source_keys:
|
||
source_keys = ["runninghub"]
|
||
|
||
if has_api_analysis and "api" in source_keys:
|
||
default_source = "api"
|
||
elif has_runninghub and "runninghub" in source_keys:
|
||
default_source = "runninghub"
|
||
elif "selfhost" in source_keys:
|
||
default_source = "selfhost"
|
||
else:
|
||
default_source = source_keys[0]
|
||
default_source_index = source_keys.index(default_source)
|
||
|
||
if st.session_state.get("asset_source") not in source_keys:
|
||
st.session_state.pop("asset_source", None)
|
||
|
||
source = st.radio(
|
||
"素材分析服务" if get_language() == "zh_CN" else "Asset analysis service",
|
||
options=source_keys,
|
||
format_func=lambda x: source_options[x],
|
||
index=default_source_index,
|
||
horizontal=True,
|
||
key="asset_source",
|
||
label_visibility="visible",
|
||
help=workflow_source_help("素材分析" if get_language() == "zh_CN" else "asset analysis"),
|
||
)
|
||
|
||
def build_analysis_workflows(source_name: str) -> list[dict]:
|
||
if source_name == "api":
|
||
return [
|
||
{
|
||
"display_name": model_info["display_name"],
|
||
"image_workflow": None,
|
||
"video_workflow": None,
|
||
"model": model_info["model"],
|
||
}
|
||
for model_info in api_vlm_models
|
||
]
|
||
|
||
source_dir = Path("workflows") / source_name
|
||
needs_image = has_image_assets or not asset_paths
|
||
needs_video = has_video_assets or not asset_paths
|
||
image_workflow = None
|
||
video_workflow = None
|
||
workflow_names = []
|
||
|
||
if needs_image and (source_dir / "analyse_image.json").exists():
|
||
image_workflow = f"{source_name}/analyse_image.json"
|
||
workflow_names.append("analyse_image.json")
|
||
if needs_video and (source_dir / "analyse_video.json").exists():
|
||
video_workflow = f"{source_name}/analyse_video.json"
|
||
workflow_names.append("analyse_video.json")
|
||
|
||
if not workflow_names:
|
||
return []
|
||
|
||
return [{
|
||
"display_name": f"{' + '.join(workflow_names)} - {workflow_source_label(source_name)}",
|
||
"image_workflow": image_workflow,
|
||
"video_workflow": video_workflow,
|
||
"model": None,
|
||
}]
|
||
|
||
analysis_workflows = build_analysis_workflows(source)
|
||
analysis_options = [workflow["display_name"] for workflow in analysis_workflows]
|
||
selected_analysis_workflow = {}
|
||
|
||
if st.session_state.get("asset_analysis_workflow") not in analysis_options:
|
||
st.session_state.pop("asset_analysis_workflow", None)
|
||
|
||
if analysis_options:
|
||
selected_analysis = st.selectbox(
|
||
"素材分析工作流/模型" if get_language() == "zh_CN" else "Asset analysis workflow/model",
|
||
analysis_options,
|
||
index=0,
|
||
key="asset_analysis_workflow",
|
||
help=workflow_select_help(),
|
||
)
|
||
selected_analysis_workflow = analysis_workflows[analysis_options.index(selected_analysis)]
|
||
else:
|
||
st.warning(
|
||
"当前服务没有可用的素材分析工作流/模型。"
|
||
if get_language() == "zh_CN"
|
||
else "No asset analysis workflow/model is available for the selected service."
|
||
)
|
||
|
||
# Show hint based on selection
|
||
if source == "api":
|
||
if not has_api_analysis:
|
||
st.warning(
|
||
"未配置可用于 VLM 素材分析的 API Key（DashScope/OpenAI/Gemini）。"
|
||
if get_language() == "zh_CN"
|
||
else "No API key configured for VLM asset analysis (DashScope/OpenAI/Gemini)."
|
||
)
|
||
else:
|
||
st.info(
|
||
"使用上方选择的 API VLM 模型分析上传素材，不依赖 RunningHub/ComfyUI。"
|
||
if get_language() == "zh_CN"
|
||
else "Use the selected API VLM model to analyze uploaded assets without RunningHub/ComfyUI."
|
||
)
|
||
elif source == "runninghub":
|
||
if not has_runninghub:
|
||
st.warning(tr("asset_based.source.runninghub_not_configured"))
|
||
else:
|
||
st.info(tr("asset_based.source.runninghub_hint"))
|
||
else:
|
||
if not has_selfhost:
|
||
st.warning(tr("asset_based.source.selfhost_not_configured"))
|
||
else:
|
||
st.info(tr("asset_based.source.selfhost_hint"))
|
||
# Check and warn for selfhost mode (auto popup if not confirmed)
|
||
workflow_for_warning = (
|
||
selected_analysis_workflow.get("image_workflow")
|
||
or selected_analysis_workflow.get("video_workflow")
|
||
)
|
||
if workflow_for_warning:
|
||
check_and_warn_selfhost_workflow(workflow_for_warning)
|
||
|
||
api_video_workflow = None
|
||
api_video_params = {}
|
||
api_video_workflows = list_api_media_workflows(
|
||
pixelle_video,
|
||
"video",
|
||
required_adapter_abilities=["first_frame_i2v"],
|
||
verified_only=True,
|
||
)
|
||
animation_source_options = ["none"]
|
||
if api_video_workflows:
|
||
animation_source_options.append("api")
|
||
|
||
if st.session_state.get("asset_animation_source") not in animation_source_options:
|
||
st.session_state.pop("asset_animation_source", None)
|
||
|
||
def animation_source_label(value: str) -> str:
|
||
if value == "none":
|
||
return "不启用" if get_language() == "zh_CN" else "Disabled"
|
||
return workflow_source_label(value)
|
||
|
||
animation_source = st.radio(
|
||
"素材动画服务" if get_language() == "zh_CN" else "Asset animation service",
|
||
animation_source_options,
|
||
format_func=animation_source_label,
|
||
horizontal=True,
|
||
key="asset_animation_source",
|
||
help=(
|
||
"选择是否把匹配到的图片素材动画化。不启用时保留原素材静态合成；API 模型会调用已验证的图生视频模型。"
|
||
if get_language() == "zh_CN"
|
||
else "Choose whether to animate matched image assets. Disabled keeps the original static asset composition; API models call verified image-to-video providers."
|
||
),
|
||
)
|
||
|
||
if animation_source == "api":
|
||
animation_workflows = api_video_workflows
|
||
animation_options = [wf["display_name"] for wf in animation_workflows]
|
||
selected_animation = st.selectbox(
|
||
"素材动画工作流/模型" if get_language() == "zh_CN" else "Asset animation workflow/model",
|
||
animation_options,
|
||
index=0,
|
||
key="asset_animation_workflow",
|
||
help=workflow_select_help(),
|
||
)
|
||
selected_index = animation_options.index(selected_animation)
|
||
selected_workflow = animation_workflows[selected_index]
|
||
api_video_workflow = selected_workflow["key"]
|
||
api_video_params = render_api_video_controls(
|
||
selected_workflow,
|
||
key_prefix="asset",
|
||
default_duration=5,
|
||
allow_audio_driven=True,
|
||
show_duration=False,
|
||
)
|
||
|
||
# TTS configuration
|
||
with st.container(border=True):
|
||
st.markdown(f"**{tr('section.tts')}**")
|
||
|
||
# Import voice configuration
|
||
from pixelle_video.tts_voices import EDGE_TTS_VOICES, get_voice_display_name
|
||
|
||
# Get saved voice from config
|
||
comfyui_config = config_manager.get_comfyui_config()
|
||
tts_config = comfyui_config.get("tts", {})
|
||
local_config = tts_config.get("local", {})
|
||
saved_voice = local_config.get("voice", "zh-CN-YunjianNeural")
|
||
saved_speed = local_config.get("speed", 1.2)
|
||
|
||
# Build voice options with i18n
|
||
voice_options = []
|
||
voice_ids = []
|
||
default_voice_index = 0
|
||
|
||
for idx, voice_config in enumerate(EDGE_TTS_VOICES):
|
||
voice_id = voice_config["id"]
|
||
display_name = get_voice_display_name(voice_id, tr, get_language())
|
||
voice_options.append(display_name)
|
||
voice_ids.append(voice_id)
|
||
|
||
if voice_id == saved_voice:
|
||
default_voice_index = idx
|
||
|
||
# Two-column layout
|
||
voice_col, speed_col = st.columns([1, 1])
|
||
|
||
with voice_col:
|
||
selected_voice_display = st.selectbox(
|
||
tr("tts.voice_selector"),
|
||
voice_options,
|
||
index=default_voice_index,
|
||
key="asset_tts_voice"
|
||
)
|
||
selected_voice_index = voice_options.index(selected_voice_display)
|
||
voice_id = voice_ids[selected_voice_index]
|
||
|
||
with speed_col:
|
||
tts_speed = st.slider(
|
||
tr("tts.speed"),
|
||
min_value=0.5,
|
||
max_value=2.0,
|
||
value=saved_speed,
|
||
step=0.1,
|
||
format="%.1fx",
|
||
key="asset_tts_speed"
|
||
)
|
||
st.caption(tr("tts.speed_label", speed=f"{tts_speed:.1f}"))
|
||
|
||
return {
|
||
"duration": duration,
|
||
"source": source,
|
||
"analysis_image_workflow": selected_analysis_workflow.get("image_workflow"),
|
||
"analysis_video_workflow": selected_analysis_workflow.get("video_workflow"),
|
||
"analysis_vlm_model": selected_analysis_workflow.get("model"),
|
||
"api_video_workflow": api_video_workflow,
|
||
"api_video_params": api_video_params,
|
||
"voice_id": voice_id,
|
||
"tts_speed": tts_speed
|
||
}
|
||
|
||
def _render_output_preview(self, pixelle_video: Any, video_params: dict):
|
||
"""Render output preview section"""
|
||
with st.container(border=True):
|
||
st.markdown(f"**{tr('section.video_generation')}**")
|
||
|
||
# Check configuration
|
||
if not config_manager.validate():
|
||
st.warning(tr("settings.not_configured"))
|
||
|
||
# Check if assets are provided
|
||
assets = video_params.get("assets", [])
|
||
if not assets:
|
||
st.info(tr("asset_based.output.no_assets"))
|
||
st.button(
|
||
tr("btn.generate"),
|
||
type="primary",
|
||
use_container_width=True,
|
||
disabled=True,
|
||
key="asset_generate_disabled"
|
||
)
|
||
return
|
||
|
||
# Show asset summary
|
||
st.info(tr("asset_based.output.ready", count=len(assets)))
|
||
|
||
# Generate button
|
||
if st.button(tr("btn.generate"), type="primary", use_container_width=True, key="asset_generate"):
|
||
# Validate
|
||
if not config_manager.validate():
|
||
st.error(tr("settings.not_configured"))
|
||
st.stop()
|
||
|
||
# Show progress
|
||
progress_bar = st.progress(0)
|
||
status_text = st.empty()
|
||
|
||
start_time = time.time()
|
||
|
||
try:
|
||
# Import pipeline
|
||
from pixelle_video.pipelines.asset_based import AssetBasedPipeline
|
||
|
||
# Create pipeline
|
||
pipeline = AssetBasedPipeline(pixelle_video)
|
||
|
||
# Progress callback
|
||
def update_progress(event: ProgressEvent):
|
||
if event.event_type == "analyzing_assets":
|
||
if event.extra_info == "start":
|
||
message = tr("asset_based.progress.analyzing_start", total=event.frame_total)
|
||
else:
|
||
message = tr("asset_based.progress.analyzing_complete", count=event.frame_total)
|
||
elif event.event_type == "analyzing_asset":
|
||
message = tr(
|
||
"asset_based.progress.analyzing_asset",
|
||
current=event.frame_current,
|
||
total=event.frame_total,
|
||
name=event.extra_info or ""
|
||
)
|
||
elif event.event_type == "generating_script":
|
||
if event.extra_info == "complete":
|
||
message = tr("asset_based.progress.script_complete")
|
||
else:
|
||
message = tr("asset_based.progress.generating_script")
|
||
elif event.event_type == "frame_step":
|
||
action_key = f"progress.step_{event.action}"
|
||
action_text = tr(action_key)
|
||
message = tr(
|
||
"progress.frame_step",
|
||
current=event.frame_current,
|
||
total=event.frame_total,
|
||
step=event.step,
|
||
action=action_text
|
||
)
|
||
elif event.event_type == "processing_frame":
|
||
message = tr(
|
||
"progress.frame",
|
||
current=event.frame_current,
|
||
total=event.frame_total
|
||
)
|
||
elif event.event_type == "concatenating":
|
||
if event.extra_info == "complete":
|
||
message = tr("asset_based.progress.concat_complete")
|
||
else:
|
||
message = tr("progress.concatenating")
|
||
elif event.event_type == "completed":
|
||
message = tr("progress.completed")
|
||
else:
|
||
message = tr(f"progress.{event.event_type}")
|
||
|
||
status_text.text(message)
|
||
progress_bar.progress(min(int(event.progress * 100), 99))
|
||
|
||
# Execute pipeline with progress callback
|
||
if video_params.get("source") == "api" and not video_params.get("analysis_vlm_model"):
|
||
raise RuntimeError(
|
||
"请先在素材分析服务中选择 API VLM 模型。"
|
||
if get_language() == "zh_CN"
|
||
else "Please select an API VLM model in the asset analysis service settings."
|
||
)
|
||
|
||
ctx = run_async(pipeline(
|
||
assets=video_params["assets"],
|
||
video_title=video_params.get("video_title", ""),
|
||
intent=video_params.get("intent"),
|
||
duration=video_params.get("duration", 30),
|
||
source=video_params.get("source", "runninghub"),
|
||
analysis_image_workflow=video_params.get("analysis_image_workflow"),
|
||
analysis_video_workflow=video_params.get("analysis_video_workflow"),
|
||
analysis_vlm_model=video_params.get("analysis_vlm_model"),
|
||
bgm_path=video_params.get("bgm_path"),
|
||
bgm_volume=video_params.get("bgm_volume", 0.2),
|
||
bgm_mode=video_params.get("bgm_mode", "loop"),
|
||
api_video_workflow=video_params.get("api_video_workflow"),
|
||
api_video_params=video_params.get("api_video_params"),
|
||
voice_id=video_params.get("voice_id", "zh-CN-YunjianNeural"),
|
||
tts_speed=video_params.get("tts_speed", 1.2),
|
||
progress_callback=update_progress
|
||
))
|
||
|
||
total_time = time.time() - start_time
|
||
|
||
progress_bar.progress(100)
|
||
status_text.text(tr("status.success"))
|
||
|
||
# Display result
|
||
st.success(tr("status.video_generated", path=ctx.final_video_path))
|
||
|
||
st.markdown("---")
|
||
|
||
# Video info
|
||
if os.path.exists(ctx.final_video_path):
|
||
file_size_mb = os.path.getsize(ctx.final_video_path) / (1024 * 1024)
|
||
n_scenes = len(ctx.storyboard.frames) if ctx.storyboard else 0
|
||
|
||
info_text = (
|
||
f"⏱️ {tr('info.generation_time')} {total_time:.1f}s "
|
||
f"📦 {file_size_mb:.2f}MB "
|
||
f"🎬 {n_scenes}{tr('info.scenes_unit')}"
|
||
)
|
||
st.caption(info_text)
|
||
|
||
st.markdown("---")
|
||
|
||
# Video preview
|
||
st.video(ctx.final_video_path)
|
||
|
||
# Download button
|
||
with open(ctx.final_video_path, "rb") as video_file:
|
||
video_bytes = video_file.read()
|
||
video_filename = os.path.basename(ctx.final_video_path)
|
||
st.download_button(
|
||
label="⬇️ 下载视频" if get_language() == "zh_CN" else "⬇️ Download Video",
|
||
data=video_bytes,
|
||
file_name=video_filename,
|
||
mime="video/mp4",
|
||
use_container_width=True
|
||
)
|
||
else:
|
||
st.error(tr("status.video_not_found", path=ctx.final_video_path))
|
||
|
||
except Exception as e:
|
||
status_text.text("")
|
||
progress_bar.empty()
|
||
st.error(tr("status.error", error=str(e)))
|
||
logger.exception(e)
|
||
st.stop()
|
||
|
||
|
||
# Register self
|
||
register_pipeline_ui(AssetBasedPipelineUI)
|