app.py

import glob
import os
import time
import uuid
from pathlib import Path

import gradio as gr

from cli import ProxyRotator, VeoEngine

OUTPUT_DIR = Path("outputs")
PROXY_FILE = Path("proxies-res-any-1777564385571.txt")


def cleanup_outputs(max_age_seconds=3600):
    OUTPUT_DIR.mkdir(exist_ok=True)
    now = time.time()
    for path in glob.glob(str(OUTPUT_DIR / "*.mp4")):
        try:
            if os.path.getmtime(path) < now - max_age_seconds:
                os.remove(path)
        except OSError:
            pass


def build_engine(model_key="veo"):
    proxy_file = str(PROXY_FILE) if PROXY_FILE.exists() else str(PROXY_FILE)
    rotator = ProxyRotator(proxy_file)
    return VeoEngine(rotator, model_key=model_key)


def generate_video(prompt, model_key, prompt_mode, aspect_ratio, audio_mode, require_audio, max_retries):
    prompt = (prompt or "").strip()
    if not prompt:
        return None, "請先輸入 Prompt。"

    cleanup_outputs()
    max_retries = int(max_retries or 1)
    last_error = None

    for attempt in range(1, max_retries + 1):
        try:
            engine = build_engine(model_key)

            yield None, f"Attempt {attempt}/{max_retries}\n正在取得 session nonce..."
            engine.fetch_nonce()

            if prompt_mode == "enhanced":
                yield None, f"Attempt {attempt}/{max_retries}\n正在增強 Prompt..."
                generation_prompt = engine.enhance_prompt(prompt)
            else:
                generation_prompt = prompt
                yield None, f"Attempt {attempt}/{max_retries}\n使用原始 Prompt 生成，避免提示語被改寫..."

            yield None, f"Attempt {attempt}/{max_retries}\n正在提交影片生成任務..."
            scene_id = engine.generate_video(generation_prompt, aspect_ratio, audio_mode=audio_mode)

            yield None, (
                f"Attempt {attempt}/{max_retries}\n"
                f"Scene ID: {scene_id}\n"
                f"伺服器正在生成影片，通常需要 1-5 分鐘..."
            )
            video_url = engine.wait_for_video(scene_id)
            if not video_url:
                raise RuntimeError("生成完成檢查失敗，沒有取得影片 URL。")

            filename = OUTPUT_DIR / f"veo_video_{scene_id}_{uuid.uuid4().hex[:8]}.mp4"
            yield None, (
                f"Attempt {attempt}/{max_retries}\n"
                f"Scene ID: {scene_id}\n"
                f"已取得影片 URL，正在下載...\n{video_url}"
            )
            saved_file = engine.download(video_url, str(filename))

            has_audio = engine.has_audio_track(saved_file)
            if require_audio and not has_audio:
                raise RuntimeError("已下載影片，但沒有偵測到音軌。")

            status = (
                "生成完成！\n\n"
                f"Model: {engine.MODEL_NAME}\n"
                f"Referrer: {engine.PAGE_URL}\n"
                f"Scene ID: {scene_id}\n"
                f"Aspect Ratio: {aspect_ratio}\n"
                f"Audio Mode: {audio_mode}\n"
                f"Audio Detected: {'Yes' if has_audio else 'No'}\n"
                f"Video URL: {video_url}\n\n"
                f"Prompt Mode: {prompt_mode}\n"
                f"Generation Prompt:\n{generation_prompt}"
            )
            yield saved_file, status
            return

        except Exception as exc:
            last_error = exc
            if attempt < max_retries:
                yield None, f"Attempt {attempt}/{max_retries} 失敗：{exc}\n準備重試..."
                time.sleep(2)
            else:
                yield None, f"全部重試失敗。\n最後錯誤：{last_error}"
                return


def create_demo():
    with gr.Blocks(title="Veo Video Generator") as demo:
        gr.Markdown(
            "# 🎬 AI Video Generator\n"
            "**Models: Veo 3.1 / Grok 4.5**\n\n"
            "Hugging Face Spaces 免費版 UI。選擇模型、輸入 Prompt 後會生成影片、下載 MP4，並檢查是否包含音軌。"
        )

        with gr.Row():
            with gr.Column(scale=1):
                prompt = gr.Textbox(
                    label="Prompt",
                    lines=5,
                    placeholder="A cinematic shot of a futuristic city at sunset",
                )
                model_key = gr.Radio(
                    choices=[("Veo 3.1", "veo"), ("Grok 4.5", "grok")],
                    value="veo",
                    label="Model",
                    info="Veo 3.1 使用 /veo-video-generator/；Grok 4.5 使用 /grok-ai-video-generator/。",
                )
                prompt_mode = gr.Radio(
                    choices=[("原始提示語（推薦，生成更一致）", "raw"), ("AI 增強提示語", "enhanced")],
                    value="raw",
                    label="Prompt Mode",
                    info="raw 會直接用你輸入的 Prompt 生成；enhanced 會先讓網站改寫 Prompt，畫面可能和原提示不同。",
                )
                aspect_ratio = gr.Radio(
                    choices=["16:9", "9:16"],
                    value="16:9",
                    label="Aspect Ratio",
                )
                audio_mode = gr.Radio(
                    choices=["auto", "try", "none"],
                    value="auto",
                    label="Audio Mode",
                    info="auto 使用網站預設；try 會附加實驗音效參數；none 不附加音效偏好。",
                )
                require_audio = gr.Checkbox(
                    value=False,
                    label="Require Audio",
                    info="啟用後，如果下載影片沒有偵測到音軌，任務會標記為失敗。",
                )
                max_retries = gr.Slider(
                    minimum=1,
                    maximum=3,
                    step=1,
                    value=1,
                    label="Max Retries",
                )
                generate_btn = gr.Button("Generate Video", variant="primary")

            with gr.Column(scale=1):
                video = gr.Video(label="Generated Video")
                status = gr.Textbox(label="Status", lines=14)

        gr.Markdown(
            "## 使用提示\n"
            "- 免費 Space 可能會休眠，首次啟動較慢。\n"
            "- 一次只建議生成一個任務。\n"
            "- 影片會暫存在 Space 的 outputs 目錄，重啟後可能消失。\n"
            "- 若設為 Public Space，任何人都可能使用你的介面。"
        )

        generate_btn.click(
            fn=generate_video,
            inputs=[prompt, model_key, prompt_mode, aspect_ratio, audio_mode, require_audio, max_retries],
            outputs=[video, status],
        )

    return demo


if __name__ == "__main__":
    cleanup_outputs()
    app = create_demo()
    app.queue(default_concurrency_limit=1).launch(theme=gr.themes.Soft())