app.py

#!/usr/bin/env python3
"""
AI Video Generator with Gradio
Single file application - app.py
"""

import os
import gradio as gr
import replicate
import base64
from PIL import Image, ImageDraw, ImageFont
import io
import requests
from datetime import datetime
import tempfile
import time

# Try to import video processing libraries
try:
    import cv2
    import numpy as np
    VIDEO_PROCESSING_AVAILABLE = True
except ImportError:
    VIDEO_PROCESSING_AVAILABLE = False
    print("Warning: cv2 not available. Watermark feature will be disabled.")

# API token setup
api_token = os.getenv("RAPI_TOKEN")
if api_token:
    os.environ["REPLICATE_API_TOKEN"] = api_token

# Aspect ratio options
ASPECT_RATIOS = {
    "16:9": "16:9 (YouTube, Standard Video)",
    "4:3": "4:3 (Traditional TV Format)",
    "1:1": "1:1 (Instagram Feed)",
    "3:4": "3:4 (Instagram Portrait)",
    "9:16": "9:16 (Instagram Reels, TikTok)",
    "21:9": "21:9 (Cinematic Wide)",
    "9:21": "9:21 (Ultra Vertical)"
}

# Default prompts
DEFAULT_TEXT_PROMPT = ""
DEFAULT_IMAGE_PROMPT = "Generate a video with smooth and natural movement. Objects should have visible motion while maintaining fluid transitions."

def add_watermark_cv2(input_video_path, output_video_path):
    """Add watermark to video using OpenCV"""
    if not VIDEO_PROCESSING_AVAILABLE:
        return False
    
    try:
        # Open the video
        cap = cv2.VideoCapture(input_video_path)
        
        # Get video properties
        fps = int(cap.get(cv2.CAP_PROP_FPS))
        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
        
        # Define codec and create VideoWriter
        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
        out = cv2.VideoWriter(output_video_path, fourcc, fps, (width, height))
        
        # Watermark settings
        watermark_text = "ginigen.com"
        font = cv2.FONT_HERSHEY_SIMPLEX
        font_scale = max(0.4, height * 0.001)  # Scale based on video height
        font_thickness = max(1, int(height * 0.002))
        
        # Get text size
        (text_width, text_height), baseline = cv2.getTextSize(watermark_text, font, font_scale, font_thickness)
        
        # Position (bottom right with padding)
        padding = int(width * 0.02)
        x = width - text_width - padding
        y = height - padding
        
        # Process each frame
        while True:
            ret, frame = cap.read()
            if not ret:
                break
            
            # Add semi-transparent background for text
            overlay = frame.copy()
            cv2.rectangle(overlay, 
                         (x - 5, y - text_height - 5), 
                         (x + text_width + 5, y + 5), 
                         (0, 0, 0), 
                         -1)
            frame = cv2.addWeighted(frame, 0.7, overlay, 0.3, 0)
            
            # Add text
            cv2.putText(frame, watermark_text, (x, y), font, font_scale, (255, 255, 255), font_thickness, cv2.LINE_AA)
            
            # Write frame
            out.write(frame)
        
        # Release everything
        cap.release()
        out.release()
        cv2.destroyAllWindows()
        
        return True
        
    except Exception as e:
        print(f"Watermark error: {str(e)}")
        return False

def add_watermark_simple(input_video_path, output_video_path):
    """Simple fallback - just copy the video without watermark"""
    try:
        import shutil
        shutil.copy2(input_video_path, output_video_path)
        return False
    except Exception as e:
        print(f"Copy error: {str(e)}")
        return False

def add_watermark(input_video_path, output_video_path):
    """Add watermark to video - tries cv2 first, then fallback"""
    if VIDEO_PROCESSING_AVAILABLE:
        success = add_watermark_cv2(input_video_path, output_video_path)
        if success:
            return True
    
    # Fallback - just copy without watermark
    return add_watermark_simple(input_video_path, output_video_path)

def update_prompt_placeholder(mode):
    """Update prompt placeholder based on mode"""
    if mode == "Text to Video":
        return gr.update(
            placeholder="Describe the video you want to create.\nExample: The sun rises slowly between tall buildings. [Ground-level follow shot] Bicycle tires roll over a dew-covered street at dawn.",
            value=""
        )
    else:
        return gr.update(
            placeholder="Describe how the image should move.\nExample: Camera slowly zooms in while clouds move across the sky. The subject's hair gently moves in the wind.",
            value=DEFAULT_IMAGE_PROMPT
        )

def update_image_input(mode):
    """Show/hide image input based on mode"""
    if mode == "Image to Video":
        return gr.update(visible=True)
    else:
        return gr.update(visible=False)

def wait_for_model_with_retry(model_name, max_retries=5, initial_wait=10):
    """Wait for model to be ready with retry logic"""
    for attempt in range(max_retries):
        try:
            # Try to get model info
            model = replicate.models.get(model_name)
            return True
        except Exception as e:
            if attempt < max_retries - 1:
                wait_time = initial_wait * (attempt + 1)
                print(f"Model not ready, waiting {wait_time} seconds... (Attempt {attempt + 1}/{max_retries})")
                time.sleep(wait_time)
            else:
                return False
    return False

def generate_video(mode, prompt, image, aspect_ratio, seed, api_key_input, progress=gr.Progress()):
    """Main video generation function"""
    
    # API token check
    token = api_key_input or api_token
    if not token:
        return None, "❌ API token required. Please set RAPI_TOKEN environment variable or enter your API key."
    
    os.environ["REPLICATE_API_TOKEN"] = token
    
    # Input validation
    if not prompt:
        return None, "❌ Please enter a prompt."
    
    if mode == "Image to Video" and image is None:
        return None, "❌ Please upload an image."
    
    try:
        progress(0, desc="Preparing video generation...")
        
        # Input parameters setup
        input_params = {
            "prompt": prompt,
            "duration": 5,
            "resolution": "480p",
            "aspect_ratio": aspect_ratio,
            "seed": seed
        }
        
        # Image to video mode
        if mode == "Image to Video" and image is not None:
            progress(0.1, desc="Processing image...")
            
            # Convert PIL Image to base64
            if isinstance(image, str):  # File path
                with Image.open(image) as img:
                    buffered = io.BytesIO()
                    img.save(buffered, format="PNG")
                    image_base64 = base64.b64encode(buffered.getvalue()).decode()
            else:  # PIL Image object
                buffered = io.BytesIO()
                image.save(buffered, format="PNG")
                image_base64 = base64.b64encode(buffered.getvalue()).decode()
            
            input_params["image"] = f"data:image/png;base64,{image_base64}"
        
        progress(0.2, desc="Checking model availability...")
        
        # Set up Replicate with the API token
        replicate.api_token = token
        
        # Wait for model to be ready
        model_ready = wait_for_model_with_retry("bytedance/seedance-1-lite")
        if not model_ready:
            return None, "⏳ Model is still booting up. Please try again in a few minutes."
        
        progress(0.3, desc="Calling Replicate API...")
        
        # Run Replicate with retry logic
        max_attempts = 3
        output = None
        
        for attempt in range(max_attempts):
            try:
                # Run Replicate - use the model directly without version specifier
                start_time = time.time()
                
                # Create a generator for progress tracking
                output_generator = replicate.run(
                    "bytedance/seedance-1-lite",
                    input=input_params
                )
                
                # If it's a generator, iterate through it
                if hasattr(output_generator, '__iter__') and not isinstance(output_generator, (str, bytes)):
                    for event in output_generator:
                        elapsed = time.time() - start_time
                        progress_val = min(0.3 + (elapsed / 300) * 0.4, 0.7)
                        progress(progress_val, desc=f"Generating video... ({int(elapsed)}s)")
                        output = event  # Keep the last event as output
                else:
                    # If it's not a generator, use it directly
                    output = output_generator
                
                # If we got output, break the retry loop
                if output:
                    break
                
            except replicate.exceptions.ReplicateError as e:
                error_str = str(e)
                if "cold boot" in error_str.lower() or "starting" in error_str.lower():
                    if attempt < max_attempts - 1:
                        progress(0.3, desc=f"Model is starting up, retrying... (Attempt {attempt + 2}/{max_attempts})")
                        time.sleep(30)
                        continue
                elif "timeout" in error_str.lower() and attempt < max_attempts - 1:
                    progress(0.3, desc=f"Timeout occurred, retrying... (Attempt {attempt + 2}/{max_attempts})")
                    time.sleep(10)
                    continue
                else:
                    return None, f"❌ Replicate API error: {error_str}"
            except Exception as e:
                if attempt < max_attempts - 1:
                    progress(0.3, desc=f"Error occurred, retrying... (Attempt {attempt + 2}/{max_attempts})")
                    time.sleep(5)
                    continue
                else:
                    return None, f"❌ Unexpected error: {str(e)}"
        
        # Check if we got output
        if not output:
            return None, "❌ Failed to generate video after multiple attempts."
        
        progress(0.7, desc="Downloading video...")
        
        # Get video data
        if hasattr(output, 'read'):
            video_data = output.read()
        else:
            # Download from URL with timeout
            response = requests.get(output, timeout=60)
            video_data = response.content
        
        # Save to temporary file
        with tempfile.NamedTemporaryFile(delete=False, suffix='.mp4') as tmp_file:
            tmp_file.write(video_data)
            temp_video_path = tmp_file.name
        
        # Try to add watermark
        watermark_added = False
        final_video_path = temp_video_path
        
        if VIDEO_PROCESSING_AVAILABLE:
            progress(0.8, desc="Adding watermark...")
            final_video_path = tempfile.mktemp(suffix='.mp4')
            watermark_added = add_watermark(temp_video_path, final_video_path)
            
            if not watermark_added or not os.path.exists(final_video_path):
                final_video_path = temp_video_path
        
        # Save final video
        with open(final_video_path, "rb") as f:
            final_video_data = f.read()
        with open("output.mp4", "wb") as file:
            file.write(final_video_data)
        
        # Clean up temp files
        if temp_video_path != final_video_path and os.path.exists(temp_video_path):
            try:
                os.unlink(temp_video_path)
            except:
                pass
        
        progress(1.0, desc="Complete!")
        
        # Generation info
        watermark_status = "Added" if watermark_added else "Not available (cv2 not installed)" if not VIDEO_PROCESSING_AVAILABLE else "Failed"
        info = f"""✅ Video generated successfully!

📊 Generation Info:
- Mode: {mode}
- Aspect Ratio: {aspect_ratio}
- Seed: {seed}
- Duration: 5 seconds
- Resolution: 480p
- Watermark: {watermark_status}
- File: output.mp4"""
        
        return final_video_path, info
        
    except requests.exceptions.Timeout:
        return None, "⏱️ Request timed out. The server might be under heavy load. Please try again in a few minutes."
    except Exception as e:
        error_msg = f"❌ Error occurred: {str(e)}"
        if "timeout" in str(e).lower():
            error_msg += "\n\n💡 Tip: The model might be cold starting. Please wait a minute and try again."
        return None, error_msg

# Gradio interface
with gr.Blocks(title="Bytedance Seedance Video Free", theme=gr.themes.Soft()) as app:
    gr.Markdown("""
    # 🎬 Bytedance Seedance Video' Free
    
    Generate videos from text or images using **Replicate API**.
    
    [![Powered by Ginigen](https://img.shields.io/badge/Powered%20by-Replicate-blue)](https://ginigen.com/)
    """)
    
    with gr.Row():
        with gr.Column(scale=1):
            # API Settings
            with gr.Accordion("⚙️ API Settings", open=not bool(api_token)):
                if api_token:
                    gr.Markdown("✅ API token loaded from environment variable.")
                    api_key_input = gr.Textbox(
                        label="Replicate API Token (Optional)",
                        type="password",
                        placeholder="Enter to override environment variable",
                        value=""
                    )
                else:
                    gr.Markdown("⚠️ RAPI_TOKEN environment variable not set.")
                    api_key_input = gr.Textbox(
                        label="Replicate API Token (Required)",
                        type="password",
                        placeholder="Enter your Replicate API token",
                        value=""
                    )
            
            # Generation mode
            mode = gr.Radio(
                label="🎯 Generation Mode",
                choices=["Text to Video", "Image to Video"],
                value="Text to Video"
            )
            
            # Image upload
            image_input = gr.Image(
                label="📷 Upload Image",
                type="pil",
                visible=False
            )
            
            # Aspect ratio
            aspect_ratio = gr.Dropdown(
                label="📐 Aspect Ratio",
                choices=list(ASPECT_RATIOS.keys()),
                value="16:9",
                info="Choose ratio optimized for social media platforms"
            )
            
            # Ratio description
            ratio_info = gr.Markdown(value=f"Selected ratio: {ASPECT_RATIOS['16:9']}")
            
            # Seed setting
            seed = gr.Number(
                label="🎲 Random Seed",
                value=42,
                precision=0,
                info="Use same seed value to reproduce same results"
            )
            
            # Fixed settings display
            watermark_info = "ginigen.com" if VIDEO_PROCESSING_AVAILABLE else "ginigen.com (requires cv2)"
            gr.Markdown(f"""
            ### 📋 Fixed Settings
            - **Duration**: 5 seconds
            - **Resolution**: 480p
            - **Watermark**: {watermark_info}
            """)
        
        with gr.Column(scale=2):
            # Prompt input
            prompt = gr.Textbox(
                label="✍️ Prompt",
                lines=5,
                placeholder="Describe the video you want to create.\nExample: The sun rises slowly between tall buildings. [Ground-level follow shot] Bicycle tires roll over a dew-covered street at dawn.",
                value=""
            )
            
            # Generate button
            generate_btn = gr.Button("🎬 Generate Video", variant="primary", size="lg")
            
            # Results display
            with gr.Column():
                output_video = gr.Video(
                    label="📹 Generated Video",
                    autoplay=True
                )
                output_info = gr.Textbox(
                    label="Information",
                    lines=8,
                    interactive=False
                )
    
    # Usage instructions
    with gr.Accordion("📖 How to Use", open=False):
        gr.Markdown("""
        ### Installation
        
        1. **Install required packages**:
        ```bash
        pip install gradio replicate pillow requests
        ```
        
        2. **For watermark support (optional)**:
        ```bash
        pip install opencv-python
        ```
        
        3. **Set environment variable** (optional):
        ```bash
        export RAPI_TOKEN="your-replicate-api-token"
        ```
        
        4. **Run**:
        ```bash
        python app.py
        ```
        
        ### Features
        
        - **Text to Video**: Generate video from text description only
        - **Image to Video**: Transform uploaded image into animated video
        - **Aspect Ratios**: Choose ratios optimized for various social media platforms
        - **Seed Value**: Use same seed to reproduce identical results
        - **Watermark**: Automatically adds "ginigen.com" watermark (requires opencv-python)
        
        ### Prompt Writing Tips
        
        - Use specific and detailed descriptions
        - Specify camera movements (e.g., zoom in, pan left, tracking shot)
        - Describe lighting and atmosphere (e.g., golden hour, dramatic lighting)
        - Indicate movement speed (e.g., slowly, rapidly, gently)
        
        ### Troubleshooting
        
        - **Timeout errors**: The model might be cold starting. Wait 1-2 minutes and try again.
        - **Model booting**: First requests after inactivity may take longer as the model boots up.
        - **Extended wait times**: Complex prompts or server load may cause longer generation times.
        - **Watermark not showing**: Install opencv-python for watermark support.
        """)
    
    # Examples
    gr.Examples(
        examples=[
            ["Text to Video", "A serene lake at sunrise with mist rolling over the water. Camera slowly pans across the landscape as birds fly overhead.", None, "16:9", 42],
            ["Text to Video", "Urban street scene at night with neon lights reflecting on wet pavement. People walking with umbrellas, camera tracking forward.", None, "9:16", 123],
            ["Text to Video", "Close-up of a flower blooming in time-lapse, soft natural lighting, shallow depth of field.", None, "1:1", 789],
        ],
        inputs=[mode, prompt, image_input, aspect_ratio, seed],
        label="Example Prompts"
    )
    
    # Event handlers
    mode.change(
        fn=update_prompt_placeholder,
        inputs=[mode],
        outputs=[prompt]
    )
    
    mode.change(
        fn=update_image_input,
        inputs=[mode],
        outputs=[image_input]
    )
    
    aspect_ratio.change(
        fn=lambda x: f"Selected ratio: {ASPECT_RATIOS[x]}",
        inputs=[aspect_ratio],
        outputs=[ratio_info]
    )
    
    generate_btn.click(
        fn=generate_video,
        inputs=[mode, prompt, image_input, aspect_ratio, seed, api_key_input],
        outputs=[output_video, output_info]
    )

# Run app
if __name__ == "__main__":
    app.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=False,
        inbrowser=True
    )