Adding API to Your App

# services/magichour.py
from magic_hour import Client
import os

# Initialize client with environment variable
client = Client(token=os.getenv("MAGIC_HOUR_API_KEY"))

def get_client():
    """Get configured Magic Hour client"""
    return client

# services/video_processor.py
from .magichour import get_client

def create_face_swap(face_image_url, video_url, start_time=0.0, end_time=10.0):
    """Create a face swap job and return project ID"""
    client = get_client()

    # Submit job to Magic Hour API
    create_res = client.v1.face_swap.create(
        name="Face swap job",
        assets={
            "image_file_path": face_image_url,
            "video_file_path": video_url,
            "video_source": "file"
        },
        start_seconds=start_time,
        end_seconds=end_time,
    )

    return {
        "project_id": create_res.id,
        "status": create_res.status,
        "credits_charged": create_res.credits_charged
    }

def check_job_status(project_id):
    """Check the current status of a video job"""
    client = get_client()

    try:
        # Get current status from Magic Hour
        status_res = client.v1.video_projects.get(id=project_id)

        return {
            "project_id": project_id,
            "status": status_res.status,
            "downloads": status_res.downloads if status_res.status == "complete" else None,
            "error": status_res.error if status_res.status == "error" else None
        }
    except Exception as e:
        return {
            "project_id": project_id,
            "status": "error",
            "error": {"message": str(e)}
        }

# services/file_manager.py
import requests
import os
from pathlib import Path

def download_video(download_info, output_dir="./downloads"):
    """Download completed video to local storage"""
    # Create output directory if it doesn't exist
    Path(output_dir).mkdir(exist_ok=True)

    # Generate unique filename
    filename = f"face_swap_{int(time.time())}.mp4"
    filepath = Path(output_dir) / filename

    try:
        # Download the video file
        response = requests.get(download_info["url"], stream=True, timeout=60)
        response.raise_for_status()

        # Save to disk
        with open(filepath, "wb") as file:
            for chunk in response.iter_content(chunk_size=8192):
                file.write(chunk)

        return {
            "success": True,
            "filepath": str(filepath),
            "size_bytes": filepath.stat().st_size
        }
    except Exception as e:
        return {
            "success": False,
            "error": str(e)
        }

# main.py - Complete workflow
from services.video_processor import create_face_swap, check_job_status
from services.file_manager import download_video
import time

def process_face_swap(face_image_url, video_url):
    """Complete face swap workflow"""
    print("🚀 Starting face swap...")

    # Step 1: Create the job
    job = create_face_swap(face_image_url, video_url)
    print(f"✅ Job created: {job['project_id']} (Cost: {job['credits_charged']} credits)")

    # Step 2: Wait for completion
    print("⏳ Waiting for completion...")
    while True:
        status = check_job_status(job['project_id'])
        print(f"Status: {status['status']}")

        if status['status'] == 'complete':
            print("🎉 Job completed successfully!")

            # Step 3: Download result
            result = download_video(status['downloads'][0])
            if result['success']:
                print(f"📁 Video downloaded: {result['filepath']}")
                return result['filepath']
            else:
                print(f"❌ Download failed: {result['error']}")
                return None

        elif status['status'] == 'error':
            print(f"❌ Job failed: {status['error']}")
            return None

        time.sleep(5)  # Wait 5 seconds before checking again

# Usage
if __name__ == "__main__":
    result = process_face_swap(
        "https://example.com/face.jpg",
        "https://example.com/video.mp4"
    )

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