import requests

PIKA_API_KEY = "your-pika-api-key"
BASE_URL = "https://api.pika.art/v1"

headers = {
 "Authorization": f"Bearer {PIKA_API_KEY}",
 "Content-Type": "application/json"
}

def generate_video(prompt: str, duration: int = 4, resolution: str = "720p"):
 """Generate a video from a text prompt"""
 response = requests.post(
 f"{BASE_URL}/generate",
 headers=headers,
 json={
 "prompt": prompt,
 "duration": duration,
 "resolution": resolution,
 "aspect_ratio": "16:9",
 "motion_strength": 3, # 1-5, higher = more motion
 "guidance_scale": 12, # How closely to follow prompt
 "negative_prompt": "blurry, low quality, distorted faces"
 }
 )
 return response.json()

# Example
result = generate_video(
 prompt="A futuristic city at night with flying cars and neon signs, "
 "rain falling, cyberpunk atmosphere, cinematic",
 duration=8,
 resolution="1080p"
)
task_id = result["data"]["task_id"]

import base64

def image_to_video(image_path: str, prompt: str, duration: int = 4):
 """Animate a static image"""
 with open(image_path, "rb") as f:
 image_b64 = base64.b64encode(f.read()).decode()
 
 response = requests.post(
 f"{BASE_URL}/generate/image-to-video",
 headers=headers,
 json={
 "image": f"data:image/jpeg;base64,{image_b64}",
 "prompt": prompt,
 "duration": duration,
 "motion_strength": 2,
 "camera_motion": "slow_zoom_in"
 }
 )
 return response.json()

# Animate a product photo
result = image_to_video(
 "product_hero.jpg",
 "Subtle rotation with floating particles, premium feel",
 duration=6
)

def apply_scene_effect(image_path: str, effect: str, target: str = None):
 """Apply Pika 2.2 scene effects
 
 Effects: explode, melt, inflate, crush, dissolve, 
 freeze, burn, grow, shrink, levitate
 """
 with open(image_path, "rb") as f:
 image_b64 = base64.b64encode(f.read()).decode()
 
 response = requests.post(
 f"{BASE_URL}/generate/effects",
 headers=headers,
 json={
 "image": f"data:image/jpeg;base64,{image_b64}",
 "effect": effect,
 "target_description": target, # e.g., "the red car"
 "intensity": 0.7, # 0.0-1.0
 "duration": 4
 }
 )
 return response.json()

# Make a building explode
result = apply_scene_effect(
 "city_scene.jpg",
 effect="explode",
 target="the glass skyscraper in the center"
)

# Melt a chocolate bar
result = apply_scene_effect(
 "chocolate.jpg",
 effect="melt",
 target="the chocolate bar"
)

import time

def wait_for_video(task_id: str, timeout: int = 300):
 """Poll until video is ready"""
 start = time.time()
 
 while time.time() - start < timeout:
 response = requests.get(
 f"{BASE_URL}/tasks/{task_id}",
 headers=headers
 )
 data = response.json()["data"]
 
 if data["status"] == "completed":
 return {
 "video_url": data["video_url"],
 "thumbnail": data["thumbnail_url"],
 "duration": data["duration"],
 "cost_credits": data["credits_used"]
 }
 elif data["status"] == "failed":
 raise Exception(f"Generation failed: {data.get('error', 'Unknown')}")
 
 # Adaptive polling: start fast, slow down
 elapsed = time.time() - start
 sleep_time = 2 if elapsed < 30 else 5 if elapsed < 120 else 10
 time.sleep(sleep_time)
 
 raise TimeoutError(f"Video generation timed out after {timeout}s")

# Usage
video = wait_for_video(task_id)
print(f"Video ready: {video['video_url']}")

import asyncio
from dataclasses import dataclass
from typing import Optional

@dataclass
class VideoResult:
 url: str
 model: str
 cost: float
 duration: float

class VideoGenerationPipeline:
 """Multi-model video generation with automatic fallback"""
 
 def __init__(self, crazyrouter_key: str):
 self.api_key = crazyrouter_key
 self.base_url = "https://crazyrouter.com/v1"
 
 # Model priority: cheapest first, fallback to premium
 self.models = [
 {"name": "pika-2.2", "max_retries": 2},
 {"name": "seedance-2.0", "max_retries": 2},
 {"name": "kling-2.1", "max_retries": 1},
 {"name": "veo-3", "max_retries": 1}, # Premium fallback
 ]
 
 async def generate(self, prompt: str, duration: int = 8,
 resolution: str = "720p") -> VideoResult:
 """Try each model in order until one succeeds"""
 
 last_error = None
 for model_config in self.models:
 model = model_config["name"]
 retries = model_config["max_retries"]
 
 for attempt in range(retries):
 try:
 result = await self._call_model(
 model, prompt, duration, resolution
 )
 return result
 except RateLimitError:
 # Move to next model immediately
 break
 except Exception as e:
 last_error = e
 if attempt < retries - 1:
 await asyncio.sleep(2 ** attempt)
 
 raise Exception(f"All models failed. Last error: {last_error}")
 
 async def _call_model(self, model: str, prompt: str,
 duration: int, resolution: str) -> VideoResult:
 """Call a specific model via Crazyrouter"""
 import openai
 
 client = openai.OpenAI(
 api_key=self.api_key,
 base_url=self.base_url
 )
 
 response = client.chat.completions.create(
 model=model,
 messages=[{
 "role": "user",
 "content": f"Generate video: {prompt}"
 }]
 )
 
 return VideoResult(
 url=response.choices[0].message.content,
 model=model,
 cost=0.0, # Tracked by Crazyrouter dashboard
 duration=duration
 )

# Usage
pipeline = VideoGenerationPipeline("sk-cr-your-key")

# Generates with Pika 2.2 first, falls back automatically
video = asyncio.run(pipeline.generate(
 "Product unboxing animation with confetti and sparkles",
 duration=8,
 resolution="720p"
))
print(f"Generated with {video.model}: {video.url}")