papp.py

import gradio as gr
import os
import time
from byteplussdkarkruntime import Ark
import tempfile
import requests
from PIL import Image
from io import BytesIO

# Initialize the client (will be set when API key is provided)
client = None

def initialize_client(api_key):
    """Initialize the Ark client with the provided API key"""
    global client
    try:
        client = Ark(
            base_url="https://ark.ap-southeast.bytepluses.com/api/v3",
            api_key=api_key,
        )
        return "✅ Client initialized successfully!"
    except Exception as e:
        return f"❌ Error initializing client: {str(e)}"

def generate_video(api_key, prompt_text, image_url, model_id="seedance-1-5-pro-251215"):
    """Generate video using the Ark API"""
    global client
    
    # Initialize client if not already done or if API key changed
    if client is None:
        init_result = initialize_client(api_key)
        if "❌" in init_result:
            return init_result, None
    else:
        # Update client with new API key (optional - you might want to recreate)
        try:
            client = Ark(base_url="https://ark.ap-southeast.bytepluses.com/api/v3", api_key=api_key)
        except:
            pass
    
    try:
        print("🚀 Creating video generation request...")
        
        # Prepare content
        content = [
            {
                "type": "text",
                "text": prompt_text if prompt_text else "At breakneck speed, drones thread through intricate obstacles --duration 5 --camerafixed false"
            },
            {
                "type": "image_url",
                "image_url": {
                    "url": image_url if image_url else "https://ark-doc.tos-ap-southeast-1.bytepluses.com/seepro_i2v%20.png"
                }
            }
        ]
        
        # Create task
        create_result = client.content_generation.tasks.create(
            model=model_id,
            content=content
        )
        
        task_id = create_result.id
        print(f"📋 Task ID: {task_id}")
        
        # Polling for result
        yield f"⏳ Task created with ID: {task_id}. Waiting for completion...", None
        
        max_attempts = 60  # Maximum polling attempts (60 seconds)
        attempts = 0
        
        while attempts < max_attempts:
            get_result = client.content_generation.tasks.get(task_id=task_id)
            status = get_result.status
            
            if status == "succeeded":
                print("✅ Task succeeded!")
                # Extract video URL from result
                # Note: Adjust this based on actual response structure
                video_url = None
                if hasattr(get_result, 'output') and get_result.output:
                    if isinstance(get_result.output, list) and len(get_result.output) > 0:
                        video_url = get_result.output[0].get('video_url')
                
                if video_url:
                    yield f"✅ Video generated successfully!", video_url
                else:
                    yield f"✅ Task completed, but no video URL found in response. Full response: {get_result}", None
                break
                
            elif status == "failed":
                error_msg = get_result.error if hasattr(get_result, 'error') else "Unknown error"
                yield f"❌ Task failed: {error_msg}", None
                break
            else:
                progress_msg = f"⏳ Current status: {status}, waiting... (attempt {attempts + 1}/{max_attempts})"
                print(progress_msg)
                yield progress_msg, None
                time.sleep(1)
                attempts += 1
        
        if attempts >= max_attempts:
            yield "⏰ Timeout: Task took too long to complete. Please try again.", None
            
    except Exception as e:
        yield f"❌ Error: {str(e)}", None

def upload_image(image_file):
    """Handle image upload and return URL"""
    if image_file is None:
        return "https://ark-doc.tos-ap-southeast-1.bytepluses.com/seepro_i2v%20.png"
    
    # For Hugging Face Spaces, we can use a temporary file approach
    # In production, you might want to upload to a hosting service
    return image_file.name

# Create Gradio interface
with gr.Blocks(title="BytePlus Video Generation", theme=gr.themes.Soft()) as demo:
    gr.Markdown("""
    # 🎥 BytePlus Video Generation
    Generate stunning videos from images and text prompts using the BytePlus API.
    """)
    
    with gr.Row():
        with gr.Column(scale=1):
            # API Key input
            api_key = gr.Textbox(
                label="🔑 API Key",
                placeholder="Enter your BytePlus API key here",
                type="password",
                value="key"  # This will be "key" as per your instruction
            )
            
            # Model selection
            model_id = gr.Textbox(
                label="🤖 Model ID",
                value="seedance-1-5-pro-251215",
                info="Default model for video generation"
            )
            
            # Image upload
            image_input = gr.Image(
                label="🖼️ Upload Starting Image (Optional)",
                type="filepath"
            )
            
            # Manual image URL input
            image_url_input = gr.Textbox(
                label="🔗 Or Enter Image URL",
                placeholder="https://example.com/image.jpg",
                value="https://ark-doc.tos-ap-southeast-1.bytepluses.com/seepro_i2v%20.png"
            )
            
            # Text prompt
            prompt_input = gr.Textbox(
                label="📝 Text Prompt",
                placeholder="Describe your video...",
                value="At breakneck speed, drones thread through intricate obstacles or stunning natural wonders, delivering an immersive, heart-pounding flying experience.  --duration 5 --camerafixed false",
                lines=4
            )
            
            # Generate button
            generate_btn = gr.Button("🚀 Generate Video", variant="primary")
        
        with gr.Column(scale=1):
            # Status output
            status_output = gr.Textbox(
                label="📊 Status",
                placeholder="Generation status will appear here...",
                lines=3
            )
            
            # Video output
            video_output = gr.Video(
                label="🎬 Generated Video",
                interactive=False
            )
            
            # Direct video URL
            video_url_output = gr.Textbox(
                label="🔗 Video URL",
                placeholder="Video URL will appear here...",
                lines=2,
                interactive=False
            )
    
    # Example section
    gr.Markdown("""
    ## 📋 Examples
    Try these example prompts:
    """)
    
    with gr.Row():
        example1 = gr.Button("🌄 Nature drone shot")
        example2 = gr.Button("🏙️ City flythrough")
        example3 = gr.Button("🌊 Ocean waves")
    
    def set_example1():
        return "Aerial drone shot flying over majestic mountains at sunrise, cinematic lighting, smooth motion --duration 5 --camerafixed false"
    
    def set_example2():
        return "Fast-paced drone racing through futuristic city streets with neon lights, dynamic angles --duration 5 --camerafixed false"
    
    def set_example3():
        return "Drone following surfers riding massive waves, slow motion, dramatic ocean views --duration 5 --camerafixed false"
    
    example1.click(fn=set_example1, outputs=prompt_input)
    example2.click(fn=set_example2, outputs=prompt_input)
    example3.click(fn=set_example3, outputs=prompt_input)
    
    # Handle generation
    generate_btn.click(
        fn=generate_video,
        inputs=[api_key, prompt_input, image_url_input, model_id],
        outputs=[status_output, video_output]
    ).then(
        fn=lambda url: url if url else "No URL available",
        inputs=[video_output],
        outputs=[video_url_output]
    )
    
    # When image is uploaded, update the URL input
    image_input.change(
        fn=upload_image,
        inputs=image_input,
        outputs=image_url_input
    )

if __name__ == "__main__":
    demo.launch()