app.py

import gradio as gr
import os
import tempfile
import shutil
from typing import Optional, Union
from pathlib import Path
from huggingface_hub import InferenceClient

# -------------------------
# Utilities
# -------------------------

def cleanup_temp_files():
    try:
        temp_dir = tempfile.gettempdir()
        for file_path in Path(temp_dir).glob("*.mp4"):
            try:
                import time
                if file_path.stat().st_mtime < (time.time() - 300):
                    file_path.unlink(missing_ok=True)
            except Exception:
                pass
    except Exception as e:
        print(f"Cleanup error: {e}")

def _client_from_token(token: Optional[str]) -> InferenceClient:
    if not token:
        raise gr.Error("Please sign in first. This app requires your Hugging Face login.")
    # IMPORTANT: do not set bill_to when using user OAuth tokens
    return InferenceClient(
        provider="fal-ai",
        api_key=token,
    )

def _save_bytes_as_temp_mp4(data: bytes) -> str:
    temp_file = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
    try:
        temp_file.write(data)
        temp_file.flush()
        return temp_file.name
    finally:
        temp_file.close()

# -------------------------
# Inference wrappers (no env fallback; always require LoginButton)
# -------------------------

def generate_video(
    prompt: str,
    token: gr.OAuthToken | None,
    duration: int = 8,          # kept for future use
    size: str = "1280x720",     # kept for future use
    *_                          # tolerate extra event payloads
) -> Optional[str]:
    if token is None or not getattr(token, "token", None):
        raise gr.Error("Sign in with Hugging Face to continue. This app uses your inference provider credits.")
    if not prompt or not prompt.strip():
        return None

    cleanup_temp_files()
    try:
        client = _client_from_token(token.token)
        # Ensure model id matches what users can access. Change if you intend provider repo.
        model_id = "akhaliq/sora-2"
        try:
            video_bytes = client.text_to_video(prompt, model=model_id)
        except Exception as e:
            # Provide a clearer message if this is an HTTP 403 from requests
            import requests
            if isinstance(e, requests.HTTPError) and getattr(e.response, "status_code", None) == 403:
                raise gr.Error(
                    "Access denied by provider (403). Make sure your HF account has credits/permission "
                    f"for provider 'fal-ai' and model '{model_id}'."
                )
            raise
        return _save_bytes_as_temp_mp4(video_bytes)
    except gr.Error:
        raise
    except Exception:
        raise gr.Error("Generation failed. If this keeps happening, check your provider quota or try again later.")

def generate_video_from_image(
    image: Union[str, bytes, None],
    prompt: str,
    token: gr.OAuthToken | None,
    *_
) -> Optional[str]:
    if token is None or not getattr(token, "token", None):
        raise gr.Error("Sign in with Hugging Face to continue. This app uses your inference provider credits.")
    if not image or not prompt or not prompt.strip():
        return None

    cleanup_temp_files()
    try:
        # Load image bytes
        if isinstance(image, str):
            with open(image, "rb") as f:
                input_image = f.read()
        elif isinstance(image, (bytes, bytearray)):
            input_image = image
        else:
            return None

        client = _client_from_token(token.token)
        model_id = "akhaliq/sora-2-image-to-video"
        try:
            video_bytes = client.image_to_video(
                input_image,
                prompt=prompt,
                model=model_id,
            )
        except Exception as e:
            import requests
            if isinstance(e, requests.HTTPError) and getattr(e.response, "status_code", None) == 403:
                raise gr.Error(
                    "Access denied by provider (403). Make sure your HF account has credits/permission "
                    f"for provider 'fal-ai' and model '{model_id}'."
                )
            raise
        return _save_bytes_as_temp_mp4(video_bytes)
    except gr.Error:
        raise
    except Exception:
        raise gr.Error("Generation failed. If this keeps happening, check your provider quota or try again later.")

# -------------------------
# UI
# -------------------------

def create_ui():
    css = '''
    .logo-dark{display: none}
    .dark .logo-dark{display: block !important}
    .dark .logo-light{display: none}
    #sub_title{margin-top: -20px !important}
    .notice {
        background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
        color: white;
        padding: 14px 16px;
        border-radius: 12px;
        margin: 18px auto 6px;
        max-width: 860px;
        text-align: center;
        font-size: 0.98rem;
    }
    '''

    with gr.Blocks(title="Sora-2 (uses your provider credits)", theme=gr.themes.Soft(), css=css) as demo:
        gr.HTML("""
            <div style="text-align:center; max-width:900px; margin:0 auto;">
                <h1 style="font-size:2.2em; margin-bottom:6px;">🎬 Sora-2</h1>
                <p style="color:#777; margin:0 0 8px;">Generate videos via the Hugging Face Inference API (provider: fal-ai)</p>
                <div class="notice">
                    <b>Heads up:</b> This is a paid app that uses <b>your</b> inference provider credits when you run generations.
                    Free users get <b>$0.10 in included credits</b>. <b>PRO users</b> get <b>$2 in included credits</b> 
                    and can continue using beyond that (with billing). 
                    <a href='http://huggingface.co/subscribe/pro?source=sora_2' target='_blank' style='color:#fff; text-decoration:underline; font-weight:bold;'>Subscribe to PRO</a> 
                    for more credits. Please sign in with your Hugging Face account to continue.
                </div>
                <p style="font-size: 0.9em; color: #999; margin-top: 10px;">
                    Built with <a href="https://huggingface.co/spaces/akhaliq/anycoder" target="_blank" style="color:#fff; text-decoration:underline;">anycoder</a>
                </p>
            </div>
        """)

        login_btn = gr.LoginButton("Sign in with Hugging Face")

        # Text -> Video
        with gr.Row():
            with gr.Column(scale=1):
                prompt_input = gr.Textbox(
                    label="Enter your prompt",
                    placeholder="Describe the video you want to create…",
                    lines=4,
                    elem_id="prompt-text-input"
                )
                generate_btn = gr.Button("🎥 Generate Video", variant="primary")
            with gr.Column(scale=1):
                video_output = gr.Video(
                    label="Generated Video",
                    height=400,
                    interactive=False,
                    show_download_button=True,
                    elem_id="text-to-video"
                )

        # Order of inputs: prompt, token
        generate_btn.click(
            fn=generate_video,
            inputs=[prompt_input, login_btn],
            outputs=[video_output],
        )

        # Image -> Video
        gr.HTML("""
            <div style="text-align:center; margin: 34px 0 10px;">
                <h3 style="margin-bottom:6px;">🖼️ ➜ 🎬 Image → Video (beta)</h3>
                <p style="color:#666; margin:0;">Turn a single image into a short video with a guiding prompt.</p>
            </div>
        """)
        with gr.Row():
            with gr.Column(scale=1):
                image_input = gr.Image(label="Upload an image", type="filepath")
                img_prompt_input = gr.Textbox(
                    label="Describe how the scene should evolve",
                    placeholder="e.g., The cat starts to dance and spins playfully",
                    lines=3,
                    elem_id="img-prompt-text-input"
                )
                generate_img_btn = gr.Button("🎥 Generate from Image", variant="primary")
            with gr.Column(scale=1):
                video_output_img = gr.Video(
                    label="Generated Video (from Image)",
                    height=400,
                    interactive=False,
                    show_download_button=True,
                    elem_id="image-to-video"
                )

        # Order of inputs: image, prompt, token
        generate_img_btn.click(
            fn=generate_video_from_image,
            inputs=[image_input, img_prompt_input, login_btn],
            outputs=[video_output_img],
        )

        # Examples: keep UI-only (no automatic inference to avoid accidental charges)
        gr.Examples(
            examples=[["A majestic golden eagle soaring through a vibrant sunset sky"]],
            inputs=prompt_input
        )

    return demo

# -------------------------
# Entrypoint
# -------------------------

if __name__ == "__main__":
    try:
        cleanup_temp_files()
        if os.path.exists("gradio_cached_examples"):
            shutil.rmtree("gradio_cached_examples", ignore_errors=True)
    except Exception as e:
        print(f"Initial cleanup error: {e}")

    app = create_ui()
    app.queue(status_update_rate="auto", api_open=False, default_concurrency_limit=None)
    app.launch(show_api=False, enable_monitoring=False, quiet=True, ssr_mode=True)