Update app.py

app.py

@@ -1,192 +1,179 @@
+import gradio as gr
 import os
-import time
 import tempfile
 import shutil
-from pathlib import Path
 from typing import Optional, Tuple, Union
-
-import gradio as gr
 from huggingface_hub import InferenceClient, whoami
+from pathlib import Path
 
-# =========================
-# Inference client (fal-ai)
-# =========================
+# Initialize Hugging Face Inference Client with fal-ai provider
 client = InferenceClient(
     provider="fal-ai",
     api_key=os.environ.get("HF_TOKEN"),
     bill_to="huggingface",
 )
 
-# =========================
-# Auth / PRO helpers
-# =========================
 def verify_pro_status(token: Optional[Union[gr.OAuthToken, str]]) -> bool:
     """Verifies if the user is a Hugging Face PRO user or part of an enterprise org."""
     if not token:
         return False
-
+    
     if isinstance(token, gr.OAuthToken):
         token_str = token.token
     elif isinstance(token, str):
         token_str = token
     else:
         return False
-
+    
     try:
         user_info = whoami(token=token_str)
         return (
-            user_info.get("isPro", False)
-            or any(org.get("isEnterprise", False) for org in user_info.get("orgs", []))
+            user_info.get("isPro", False) or
+            any(org.get("isEnterprise", False) for org in user_info.get("orgs", []))
         )
     except Exception as e:
         print(f"Could not verify user's PRO/Enterprise status: {e}")
         return False
 
-# =========================
-# Storage hygiene
-# =========================
 def cleanup_temp_files():
-    """Clean up old temporary .mp4 files to prevent storage overflow."""
+    """Clean up old temporary video files to prevent storage overflow."""
     try:
         temp_dir = tempfile.gettempdir()
+        # Clean up old .mp4 files in temp directory
         for file_path in Path(temp_dir).glob("*.mp4"):
             try:
                 # Remove files older than 5 minutes
+                import time
                 if file_path.stat().st_mtime < (time.time() - 300):
                     file_path.unlink(missing_ok=True)
             except Exception:
-                pass  # Ignore errors for individual files
+                pass
     except Exception as e:
         print(f"Cleanup error: {e}")
 
-def _write_video_bytes_to_tempfile(video_bytes: bytes) -> str:
-    temp_file = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
-    try:
-        temp_file.write(video_bytes)
-        temp_file.flush()
-        return temp_file.name
-    finally:
-        temp_file.close()
-
-# =========================
-# Generation (Text → Video)
-# =========================
 def generate_video(
     prompt: str,
     duration: int = 8,
     size: str = "1280x720",
-    api_key: Optional[str] = None,
+    api_key: Optional[str] = None
 ) -> Tuple[Optional[str], str]:
-    """
-    Generate video using Sora-2 via Hugging Face Inference API (fal-ai provider).
-    Returns (video_path, status_message).
-    """
+    """Generate video using Sora-2 through Hugging Face Inference API with fal-ai provider."""
     cleanup_temp_files()
-
     try:
-        # Use provided API key or environment variable
-        temp_client = (
-            InferenceClient(provider="fal-ai", api_key=api_key, bill_to="huggingface")
-            if api_key
-            else client
-        )
-        if not (api_key or os.environ.get("HF_TOKEN")):
-            return None, "❌ Please set HF_TOKEN environment variable."
-
-        # Call text-to-video
+        if api_key:
+            temp_client = InferenceClient(
+                provider="fal-ai",
+                api_key=api_key,
+                bill_to="huggingface",
+            )
+        else:
+            temp_client = client
+            if not os.environ.get("HF_TOKEN") and not api_key:
+                return None, "❌ Please set HF_TOKEN environment variable."
+        
         video_bytes = temp_client.text_to_video(
             prompt,
             model="akhaliq/sora-2",
-            # If your backend supports these, you can forward them as kwargs:
-            # duration=duration,
-            # size=size,
         )
-
-        video_path = _write_video_bytes_to_tempfile(video_bytes)
+        
+        temp_file = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
+        try:
+            temp_file.write(video_bytes)
+            temp_file.flush()
+            video_path = temp_file.name
+        finally:
+            temp_file.close()
+        
         return video_path, "✅ Video generated successfully!"
     except Exception as e:
         return None, f"❌ Error generating video: {str(e)}"
 
-# =========================
-# Generation (Image → Video)
-# =========================
+# --- NEW: image -> video support ---
 def generate_video_from_image(
+    image: Union[str, bytes],
     prompt: str,
-    image_path: str,
-    api_key: Optional[str] = None,
+    api_key: Optional[str] = None
 ) -> Tuple[Optional[str], str]:
-    """
-    Generate video from a single input image using Sora-2 image-to-video.
-    Returns (video_path, status_message).
-    """
+    """Generate a video from a single input image + prompt using Sora-2 image-to-video."""
     cleanup_temp_files()
-
-    if not image_path or not Path(image_path).exists():
-        return None, "❌ Please upload an image."
-
+    if not prompt or prompt.strip() == "":
+        return None, "❌ Please enter a prompt"
     try:
-        temp_client = (
-            InferenceClient(provider="fal-ai", api_key=api_key, bill_to="huggingface")
-            if api_key
-            else client
-        )
-        if not (api_key or os.environ.get("HF_TOKEN")):
-            return None, "❌ Please set HF_TOKEN environment variable."
-
-        with open(image_path, "rb") as f:
-            input_image = f.read()
+        if api_key:
+            temp_client = InferenceClient(
+                provider="fal-ai",
+                api_key=api_key,
+                bill_to="huggingface",
+            )
+        else:
+            temp_client = client
+            if not os.environ.get("HF_TOKEN") and not api_key:
+                return None, "❌ Please set HF_TOKEN environment variable."
+
+        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, "❌ Invalid image input. Please upload an image."
 
         video_bytes = temp_client.image_to_video(
             input_image,
-            prompt=prompt or "",
+            prompt=prompt,
             model="akhaliq/sora-2-image-to-video",
         )
 
-        video_path = _write_video_bytes_to_tempfile(video_bytes)
-        return video_path, "✅ Video generated successfully from image!"
+        temp_file = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
+        try:
+            temp_file.write(video_bytes)
+            temp_file.flush()
+            video_path = temp_file.name
+        finally:
+            temp_file.close()
+
+        return video_path, "✅ Video generated from image successfully!"
     except Exception as e:
         return None, f"❌ Error generating video from image: {str(e)}"
 
-# =========================
-# PRO wrapper (uses request)
-# =========================
 def generate_with_pro_auth(
-    mode: str,
-    prompt: str,
-    image_path: Optional[str],
-    request: gr.Request,
+    prompt: str, 
+    oauth_token: Optional[gr.OAuthToken] = None
 ) -> Tuple[Optional[str], str]:
-    """
-    Check PRO status from the request's OAuth token, then route to the
-    appropriate generation function based on mode.
-    """
-    oauth_token = getattr(request, "oauth_token", None)
+    """Wrapper function that checks if user is PRO before generating video."""
     if not verify_pro_status(oauth_token):
-        raise gr.Error(
-            "Access Denied. This app is exclusively for Hugging Face PRO users. Please subscribe to PRO to use this app."
-        )
-
-    if mode == "Text → Video":
-        if not prompt or not prompt.strip():
-            return None, "❌ Please enter a prompt."
-        return generate_video(prompt, duration=8, size="1280x720", api_key=None)
+        raise gr.Error("Access Denied. This app is exclusively for Hugging Face PRO users.")
+    
+    if not prompt or prompt.strip() == "":
+        return None, "❌ Please enter a prompt"
+    
+    return generate_video(
+        prompt, 
+        duration=8, 
+        size="1280x720", 
+        api_key=None
+    )
 
-    # Image → Video
+# --- NEW: PRO-gated wrapper for image -> video ---
+def generate_with_pro_auth_image(
+    prompt: str,
+    image_path: Optional[str] = None,
+    oauth_token: Optional[gr.OAuthToken] = None
+) -> Tuple[Optional[str], str]:
+    """Checks PRO status then calls image->video generator."""
+    if not verify_pro_status(oauth_token):
+        raise gr.Error("Access Denied. This app is exclusively for Hugging Face PRO users.")
     if not image_path:
-        return None, "❌ Please upload an image."
-    # Prompt is optional for image→video; pass empty string if not provided
-    return generate_video_from_image(prompt or "", image_path, api_key=None)
+        return None, "❌ Please upload an image"
+    return generate_video_from_image(image=image_path, prompt=prompt, api_key=None)
 
 def simple_generate(prompt: str) -> Optional[str]:
-    """Examples: only return the video path (text→video)."""
-    if not prompt or not prompt.strip():
+    """Simplified wrapper for examples that only returns video."""
+    if not prompt or prompt.strip() == "":
         return None
     video_path, _ = generate_video(prompt, duration=8, size="1280x720", api_key=None)
     return video_path
 
-# =========================
-# UI
-# =========================
 def create_ui():
     css = '''
     .logo-dark{display: none}
@@ -204,7 +191,7 @@ def create_ui():
         margin-left: 8px;
     }
     '''
-
+    
     with gr.Blocks(title="Sora-2 Text-to-Video Generator", theme=gr.themes.Soft(), css=css) as demo:
         gr.HTML("""
             <div style="text-align: center; max-width: 800px; margin: 0 auto;">
@@ -222,182 +209,87 @@ def create_ui():
                 </p>
             </div>
         """)
-
-        # HF OAuth (Spaces must have hf_oauth: true)
+        
         gr.LoginButton()
-
-        # Hidden by default; we’ll toggle based on PRO status
         pro_message = gr.Markdown(visible=False)
         main_interface = gr.Column(visible=False)
-
+        
         with main_interface:
-            gr.HTML("""
-                <div style="text-align: center; margin: 20px 0;">
-                    <p style="color: #28a745; font-weight: bold;">✨ Welcome PRO User! You have full access to Sora-2.</p>
-                </div>
-            """)
-
+            gr.HTML("""<div style="text-align: center; margin: 20px 0;">
+                <p style="color: #28a745; font-weight: bold;">✨ Welcome PRO User! You have full access to Sora-2.</p>
+            </div>""")
+            
+            # Text -> Video
             with gr.Row():
                 with gr.Column(scale=1):
-                    mode_radio = gr.Radio(
-                        choices=["Text → Video", "Image → Video"],
-                        value="Text → Video",
-                        label="Mode",
-                    )
                     prompt_input = gr.Textbox(
-                        label="Prompt",
-                        placeholder="Describe the video you want to create (optional for image→video)...",
-                        lines=4,
+                        label="Enter your prompt",
+                        placeholder="Describe the video you want to create...",
+                        lines=4
                     )
-
-                    image_group = gr.Group(visible=False)
-                    with image_group:
-                        image_input = gr.Image(
-                            label="Input Image (for Image → Video)",
-                            type="filepath",
-                            sources=["upload", "clipboard"],
-                            image_mode="RGB",
-                        )
-
-                    with gr.Accordion("Advanced Settings", open=False):
-                        gr.Markdown("*Coming soon: Duration and resolution controls*")
-
                     generate_btn = gr.Button("🎥 Generate Video", variant="primary", size="lg")
-
                 with gr.Column(scale=1):
-                    video_output = gr.Video(
-                        label="Generated Video",
-                        height=400,
-                        interactive=False,
-                        show_download_button=True,
-                    )
-                    status_output = gr.Textbox(
-                        label="Status",
-                        interactive=False,
-                        visible=True,
-                    )
-
-            # Examples (text→video only)
-            gr.Examples(
-                examples=[
-                    "A serene beach at sunset with waves gently rolling onto the shore",
-                    "A butterfly emerging from its chrysalis in slow motion",
-                    "Northern lights dancing across a starry night sky",
-                    "A bustling city street transitioning from day to night in timelapse",
-                    "A close-up of coffee being poured into a cup with steam rising",
-                    "Cherry blossoms falling in slow motion in a Japanese garden",
-                ],
-                inputs=prompt_input,
-                outputs=video_output,
-                fn=simple_generate,
-                cache_examples=False,
-                api_name=False,
-                show_api=False,
-            )
-
-            # Toggle image upload visibility with mode
-            def _toggle_image_group(mode: str):
-                return gr.update(visible=(mode == "Image → Video"))
-
-            mode_radio.change(
-                _toggle_image_group,
-                inputs=[mode_radio],
-                outputs=[image_group],
-                show_progress=False,
-            )
-
-            # Generation handler (uses request to read OAuth token)
+                    video_output = gr.Video(label="Generated Video", height=400, interactive=False, show_download_button=True)
+                    status_output = gr.Textbox(label="Status", interactive=False, visible=True)
+            
             generate_btn.click(
                 fn=generate_with_pro_auth,
-                inputs=[mode_radio, prompt_input, image_input],
+                inputs=[prompt_input],
                 outputs=[video_output, status_output],
-                queue=False,
-                api_name=False,
-                show_api=False,
+                queue=False
             )
 
-            # Footer
+            # --- NEW: Image -> Video UI ---
             gr.HTML("""
-                <div style="text-align: center; margin-top: 40px; padding: 20px; border-top: 1px solid #e0e0e0;">
-                    <h3 style="color: #667eea;">Thank you for being a PRO user! 🤗</h3>
+                <div style="text-align: center; margin: 40px 0 10px;">
+                    <h3 style="margin-bottom: 8px;">🖼️ ➜ 🎬 Image → Video (beta)</h3>
+                    <p style="color:#666; margin:0;">Turn a single image into a short video with a guiding prompt.</p>
                 </div>
             """)
-
-        # Use request to check access on load
-        def control_access(request: gr.Request):
-            oauth_profile = getattr(request, "oauth_profile", None)
-            oauth_token = getattr(request, "oauth_token", None)
-
-            if not oauth_profile:
-                # Not logged in
+            with gr.Row():
+                with gr.Column(scale=1):
+                    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,
+                    )
+                    image_input = gr.Image(label="Upload an image", type="filepath")
+                    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)
+                    status_output_img = gr.Textbox(label="Status", interactive=False, visible=True)
+
+            generate_img_btn.click(
+                fn=generate_with_pro_auth_image,
+                inputs=[img_prompt_input, image_input],
+                outputs=[video_output_img, status_output_img],
+                queue=False
+            )
+            
+            gr.HTML("""<div style="text-align: center; margin-top: 40px; padding: 20px; border-top: 1px solid #e0e0e0;">
+                <h3 style="color: #667eea;">Thank you for being a PRO user! 🤗</h3>
+            </div>""")
+        
+        def control_access(profile: Optional[gr.OAuthProfile] = None, oauth_token: Optional[gr.OAuthToken] = None):
+            if not profile:
                 return gr.update(visible=False), gr.update(visible=False)
-
             if verify_pro_status(oauth_token):
                 return gr.update(visible=True), gr.update(visible=False)
             else:
-                message = """
-                ## ✨ Exclusive Access for PRO Users
-
-                Thank you for your interest in the Sora-2 Text-to-Video Generator!
-
-                This advanced AI video generation tool is available exclusively for Hugging Face **PRO** members.
-
-                ### What you get with PRO:
-                - ✅ Unlimited access to Sora-2 video generation
-                - ✅ High-quality video outputs up to 1280x720
-                - ✅ Fast generation times with priority queue
-                - ✅ Access to other exclusive PRO Spaces
-                - ✅ Support the development of cutting-edge AI tools
-
-                ### Ready to create amazing videos?
-
-                <div style="text-align: center; margin: 30px 0;">
-                    <a href="http://huggingface.co/subscribe/pro?source=sora2_video" target="_blank" style="
-                        display: inline-block;
-                        background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
-                        color: white;
-                        padding: 12px 30px;
-                        border-radius: 25px;
-                        text-decoration: none;
-                        font-weight: bold;
-                        font-size: 1.1em;
-                        box-shadow: 0 4px 15px rgba(102, 126, 234, 0.3);
-                        transition: transform 0.2s;
-                    ">
-                        🚀 Become a PRO Today!
-                    </a>
-                </div>
-
-                <p style="text-align: center; color: #666; margin-top: 20px;">
-                    Join thousands of creators who are already using PRO tools to bring their ideas to life.
-                </p>
-                """
+                message = "## ✨ Exclusive Access for PRO Users\n\nThis tool is available exclusively for Hugging Face **PRO** members."
                 return gr.update(visible=False), gr.update(visible=True, value=message)
-
-        demo.load(
-            control_access,
-            inputs=None,
-            outputs=[main_interface, pro_message],
-        )
-
+        
+        demo.load(control_access, inputs=None, outputs=[main_interface, pro_message])
+    
     return demo
 
-# =========================
-# Entrypoint
-# =========================
 if __name__ == "__main__":
-    # Clean up any leftover files on startup
     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.launch(
-        show_api=False,
-        enable_monitoring=False,
-        quiet=True,
-        max_threads=10,
-    )
+    app.launch(show_api=False, enable_monitoring=False, quiet=True, max_threads=10)