"Sora 2 API Tutorial: How to Generate Videos with OpenAI's Latest Model"

# Set your API key
export CRAZYROUTER_API_KEY="your-api-key"

import requests
import time
import os

API_KEY = os.environ["CRAZYROUTER_API_KEY"]
BASE_URL = "https://api.crazyrouter.com/v1"

def generate_video(prompt: str, duration: int = 5, resolution: str = "720p",
                   aspect_ratio: str = "16:9") -> str:
    """Generate a video from a text prompt using Sora 2."""
    
    # Step 1: Submit generation request
    response = requests.post(
        f"{BASE_URL}/videos/generations",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json"
        },
        json={
            "model": "sora-2",
            "prompt": prompt,
            "duration": duration,
            "resolution": resolution,
            "aspect_ratio": aspect_ratio,
            "n": 1
        }
    )
    response.raise_for_status()
    task = response.json()
    task_id = task["id"]
    print(f"Task submitted: {task_id}")
    
    # Step 2: Poll for completion
    while True:
        status_response = requests.get(
            f"{BASE_URL}/videos/generations/{task_id}",
            headers={"Authorization": f"Bearer {API_KEY}"}
        )
        status = status_response.json()
        
        if status["status"] == "completed":
            video_url = status["data"][0]["url"]
            print(f"Video ready: {video_url}")
            return video_url
        elif status["status"] == "failed":
            raise Exception(f"Generation failed: {status.get('error', 'Unknown error')}")
        
        print(f"Status: {status['status']}... waiting")
        time.sleep(10)

# Generate a video
video_url = generate_video(
    prompt="A golden retriever running through a field of sunflowers at sunset, "
           "cinematic lighting, slow motion, 4K quality",
    duration=10,
    resolution="1080p"
)

const API_KEY = process.env.CRAZYROUTER_API_KEY;
const BASE_URL = 'https://api.crazyrouter.com/v1';

async function generateVideo(prompt, options = {}) {
  const { duration = 5, resolution = '720p', aspectRatio = '16:9' } = options;

  // Submit generation request
  const submitResponse = await fetch(`${BASE_URL}/videos/generations`, {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${API_KEY}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({
      model: 'sora-2',
      prompt,
      duration,
      resolution,
      aspect_ratio: aspectRatio,
      n: 1,
    }),
  });

  const task = await submitResponse.json();
  console.log(`Task submitted: ${task.id}`);

  // Poll for completion
  while (true) {
    const statusResponse = await fetch(
      `${BASE_URL}/videos/generations/${task.id}`,
      { headers: { 'Authorization': `Bearer ${API_KEY}` } }
    );
    const status = await statusResponse.json();

    if (status.status === 'completed') {
      console.log(`Video ready: ${status.data[0].url}`);
      return status.data[0].url;
    }
    if (status.status === 'failed') {
      throw new Error(`Generation failed: ${status.error || 'Unknown'}`);
    }

    console.log(`Status: ${status.status}... waiting`);
    await new Promise((r) => setTimeout(r, 10000));
  }
}

// Generate a video
const url = await generateVideo(
  'A futuristic city at night with flying cars and neon lights, cyberpunk style, rain',
  { duration: 10, resolution: '1080p' }
);

# Submit video generation
curl -X POST https://api.crazyrouter.com/v1/videos/generations \
  -H "Authorization: Bearer $CRAZYROUTER_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "sora-2",
    "prompt": "Ocean waves crashing on a rocky coastline at sunrise, aerial drone shot",
    "duration": 5,
    "resolution": "720p",
    "aspect_ratio": "16:9"
  }'

# Check status (replace TASK_ID)
curl https://api.crazyrouter.com/v1/videos/generations/TASK_ID \
  -H "Authorization: Bearer $CRAZYROUTER_API_KEY"

import base64

def image_to_video(image_path: str, prompt: str, duration: int = 5) -> str:
    """Animate a static image into a video."""
    
    with open(image_path, "rb") as f:
        image_b64 = base64.b64encode(f.read()).decode()
    
    response = requests.post(
        f"{BASE_URL}/videos/generations",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json"
        },
        json={
            "model": "sora-2",
            "prompt": prompt,
            "image": f"data:image/png;base64,{image_b64}",
            "duration": duration,
            "resolution": "720p"
        }
    )
    response.raise_for_status()
    task = response.json()
    
    # Poll for completion (same as above)
    return poll_for_completion(task["id"])

# Animate a product photo
video_url = image_to_video(
    "product_photo.png",
    "Camera slowly orbits around the product, studio lighting, white background",
    duration=5
)

[Subject] + [Action] + [Setting] + [Style] + [Camera] + [Lighting]

prompts = {
    "cinematic": (
        "A lone astronaut walking across the surface of Mars, "
        "red dust swirling around their boots, Earth visible in the sky, "
        "cinematic wide shot, golden hour lighting"
    ),
    "product": (
        "A sleek smartphone rotating 360 degrees on a reflective surface, "
        "studio lighting, dark background, product showcase style"
    ),
    "nature": (
        "Time-lapse of a flower blooming in a garden, "
        "macro close-up, soft natural lighting, shallow depth of field"
    ),
    "social_media": (
        "A coffee cup being poured with latte art forming, "
        "top-down view, warm cafe lighting, Instagram aesthetic, 9:16 vertical"
    ),
}

import time
from typing import Optional

def generate_video_robust(prompt: str, max_retries: int = 3, 
                          **kwargs) -> Optional[str]:
    """Generate video with retry logic and error handling."""
    
    for attempt in range(max_retries):
        try:
            return generate_video(prompt, **kwargs)
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429:
                wait_time = 2 ** attempt * 30  # Exponential backoff
                print(f"Rate limited. Waiting {wait_time}s...")
                time.sleep(wait_time)
            elif e.response.status_code >= 500:
                print(f"Server error. Retrying in 10s...")
                time.sleep(10)
            else:
                raise
        except Exception as e:
            print(f"Attempt {attempt + 1} failed: {e}")
            if attempt < max_retries - 1:
                time.sleep(10)
    
    return None

# If supported, use webhooks instead of polling
response = requests.post(
    f"{BASE_URL}/videos/generations",
    headers={
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    },
    json={
        "model": "sora-2",
        "prompt": "A cat playing piano, funny, viral video style",
        "duration": 5,
        "webhook_url": "https://your-app.com/api/video-callback"
    }
)