Adding API to Your App

from magic_hour import Client
import time

client = Client(token="YOUR_API_KEY")

# 1. Create job
print("Creating face swap job...")
create_res = client.v1.face_swap.create(
    name="My face swap",
    assets={
        "image_file_path": "https://example.com/face.jpg",
        "video_file_path": "https://example.com/video.mp4",
        "video_source": "file"
    },
    start_seconds=0.0,
    end_seconds=10.0,
)

project_id = create_res.id
print(f"Job created: {project_id}")

# 2. Poll for completion
print("Waiting for completion...")
while True:
    status_res = client.v1.video_projects.get(id=project_id)
    print(f"Status: {status_res.status}")

    if status_res.status == "complete":
        print("✅ Render complete!")

        # 3. Download result
        download_url = status_res.downloads[0].url
        import requests
        with open("result.mp4", "wb") as file:
            response = requests.get(download_url)
            file.write(response.content)
        print("📁 File downloaded: result.mp4")
        break

    elif status_res.status == "error":
        print(f"❌ Error: {status_res.error}")
        break

    time.sleep(5)  # Wait 5 seconds before next check

try:
    create_res = client.v1.face_swap.create(...)

    # Poll with error handling
    while True:
        status_res = client.v1.video_projects.get(id=create_res.id)

        if status_res.status == "complete":
            # Success - download result
            break
        elif status_res.status == "error":
            error_code = status_res.error.get("code", "unknown")
            error_msg = status_res.error.get("message", "Unknown error")

            # Handle specific errors
            if error_code == "no_source_face":
                print("❌ No face detected. Please use a different image.")
            else:
                print(f"❌ Error ({error_code}): {error_msg}")
            break

        time.sleep(5)

except Exception as e:
    print(f"❌ Request failed: {e}")

import time

def wait_for_completion(client, project_id, project_type="video"):
    """Smart polling with exponential backoff"""

    get_method = (client.v1.video_projects.get if project_type == "video"
                 else client.v1.image_projects.get)

    while True:
        try:
            res = get_method(id=project_id)
            print(f"Status: {res.status}")

            if res.status == "complete":
                return res  # Success!
            elif res.status == "error":
                raise Exception(f"Job failed: {res.error}")
            time.sleep(3)
        except Exception as e:
            print(f"Error checking status: {e}")

    raise TimeoutError("Job did not complete within timeout")

import requests
import os
from pathlib import Path

def download_result(download_info, output_dir="./downloads"):
    # Create output directory
    Path(output_dir).mkdir(exist_ok=True)

    # Generate filename from URL or use timestamp
    url = download_info["url"]
    filename = f"result_{int(time.time())}.mp4"  # or extract from URL
    filepath = Path(output_dir) / filename

    try:
        print(f"Downloading to {filepath}...")
        response = requests.get(url, stream=True, timeout=60)
        response.raise_for_status()

        # Stream download for large files
        with open(filepath, "wb") as file:
            for chunk in response.iter_content(chunk_size=8192):
                file.write(chunk)

        print(f"✅ Downloaded: {filepath} ({filepath.stat().st_size} bytes)")
        return str(filepath)

    except requests.exceptions.RequestException as e:
        print(f"❌ Download failed: {e}")
        return None

# Handle multiple downloads
for i, download in enumerate(status_res.downloads):
    filename = f"output_{i+1}.{'mp4' if 'video' in download.url else 'jpg'}"
    download_result(download, filename)

client.v1.video_projects.delete(id="cuid")
)

from magic_hour import Client
from magic_hour.environment import Environment

# Use mock server for development
client = Client(
    token="API_TOKEN",
    environment=Environment.MOCK_SERVER
)

# All API calls will return mock data
result = client.v1.face_swap.create(...)  # No credits charged