runway_video_examples.py

#!/usr/bin/env python3
"""
Runway Video Generation Examples

This script demonstrates how to use the Runway video generation functionality
both directly with the processor and via the API endpoints.
"""

import json
import requests
import time
from pathlib import Path

# Configuration
API_BASE_URL = "http://localhost:8000"
EXAMPLE_IMAGE_URL = "https://images.unsplash.com/photo-1506905925346-21bda4d32df4?w=800"


def test_api_video_generation():
    """Test video generation via the API endpoint."""
    print("🎬 Testing video generation via API...")

    # Prepare request
    payload = {
        "image_url": EXAMPLE_IMAGE_URL,
        "prompt_text": "A serene mountain landscape with clouds slowly drifting across the sky",
        "ratio": "1280:720",
        "duration": 5,
        "model": "gen4_turbo"
    }

    try:
        print(f"📤 Sending request to {API_BASE_URL}/generate-video")
        print(f"   Image: {payload['image_url']}")
        print(f"   Prompt: {payload['prompt_text']}")
        print(f"   Duration: {payload['duration']}s")

        response = requests.post(
            f"{API_BASE_URL}/generate-video",
            json=payload,
            headers={"Content-Type": "application/json"},
            timeout=360  # 6 minutes timeout for video generation
        )

        if response.status_code == 200:
            result = response.json()
            print("✅ Video generation successful!")
            print(f"   Status: {result['status']}")
            print(f"   Task ID: {result.get('task_id', 'N/A')}")
            print(f"   Video URL: {result.get('video_url', 'N/A')}")
            print(f"   Processing time: {result['processing_time_seconds']:.1f}s")

            # Save result to file
            timestamp = time.strftime("%Y%m%d_%H%M%S")
            output_file = f"runway_video_result_{timestamp}.json"
            with open(output_file, 'w') as f:
                json.dump(result, f, indent=2)
            print(f"📁 Result saved to: {output_file}")

        else:
            print(f"❌ Request failed with status {response.status_code}")
            print(f"   Error: {response.text}")

    except requests.exceptions.Timeout:
        print("⏰ Request timed out - video generation may still be processing")
    except requests.exceptions.RequestException as e:
        print(f"❌ Request failed: {e}")


def test_direct_processor():
    """Test video generation using the processor directly."""
    print("🎬 Testing video generation via direct processor...")

    try:
        # Import the necessary components
        from src.text_extractor import TextExtractor
        from src.config import Config

        # Check if Runway API key is configured
        if not Config.RUNWAY_API_KEY:
            print("❌ RUNWAY_API_KEY not configured. Please set it in your .env file.")
            return

        # Initialize extractor
        extractor = TextExtractor()

        if not hasattr(extractor, 'runway_processor') or not extractor.runway_processor:
            print("❌ Runway processor not available. Please check your configuration.")
            return

        print("✅ Runway processor initialized")

        # Generate video
        result_json = extractor.generate_video_from_image(
            image_path=EXAMPLE_IMAGE_URL,
            prompt_text="A peaceful ocean scene with gentle waves and seabirds",
            ratio="1280:720",
            duration=5
        )

        result = json.loads(result_json)

        print("✅ Video generation completed!")
        print(f"   Status: {result['status']}")
        print(f"   Video URL: {result.get('video_url', 'N/A')}")
        print(f"   Processing time: {result.get('processing_time_seconds', 0):.1f}s")

        # Save result
        timestamp = time.strftime("%Y%m%d_%H%M%S")
        output_file = f"runway_direct_result_{timestamp}.json"
        with open(output_file, 'w') as f:
            json.dump(result, f, indent=2)
        print(f"📁 Result saved to: {output_file}")

    except ImportError as e:
        print(f"❌ Import error: {e}")
        print("   Make sure you're running from the project root directory")
    except Exception as e:
        print(f"❌ Error: {e}")


def test_health_check():
    """Test if the API is running and healthy."""
    print("🔍 Checking API health...")

    try:
        response = requests.get(f"{API_BASE_URL}/health", timeout=10)

        if response.status_code == 200:
            health = response.json()
            print("✅ API is healthy")
            print(f"   Status: {health['status']}")
            print(f"   Version: {health['version']}")

            # Check API keys
            api_keys = health.get('api_keys_configured', {})
            print(f"   OpenAI configured: {api_keys.get('openai', False)}")
            print(f"   Gemini configured: {api_keys.get('gemini', False)}")

            return True
        else:
            print(f"❌ API health check failed: {response.status_code}")
            return False

    except requests.exceptions.RequestException as e:
        print(f"❌ Cannot connect to API: {e}")
        print(f"   Make sure the API server is running at {API_BASE_URL}")
        return False


def generate_curl_examples():
    """Generate curl command examples for video generation."""
    print("📋 CURL Command Examples for Video Generation:")
    print()

    # Basic example
    print("# Basic video generation:")
    curl_cmd = f'''curl -X POST "{API_BASE_URL}/generate-video" \\
  -H "Content-Type: application/json" \\
  -d '{{
    "image_url": "{EXAMPLE_IMAGE_URL}",
    "prompt_text": "A serene mountain landscape with clouds slowly moving",
    "ratio": "1280:720",
    "duration": 5
  }}\''''
    print(curl_cmd)
    print()

    # Advanced example
    print("# Advanced video generation with custom settings:")
    curl_cmd_advanced = f'''curl -X POST "{API_BASE_URL}/generate-video" \\
  -H "Content-Type: application/json" \\
  -d '{{
    "image_url": "https://your-image-url.com/image.jpg",
    "prompt_text": "A dynamic urban cityscape with traffic and people moving",
    "ratio": "1024:1024",
    "duration": 8,
    "model": "gen4_turbo"
  }}\''''
    print(curl_cmd_advanced)
    print()

    # Save to file
    with open("runway_curl_examples.sh", 'w') as f:
        f.write("#!/bin/bash\n")
        f.write("# Runway Video Generation CURL Examples\n\n")
        f.write("# Basic video generation\n")
        f.write(curl_cmd + "\n\n")
        f.write("# Advanced video generation\n")
        f.write(curl_cmd_advanced + "\n")

    print("📁 CURL examples saved to: runway_curl_examples.sh")


def main():
    """Main function to run all tests."""
    print("🚀 Runway Video Generation Test Suite")
    print("=" * 50)

    # Test API health first
    if test_health_check():
        print()

        # Test API endpoint
        test_api_video_generation()
        print()

        # Generate curl examples
        generate_curl_examples()
        print()

    # Test direct processor (works even if API is down)
    test_direct_processor()

    print()
    print("🎉 Test suite completed!")
    print()
    print("📝 Notes:")
    print("   - Video generation can take 1-5 minutes depending on complexity")
    print("   - Make sure your RUNWAY_API_KEY is set in your .env file")
    print("   - The API server needs to be running for endpoint tests")
    print("   - Generated videos will be available at the returned URL")


if __name__ == "__main__":
    main()