aidc-ai--pixelle-video
wehub-resource-sync
965 行
52 KiB
Python
965 行
52 KiB
Python
import os
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import streamlit as st
|
|
from loguru import logger
|
|
import httpx
|
|
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_version_info
|
|
from web.components.digital_tts_config import render_style_config
|
|
from web.utils.async_helpers import run_async
|
|
from web.utils.history_persistence import save_web_generation_history
|
|
from web.utils.streamlit_helpers import check_and_warn_selfhost_workflow
|
|
from pixelle_video.config import config_manager
|
|
from pixelle_video.utils.os_util import create_task_output_dir
|
|
|
|
class DigitalHumanPipelineUI(PipelineUI):
|
|
"""
|
|
UI for the Digital_Human Video Generation Pipeline.
|
|
Generates videos from user-provided assets (images&videos&audio).
|
|
"""
|
|
name = "digital_human"
|
|
icon = "🤖"
|
|
|
|
@property
|
|
def display_name(self):
|
|
return tr("pipeline.digital_human.name")
|
|
|
|
@property
|
|
def description(self):
|
|
return tr("pipeline.digital_human.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
|
|
# ====================================================================
|
|
with left_col:
|
|
asset_params = self.render_digital_human_input()
|
|
style_params = render_style_config(pixelle_video)
|
|
# bgm_params = render_bgm_section(key_prefix="asset_")
|
|
render_version_info()
|
|
|
|
# ====================================================================
|
|
# Middle Column: Video Configuration
|
|
# ====================================================================
|
|
with middle_col:
|
|
# Style configuration ()
|
|
workflow_path = self.workflow_path_config(pixelle_video)
|
|
mode_params = self.render_digital_human_mode(asset_params["character_assets"])
|
|
|
|
# ====================================================================
|
|
# Right Column: Output Preview
|
|
# ====================================================================
|
|
with right_col:
|
|
# Combine all parameters
|
|
video_params = {
|
|
**mode_params,
|
|
**asset_params,
|
|
**style_params,
|
|
"workflow_path": workflow_path
|
|
}
|
|
|
|
self._render_output_preview(pixelle_video, video_params)
|
|
|
|
def render_digital_human_input(self) -> dict:
|
|
"""Render digital human character image upload section"""
|
|
with st.container(border=True):
|
|
st.markdown(f"**{tr('digital_human.section.character_assets')}**")
|
|
|
|
with st.expander(tr("help.feature_description"), expanded=False):
|
|
st.markdown(f"**{tr('help.what')}**")
|
|
st.markdown(tr("digital_human.assets.character_what"))
|
|
st.markdown(f"**{tr('help.how')}**")
|
|
st.markdown(tr("digital_human.assets.how"))
|
|
|
|
# File uploader for multiple files
|
|
uploaded_files = st.file_uploader(
|
|
tr("digital_human.assets.upload"),
|
|
type=["jpg", "jpeg", "png", "webp"],
|
|
accept_multiple_files=True,
|
|
help=tr("digital_human.assets.upload_help"),
|
|
key="character_files"
|
|
)
|
|
|
|
# Save uploaded files to temp directory with unique session ID
|
|
character_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())
|
|
character_asset_paths.append(str(file_path.absolute()))
|
|
|
|
st.success(tr("digital_human.assets.character_sucess"))
|
|
|
|
# Preview uploaded assets
|
|
with st.expander(tr("digital_human.assets.preview"), expanded=True):
|
|
# Show in a grid (3 columns)
|
|
cols = st.columns(3)
|
|
for i, (file, path) in enumerate(zip(uploaded_files, character_asset_paths)):
|
|
with cols[i % 3]:
|
|
# Check if image
|
|
ext = Path(path).suffix.lower()
|
|
if ext in [".jpg", ".jpeg", ".png", ".webp"]:
|
|
st.image(file, caption=file.name, use_container_width=True)
|
|
else:
|
|
st.info(tr("digital_human.assets.character_empty_hint"))
|
|
|
|
return {"character_assets": character_asset_paths}
|
|
|
|
def workflow_path_config(self, pixelle_video: Any) -> dict:
|
|
# 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": workflow_source_label("api"),
|
|
}
|
|
|
|
# Check if RunningHub API key is configured
|
|
comfyui_config = config_manager.get_comfyui_config()
|
|
has_runninghub = bool(comfyui_config.get("runninghub_api_key"))
|
|
has_selfhost = bool(comfyui_config.get("comfyui_url"))
|
|
|
|
workflow_config = {
|
|
"first_workflow_path": "workflows/runninghub/digital_image.json",
|
|
"second_workflow_path": "workflows/runninghub/digital_combination.json",
|
|
"third_workflow_path": "workflows/runninghub/digital_customize.json",
|
|
}
|
|
|
|
def digital_image_workflows(source_name: str) -> list[dict]:
|
|
first_path = Path("workflows") / source_name / "digital_image.json"
|
|
third_path = Path("workflows") / source_name / "digital_customize.json"
|
|
if not first_path.exists() or not third_path.exists():
|
|
return []
|
|
return [
|
|
{
|
|
"key": f"{source_name}/digital_image.json",
|
|
"display_name": f"digital_image.json - {source_name.title()}",
|
|
"first_workflow_path": str(first_path),
|
|
"third_workflow_path": str(third_path),
|
|
}
|
|
]
|
|
|
|
def digital_video_workflows(source_name: str) -> list[dict]:
|
|
second_path = Path("workflows") / source_name / "digital_combination.json"
|
|
if not second_path.exists():
|
|
return []
|
|
return [
|
|
{
|
|
"key": f"{source_name}/digital_combination.json",
|
|
"display_name": f"digital_combination.json - {source_name.title()}",
|
|
"second_workflow_path": str(second_path),
|
|
}
|
|
]
|
|
|
|
api_image_workflows = list_api_media_workflows(pixelle_video, "image")
|
|
image_source_options = []
|
|
if digital_image_workflows("runninghub"):
|
|
image_source_options.append("runninghub")
|
|
if digital_image_workflows("selfhost"):
|
|
image_source_options.append("selfhost")
|
|
if api_image_workflows:
|
|
image_source_options.append("api")
|
|
|
|
workflow_config["api_image_workflow"] = None
|
|
if st.session_state.get("digital_human_image_service_source") not in image_source_options:
|
|
st.session_state.pop("digital_human_image_service_source", None)
|
|
image_service_source = st.radio(
|
|
"前置图片生成服务" if get_language() == "zh_CN" else "Pre-image generation service",
|
|
image_source_options,
|
|
format_func=lambda x: source_options[x],
|
|
horizontal=True,
|
|
key="digital_human_image_service_source",
|
|
help=workflow_source_help("前置图片生成" if get_language() == "zh_CN" else "pre-image generation"),
|
|
)
|
|
|
|
image_workflows = []
|
|
if image_service_source in {"runninghub", "selfhost"}:
|
|
if image_service_source == "runninghub" and not has_runninghub:
|
|
st.warning(tr("asset_based.source.runninghub_not_configured"))
|
|
if image_service_source == "selfhost" and not has_selfhost:
|
|
st.warning(tr("asset_based.source.selfhost_not_configured"))
|
|
|
|
image_workflows = digital_image_workflows(image_service_source)
|
|
elif image_service_source == "api":
|
|
if not api_image_workflows:
|
|
st.warning(
|
|
"没有找到 API 图片模型，请先配置图像模型提供商。"
|
|
if get_language() == "zh_CN"
|
|
else "No API image model found. Configure an image provider first."
|
|
)
|
|
else:
|
|
image_workflows = api_image_workflows
|
|
|
|
image_options = [wf["display_name"] for wf in image_workflows]
|
|
selected_image_workflow = st.selectbox(
|
|
"前置图片工作流/模型" if get_language() == "zh_CN" else "Pre-image workflow/model",
|
|
image_options if image_options else ["No workflow/model available"],
|
|
index=0,
|
|
key="digital_human_image_workflow",
|
|
disabled=not image_options,
|
|
help=workflow_select_help(),
|
|
)
|
|
if image_options:
|
|
selected_index = image_options.index(selected_image_workflow)
|
|
selected_workflow = image_workflows[selected_index]
|
|
if image_service_source == "api":
|
|
workflow_config["api_image_workflow"] = selected_workflow["key"]
|
|
else:
|
|
workflow_config["first_workflow_path"] = selected_workflow["first_workflow_path"]
|
|
workflow_config["third_workflow_path"] = selected_workflow["third_workflow_path"]
|
|
|
|
workflow_config["api_video_workflow"] = None
|
|
workflow_config["api_video_params"] = {}
|
|
api_video_workflows = list_api_media_workflows(
|
|
pixelle_video,
|
|
"video",
|
|
required_adapter_abilities=["digital_human"],
|
|
verified_only=True,
|
|
)
|
|
video_source_options = []
|
|
if digital_video_workflows("runninghub"):
|
|
video_source_options.append("runninghub")
|
|
if digital_video_workflows("selfhost"):
|
|
video_source_options.append("selfhost")
|
|
if api_video_workflows:
|
|
video_source_options.append("api")
|
|
|
|
if st.session_state.get("digital_human_video_service_source") not in video_source_options:
|
|
st.session_state.pop("digital_human_video_service_source", None)
|
|
video_service_source = st.radio(
|
|
"口播视频合成服务" if get_language() == "zh_CN" else "Talking-video synthesis service",
|
|
video_source_options,
|
|
format_func=lambda x: source_options[x],
|
|
horizontal=True,
|
|
key="digital_human_video_service_source",
|
|
help=workflow_source_help("口播视频合成" if get_language() == "zh_CN" else "talking-video synthesis"),
|
|
)
|
|
|
|
video_workflows = []
|
|
if video_service_source in {"runninghub", "selfhost"}:
|
|
if video_service_source == "runninghub" and not has_runninghub:
|
|
st.warning(tr("asset_based.source.runninghub_not_configured"))
|
|
if video_service_source == "selfhost" and not has_selfhost:
|
|
st.warning(tr("asset_based.source.selfhost_not_configured"))
|
|
|
|
video_workflows = digital_video_workflows(video_service_source)
|
|
elif video_service_source == "api":
|
|
if not api_video_workflows:
|
|
st.warning(
|
|
"没有找到已验证的 API 参考生视频模型，请先配置 DashScope 等提供商。"
|
|
if get_language() == "zh_CN"
|
|
else "No verified API reference-to-video model found. Configure a provider first."
|
|
)
|
|
else:
|
|
video_workflows = api_video_workflows
|
|
|
|
video_options = [wf["display_name"] for wf in video_workflows]
|
|
selected_video_workflow = st.selectbox(
|
|
"口播视频工作流/模型" if get_language() == "zh_CN" else "Talking-video workflow/model",
|
|
video_options if video_options else ["No workflow/model available"],
|
|
index=0,
|
|
key="digital_human_video_workflow",
|
|
disabled=not video_options,
|
|
help=workflow_select_help(),
|
|
)
|
|
if video_options:
|
|
selected_index = video_options.index(selected_video_workflow)
|
|
selected_workflow = video_workflows[selected_index]
|
|
if video_service_source == "api":
|
|
workflow_config["api_video_workflow"] = selected_workflow["key"]
|
|
workflow_config["api_video_params"] = render_api_video_controls(
|
|
selected_workflow,
|
|
key_prefix="digital_human",
|
|
default_duration=5,
|
|
)
|
|
else:
|
|
workflow_config["second_workflow_path"] = selected_workflow["second_workflow_path"]
|
|
|
|
missing_workflows = [
|
|
path for key, path in workflow_config.items()
|
|
if key.endswith("_workflow_path") and isinstance(path, str) and not Path(path).exists()
|
|
]
|
|
if missing_workflows:
|
|
st.warning(
|
|
(
|
|
"当前选择缺少数字人口播工作流文件："
|
|
+ "、".join(missing_workflows)
|
|
)
|
|
if get_language() == "zh_CN"
|
|
else "The current selection is missing digital-human workflow files: "
|
|
+ ", ".join(missing_workflows)
|
|
)
|
|
|
|
return workflow_config
|
|
|
|
def render_digital_human_mode(self, character_asset_paths: list) -> dict:
|
|
with st.container(border=True):
|
|
st.markdown(f"**{tr('digital_human.section.select_mode')}**")
|
|
|
|
with st.expander(tr("help.feature_description"), expanded=False):
|
|
st.markdown(f"**{tr('help.what')}**")
|
|
st.markdown(tr("digital_human.assets.mode_what"))
|
|
st.markdown(f"**{tr('help.how')}**")
|
|
st.markdown(tr("digital_human.assets.select_how"))
|
|
|
|
mode = st.radio(
|
|
"Processing Mode",
|
|
["digital", "customize"],
|
|
horizontal=True,
|
|
format_func=lambda x: tr(f"mode.{x}"),
|
|
label_visibility="collapsed",
|
|
key="mode_selection"
|
|
)
|
|
|
|
# Text input (unified for both modes)
|
|
text_placeholder = tr("digital_human.input.topic_placeholder") if mode == "digital" else tr("digital_human.input.content_placeholder")
|
|
text_height = 120 if mode == "digital" else 200
|
|
text_help = tr("input.text_help_digital") if mode == "digital" else tr("input.text_help_fixed")
|
|
|
|
if mode == "digital":
|
|
# File uploader for multiple files
|
|
uploaded_files = st.file_uploader(
|
|
tr("digital_human.assets.upload"),
|
|
type=["jpg", "jpeg", "png", "webp"],
|
|
accept_multiple_files=True,
|
|
help=tr("digital_human.assets.upload_help"),
|
|
key="digital_files"
|
|
)
|
|
|
|
# Save uploaded files to temp directory with unique session ID
|
|
goods_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())
|
|
goods_asset_paths.append(str(file_path.absolute()))
|
|
|
|
st.success(tr("digital_human.assets.goods_sucess"))
|
|
|
|
# Preview uploaded assets
|
|
with st.expander(tr("digital_human.assets.preview"), expanded=True):
|
|
# Show in a grid (3 columns)
|
|
cols = st.columns(3)
|
|
for i, (file, path) in enumerate(zip(uploaded_files, goods_asset_paths)):
|
|
with cols[i % 3]:
|
|
# Check if image
|
|
ext = Path(path).suffix.lower()
|
|
if ext in [".jpg", ".jpeg", ".png", ".webp"]:
|
|
st.image(file, caption=file.name, use_container_width=True)
|
|
else:
|
|
st.info(tr("digital_human.assets.goods_empty_hint"))
|
|
# Text input
|
|
goods_text = st.text_area(
|
|
tr("digital_human.input_text"),
|
|
placeholder=text_placeholder,
|
|
height=text_height,
|
|
help=text_help,
|
|
key="digital_box"
|
|
)
|
|
|
|
goods_title = st.text_input(
|
|
tr("digital_human.goods_title"),
|
|
placeholder=tr("digital_human.goods_title_placeholder"),
|
|
help=tr("digital_human.goods_title_help"),
|
|
key="goods_title"
|
|
)
|
|
|
|
return {
|
|
"character_assets": character_asset_paths,
|
|
"goods_title": goods_title,
|
|
"goods_assets": goods_asset_paths,
|
|
"goods_text": goods_text,
|
|
"mode": mode
|
|
}
|
|
|
|
else:
|
|
goods_text = st.text_area(
|
|
tr("digital_human.customize_text"),
|
|
placeholder=text_placeholder,
|
|
height=text_height,
|
|
help=text_help,
|
|
key="customize_box"
|
|
)
|
|
|
|
return {
|
|
"character_assets": character_asset_paths,
|
|
"goods_text": goods_text,
|
|
"mode": mode
|
|
}
|
|
|
|
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"))
|
|
|
|
# Get input data
|
|
character_assets = video_params.get("character_assets", [])
|
|
goods_assets = video_params.get("goods_assets", [])
|
|
goods_title = video_params.get("goods_title", "")
|
|
goods_text = video_params.get("goods_text", "")
|
|
mode = video_params.get("mode")
|
|
tts_voice = video_params.get("tts_voice", "zh-CN-YunjianNeural")
|
|
tts_speed = video_params.get("tts_speed", 1.2)
|
|
|
|
logger.info(f"🔧 The obtained TTS parameters:")
|
|
logger.info(f" - tts_voice: {tts_voice}")
|
|
logger.info(f" - tts_speed: {tts_speed}")
|
|
logger.info(f" - video_params中的tts_voice: {video_params.get('tts_voice', 'NOT_FOUND')}")
|
|
logger.info(f" - video_params: {video_params}")
|
|
|
|
# Validation
|
|
if not character_assets:
|
|
st.info(tr("digital_human.assets.character_warning"))
|
|
st.button(
|
|
tr("btn.generate"),
|
|
type="primary",
|
|
use_container_width=True,
|
|
disabled=True,
|
|
key="digital_human_generate_disabled"
|
|
)
|
|
return
|
|
|
|
if mode == "digital" and not goods_assets:
|
|
st.info(tr("digital_human.assets.goods_warning"))
|
|
st.button(
|
|
tr("btn.generate"),
|
|
type="primary",
|
|
use_container_width=True,
|
|
disabled=True,
|
|
key="digital_human_goods_vaiidation"
|
|
)
|
|
return
|
|
|
|
if mode == "digital" and not (goods_text or goods_title):
|
|
st.info(tr("digital_human.assets.digital_mode"))
|
|
st.button(
|
|
tr("btn.generate"),
|
|
type="primary",
|
|
use_container_width=True,
|
|
disabled=True,
|
|
key="digital_human_digital_disable"
|
|
)
|
|
return
|
|
|
|
if mode == "digital" and (goods_text or goods_title):
|
|
st.warning(tr("digital_human.assets.digital_mode_warning"))
|
|
|
|
if mode == "customize" and not goods_text:
|
|
st.info(tr("digital_human.assets.customize_mode"))
|
|
st.button(
|
|
tr("btn.generate"),
|
|
type="primary",
|
|
use_container_width=True,
|
|
disabled=True,
|
|
key="digital_human_customize_disable"
|
|
)
|
|
return
|
|
|
|
# Generate button
|
|
if st.button(tr("btn.generate"), type="primary", use_container_width=True, key="digital_human_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:
|
|
# Define async generation function
|
|
async def generate_digital_human_video():
|
|
task_dir, task_id = create_task_output_dir()
|
|
workflow_path = video_params["workflow_path"]
|
|
api_video_workflow = workflow_path.get("api_video_workflow")
|
|
api_video_params = dict(workflow_path.get("api_video_params") or {})
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
async def generate_tts_reference(text: str) -> str:
|
|
audio_path = os.path.join(task_dir, "narration.mp3")
|
|
tts_inference_mode = video_params.get("tts_inference_mode", "local")
|
|
tts_voice = video_params.get("tts_voice")
|
|
tts_speed = video_params.get("tts_speed")
|
|
tts_workflow = video_params.get("tts_workflow")
|
|
ref_audio = video_params.get("ref_audio")
|
|
|
|
tts_kwargs = {
|
|
"text": text,
|
|
"output_path": audio_path,
|
|
"inference_mode": tts_inference_mode,
|
|
}
|
|
if tts_inference_mode == "local":
|
|
tts_kwargs["voice"] = tts_voice
|
|
tts_kwargs["speed"] = tts_speed
|
|
elif tts_inference_mode == "comfyui":
|
|
if tts_workflow:
|
|
tts_kwargs["workflow"] = tts_workflow
|
|
if ref_audio:
|
|
tts_kwargs["ref_audio"] = ref_audio
|
|
|
|
await pixelle_video.tts(**tts_kwargs)
|
|
return audio_path
|
|
|
|
async def generate_api_digital_human(text: str) -> str:
|
|
status_text.text(tr("progress.step_audio"))
|
|
progress_bar.progress(25)
|
|
audio_path = await generate_tts_reference(text)
|
|
|
|
reference_image_paths = [character_assets[0]]
|
|
if mode == "digital" and goods_assets:
|
|
reference_image_paths.append(goods_assets[0])
|
|
|
|
subject_prompt = (
|
|
"参考图1中的人物面对镜头自然口播。"
|
|
if get_language() == "zh_CN"
|
|
else "The person in reference image 1 speaks naturally to camera."
|
|
)
|
|
if mode == "digital" and goods_assets:
|
|
subject_prompt += (
|
|
"结合参考图2中的商品，生成竖屏商业口播视频。"
|
|
if get_language() == "zh_CN"
|
|
else "Use the product in reference image 2 and create a vertical product-promotion talking video."
|
|
)
|
|
prompt = f"{subject_prompt} 口播文案：{text}"
|
|
|
|
final_video_path = os.path.join(task_dir, "final.mp4")
|
|
duration = int(api_video_params.pop("duration", 5))
|
|
media_params = {
|
|
**api_video_params,
|
|
"prompt": prompt,
|
|
"workflow": api_video_workflow,
|
|
"media_type": "video",
|
|
"output_path": final_video_path,
|
|
"duration": duration,
|
|
"reference_image_paths": reference_image_paths,
|
|
"reference_audio_path": audio_path,
|
|
"audio": True,
|
|
"video_ratio": api_video_params.get("video_ratio", "9:16"),
|
|
}
|
|
progress_bar.progress(60)
|
|
status_text.text(tr("progress.generation"))
|
|
media_result = await pixelle_video.media(**media_params)
|
|
progress_bar.progress(100)
|
|
status_text.text(tr("status.success"))
|
|
return media_result.url
|
|
|
|
if api_video_workflow:
|
|
if mode == "customize":
|
|
generated_text = goods_text
|
|
elif goods_text and goods_text.strip():
|
|
generated_text = goods_text
|
|
else:
|
|
generated_text = await pixelle_video.llm(
|
|
prompt=(
|
|
f"请为商品“{goods_title}”写一段适合数字人口播短视频的中文推广文案。"
|
|
"要求自然、有吸引力，控制在80字以内，只输出文案正文。"
|
|
),
|
|
temperature=0.7,
|
|
max_tokens=300,
|
|
)
|
|
return await generate_api_digital_human(generated_text)
|
|
|
|
kit = await pixelle_video._get_or_create_comfykit()
|
|
|
|
if mode == "customize":
|
|
status_text.text(tr("progress.step_audio"))
|
|
progress_bar.progress(25)
|
|
generated_image_path = character_assets[0]
|
|
generated_text = goods_text
|
|
|
|
# TTS
|
|
audio_path = os.path.join(task_dir, "narration.mp3")
|
|
tts_inference_mode = video_params.get("tts_inference_mode", "local")
|
|
tts_voice = video_params.get("tts_voice")
|
|
tts_speed = video_params.get("tts_speed")
|
|
tts_workflow = video_params.get("tts_workflow")
|
|
ref_audio = video_params.get("ref_audio")
|
|
|
|
tts_kwargs = {
|
|
"text": generated_text,
|
|
"output_path": audio_path,
|
|
"inference_mode": tts_inference_mode
|
|
}
|
|
if tts_inference_mode == "local":
|
|
tts_kwargs["voice"] = tts_voice
|
|
tts_kwargs["speed"] = tts_speed
|
|
elif tts_inference_mode == "comfyui":
|
|
if tts_workflow:
|
|
tts_kwargs["workflow"] = tts_workflow
|
|
if ref_audio:
|
|
tts_kwargs["ref_audio"] = ref_audio
|
|
|
|
await pixelle_video.tts(**tts_kwargs)
|
|
progress_bar.progress(65)
|
|
status_text.text(tr("progress.concatenating"))
|
|
|
|
# Directly call the second workflow
|
|
second_workflow_path = Path(workflow_path.get("second_workflow_path"))
|
|
if not second_workflow_path.exists():
|
|
raise Exception(f"The second step workflow file does not exist:{second_workflow_path}")
|
|
with open(second_workflow_path, 'r', encoding='utf-8') as f:
|
|
second_workflow_config = json.load(f)
|
|
second_workflow_params = {
|
|
"videoimage": generated_image_path,
|
|
"audio": audio_path
|
|
}
|
|
if second_workflow_config.get("source") == "runninghub" and "workflow_id" in second_workflow_config:
|
|
workflow_input = second_workflow_config["workflow_id"]
|
|
else:
|
|
workflow_input = str(second_workflow_config)
|
|
second_result = await kit.execute(workflow_input, second_workflow_params)
|
|
# Video Link Extraction
|
|
generated_video_url = None
|
|
if hasattr(second_result, 'videos') and second_result.videos:
|
|
generated_video_url = second_result.videos[0]
|
|
elif hasattr(second_result, 'outputs') and second_result.outputs:
|
|
for node_id, node_output in second_result.outputs.items():
|
|
if isinstance(node_output, dict) and 'videos' in node_output:
|
|
videos = node_output['videos']
|
|
if videos and len(videos) > 0:
|
|
generated_video_url = videos[0]
|
|
break
|
|
if not generated_video_url:
|
|
raise Exception("The second step of the workflow did not return a video. Please check the workflow configuration.")
|
|
|
|
final_video_path = os.path.join(task_dir, "final.mp4")
|
|
timeout = httpx.Timeout(300.0)
|
|
async with httpx.AsyncClient(timeout=timeout) as client:
|
|
response = await client.get(generated_video_url)
|
|
response.raise_for_status()
|
|
with open(final_video_path, 'wb') as f:
|
|
f.write(response.content)
|
|
progress_bar.progress(100)
|
|
status_text.text(tr("status.success"))
|
|
return final_video_path
|
|
|
|
else:
|
|
#Initialization and parameter preparation
|
|
task_dir, task_id = create_task_output_dir()
|
|
logger.info(f"[Initialization] Task Directory: {task_dir}")
|
|
|
|
first_workflow_path = Path(workflow_path.get("first_workflow_path"))
|
|
third_workflow_path = Path(workflow_path.get("third_workflow_path"))
|
|
second_workflow_path = Path(workflow_path.get("second_workflow_path"))
|
|
api_image_workflow = workflow_path.get("api_image_workflow")
|
|
assert first_workflow_path.exists(), "The first_workflow file does not exist."
|
|
assert third_workflow_path.exists(), "The third_workflow file does not exist."
|
|
assert second_workflow_path.exists(), "The second_workflow file does not exist."
|
|
|
|
if goods_text and goods_text.strip():
|
|
generated_text = goods_text
|
|
|
|
status_text.text(tr("progress.step_image"))
|
|
if api_image_workflow:
|
|
image_prompt = (
|
|
f"Create a polished digital-human product promotion image. "
|
|
f"Use the first reference image as the person/character, the second reference image as the product, "
|
|
f"and make the scene suitable for a short spoken ad. Script: {goods_text}"
|
|
)
|
|
generated_image_path = os.path.join(task_dir, "generated_digital_image.png")
|
|
media_result = await pixelle_video.media(
|
|
prompt=image_prompt,
|
|
workflow=api_image_workflow,
|
|
media_type="image",
|
|
image_paths=[character_assets[0], goods_assets[0]],
|
|
output_path=generated_image_path,
|
|
width=1080,
|
|
height=1920,
|
|
)
|
|
generated_image_url = media_result.url
|
|
else:
|
|
workflow_path = third_workflow_path
|
|
workflow_params = {"firstimage": character_assets[0], "secondimage": goods_assets[0]}
|
|
kit = await pixelle_video._get_or_create_comfykit()
|
|
workflow_config = json.load(open(workflow_path, 'r', encoding='utf8'))
|
|
if workflow_config.get("source") == "runninghub" and "workflow_id" in workflow_config:
|
|
workflow_input = workflow_config["workflow_id"]
|
|
else:
|
|
workflow_input = str(workflow_config)
|
|
combine_image = await kit.execute(workflow_input, workflow_params)
|
|
if combine_image.status != "completed":
|
|
raise Exception(f"workflow execution failed: {combine_image.msg}")
|
|
generated_image_url = getattr(combine_image, "images", [None])[0]
|
|
status_text.text(tr("progress.step_audio"))
|
|
audio_path = os.path.join(task_dir, "narration.mp3")
|
|
tts_inference_mode = video_params.get("tts_inference_mode", "local")
|
|
tts_voice = video_params.get("tts_voice")
|
|
tts_speed = video_params.get("tts_speed")
|
|
tts_workflow = video_params.get("tts_workflow")
|
|
ref_audio = video_params.get("ref_audio")
|
|
|
|
tts_kwargs = {
|
|
"text": generated_text,
|
|
"output_path": audio_path,
|
|
"inference_mode": tts_inference_mode
|
|
}
|
|
if tts_inference_mode == "local":
|
|
tts_kwargs["voice"] = tts_voice
|
|
tts_kwargs["speed"] = tts_speed
|
|
elif tts_inference_mode == "comfyui":
|
|
if tts_workflow:
|
|
tts_kwargs["workflow"] = tts_workflow
|
|
if ref_audio:
|
|
tts_kwargs["ref_audio"] = ref_audio
|
|
|
|
await pixelle_video.tts(**tts_kwargs)
|
|
progress_bar.progress(65)
|
|
status_text.text(tr("progress.concatenating"))
|
|
|
|
if not second_workflow_path.exists():
|
|
raise Exception(f"The second step workflow file does not exist:{second_workflow_path}")
|
|
with open(second_workflow_path, 'r', encoding='utf-8') as f:
|
|
second_workflow_config = json.load(f)
|
|
second_workflow_params = {
|
|
"videoimage": generated_image_url,
|
|
"audio": audio_path
|
|
}
|
|
if second_workflow_config.get("source") == "runninghub" and "workflow_id" in second_workflow_config:
|
|
workflow_input = second_workflow_config["workflow_id"]
|
|
else:
|
|
workflow_input = str(second_workflow_config)
|
|
second_result = await kit.execute(workflow_input, second_workflow_params)
|
|
# Video Link Extraction
|
|
generated_video_url = None
|
|
if hasattr(second_result, 'videos') and second_result.videos:
|
|
generated_video_url = second_result.videos[0]
|
|
elif hasattr(second_result, 'outputs') and second_result.outputs:
|
|
for node_id, node_output in second_result.outputs.items():
|
|
if isinstance(node_output, dict) and 'videos' in node_output:
|
|
videos = node_output['videos']
|
|
if videos and len(videos) > 0:
|
|
generated_video_url = videos[0]
|
|
break
|
|
if not generated_video_url:
|
|
raise Exception("The second step of the workflow did not return a video. Please check the workflow configuration.")
|
|
|
|
final_video_path = os.path.join(task_dir, "final.mp4")
|
|
timeout = httpx.Timeout(300.0)
|
|
async with httpx.AsyncClient(timeout=timeout) as client:
|
|
response = await client.get(generated_video_url)
|
|
response.raise_for_status()
|
|
with open(final_video_path, 'wb') as f:
|
|
f.write(response.content)
|
|
progress_bar.progress(100)
|
|
status_text.text(tr("status.success"))
|
|
return final_video_path
|
|
|
|
else:
|
|
status_text.text(tr("progress.step_image"))
|
|
if api_image_workflow:
|
|
image_prompt = (
|
|
f"Create a polished digital-human product promotion image for '{goods_title}'. "
|
|
f"Use the first reference image as the person/character and the second reference image as the product. "
|
|
f"Make it vertical, clean, commercial, and suitable for a spoken short video."
|
|
)
|
|
generated_image_path = os.path.join(task_dir, "generated_digital_image.png")
|
|
media_result = await pixelle_video.media(
|
|
prompt=image_prompt,
|
|
workflow=api_image_workflow,
|
|
media_type="image",
|
|
image_paths=[character_assets[0], goods_assets[0]],
|
|
output_path=generated_image_path,
|
|
width=1080,
|
|
height=1920,
|
|
)
|
|
generated_image_url = media_result.url
|
|
generated_text = await pixelle_video.llm(
|
|
prompt=(
|
|
f"请为商品“{goods_title}”写一段适合数字人口播短视频的中文推广文案。"
|
|
"要求自然、有吸引力，控制在80字以内，只输出文案正文。"
|
|
),
|
|
temperature=0.7,
|
|
max_tokens=300,
|
|
)
|
|
else:
|
|
workflow_path = first_workflow_path
|
|
workflow_params = {"firstimage": character_assets[0], "secondimage": goods_assets[0], "goodstype": goods_title}
|
|
kit = await pixelle_video._get_or_create_comfykit()
|
|
workflow_config = json.load(open(workflow_path, 'r', encoding='utf8'))
|
|
if workflow_config.get("source") == "runninghub" and "workflow_id" in workflow_config:
|
|
workflow_input = workflow_config["workflow_id"]
|
|
else:
|
|
workflow_input = str(workflow_config)
|
|
synthesis_result = await kit.execute(workflow_input, workflow_params)
|
|
if synthesis_result.status != "completed":
|
|
raise Exception(f"workflow execution failed: {synthesis_result.msg}")
|
|
generated_image_url = getattr(synthesis_result, "images", [None])[0]
|
|
generated_text = getattr(synthesis_result, "texts", [None])[0]
|
|
|
|
status_text.text(tr("progress.step_audio"))
|
|
audio_path = os.path.join(task_dir, "narration.mp3")
|
|
tts_inference_mode = video_params.get("tts_inference_mode", "local")
|
|
tts_voice = video_params.get("tts_voice")
|
|
tts_speed = video_params.get("tts_speed")
|
|
tts_workflow = video_params.get("tts_workflow")
|
|
ref_audio = video_params.get("ref_audio")
|
|
|
|
tts_kwargs = {
|
|
"text": generated_text,
|
|
"output_path": audio_path,
|
|
"inference_mode": tts_inference_mode
|
|
}
|
|
if tts_inference_mode == "local":
|
|
tts_kwargs["voice"] = tts_voice
|
|
tts_kwargs["speed"] = tts_speed
|
|
elif tts_inference_mode == "comfyui":
|
|
if tts_workflow:
|
|
tts_kwargs["workflow"] = tts_workflow
|
|
if ref_audio:
|
|
tts_kwargs["ref_audio"] = ref_audio
|
|
|
|
await pixelle_video.tts(**tts_kwargs)
|
|
progress_bar.progress(65)
|
|
status_text.text(tr("progress.concatenating"))
|
|
|
|
if not second_workflow_path.exists():
|
|
raise Exception(f"The second step workflow file does not exist:{second_workflow_path}")
|
|
with open(second_workflow_path, 'r', encoding='utf-8') as f:
|
|
second_workflow_config = json.load(f)
|
|
second_workflow_params = {
|
|
"videoimage": generated_image_url,
|
|
"audio": audio_path
|
|
}
|
|
if second_workflow_config.get("source") == "runninghub" and "workflow_id" in second_workflow_config:
|
|
workflow_input = second_workflow_config["workflow_id"]
|
|
else:
|
|
workflow_input = str(second_workflow_config)
|
|
second_result = await kit.execute(workflow_input, second_workflow_params)
|
|
# Video Link Extraction
|
|
generated_video_url = None
|
|
if hasattr(second_result, 'videos') and second_result.videos:
|
|
generated_video_url = second_result.videos[0]
|
|
elif hasattr(second_result, 'outputs') and second_result.outputs:
|
|
for node_id, node_output in second_result.outputs.items():
|
|
if isinstance(node_output, dict) and 'videos' in node_output:
|
|
videos = node_output['videos']
|
|
if videos and len(videos) > 0:
|
|
generated_video_url = videos[0]
|
|
break
|
|
if not generated_video_url:
|
|
raise Exception("The second step of the workflow did not return a video. Please check the workflow configuration.")
|
|
|
|
final_video_path = os.path.join(task_dir, "final.mp4")
|
|
timeout = httpx.Timeout(300.0)
|
|
async with httpx.AsyncClient(timeout=timeout) as client:
|
|
response = await client.get(generated_video_url)
|
|
response.raise_for_status()
|
|
with open(final_video_path, 'wb') as f:
|
|
f.write(response.content)
|
|
progress_bar.progress(100)
|
|
status_text.text(tr("status.success"))
|
|
return final_video_path
|
|
|
|
# Execute async generation
|
|
final_video_path = run_async(generate_digital_human_video())
|
|
run_async(save_web_generation_history(
|
|
pixelle_video,
|
|
task_id=Path(final_video_path).parent.name,
|
|
video_path=final_video_path,
|
|
pipeline="digital_human",
|
|
title="数字人口播" if get_language() == "zh_CN" else "Digital Human",
|
|
input_params={
|
|
"text": goods_text or goods_title,
|
|
"mode": mode,
|
|
"goods_title": goods_title,
|
|
"goods_text": goods_text,
|
|
"character_assets": character_assets,
|
|
"goods_assets": goods_assets,
|
|
"workflow_path": video_params.get("workflow_path"),
|
|
"tts_voice": video_params.get("tts_voice"),
|
|
"tts_speed": video_params.get("tts_speed"),
|
|
"tts_inference_mode": video_params.get("tts_inference_mode"),
|
|
},
|
|
))
|
|
|
|
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=final_video_path))
|
|
|
|
st.markdown("---")
|
|
|
|
# Video info
|
|
if os.path.exists(final_video_path):
|
|
file_size_mb = os.path.getsize(final_video_path) / (1024 * 1024)
|
|
|
|
info_text = (
|
|
f"⏱️ {tr('info.generation_time')} {total_time:.1f}s "
|
|
f"📦 {file_size_mb:.2f}MB"
|
|
)
|
|
st.caption(info_text)
|
|
|
|
st.markdown("---")
|
|
|
|
# Video preview
|
|
st.video(final_video_path)
|
|
|
|
# Download button
|
|
with open(final_video_path, "rb") as video_file:
|
|
video_bytes = video_file.read()
|
|
video_filename = os.path.basename(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=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(DigitalHumanPipelineUI)
|