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)
