Ship AI image & video generation inside your product

import asyncio, aiohttp

API = "https://api.creativeai.run/v1/images/generations"
KEY = "YOUR_CREATIVEAI_KEY"

async def generate_menu_photos(session, location: dict):
    """Generate a day's food photos for one restaurant location."""
    results = []
    for item in location["menu_items"]:
        resp = await session.post(API, headers={
            "Authorization": f"Bearer {KEY}",
            "Content-Type": "application/json",
        }, json={
            "model": "gpt-image-1",
            "prompt": f"Appetizing {item['name']}: {item['description']}, "
                      f"professional food photography, white plate, "
                      f"restaurant table, warm lighting, top-down angle",
            "size": "1024x1024",
        })
        data = await resp.json()
        results.append({
            "location_id": location["id"],
            "item": item["name"],
            "url": data["data"][0]["url"],
        })
    return results

async def daily_content_run(locations: list):
    """Batch food photos across all locations — ~$0.21/photo."""
    async with aiohttp.ClientSession() as session:
        tasks = [generate_menu_photos(session, loc) for loc in locations]
        return await asyncio.gather(*tasks)

# 30 locations × 5 daily specials = 150 photos → ~$31.50/day
# Each location gets fresh social-ready food content

import asyncio, aiohttp

API_IMG = "https://api.creativeai.run/v1/images/generations"
API_VID = "https://api.creativeai.run/v1/video/generations"
KEY = "YOUR_CREATIVEAI_KEY"
WEBHOOK = "https://your-app.com/api/webhooks/listing-video-ready"

async def listing_photo_to_video(session, listing: dict):
    """Turn listing photos into short property walkthrough clips."""
    jobs = []
    for photo in listing["photos"]:
        resp = await session.post(API_VID, headers={
            "Authorization": f"Bearer {KEY}",
            "Content-Type": "application/json",
        }, json={
            "model": "kling-v3",
            "prompt": f"Smooth cinematic pan across {photo['room_type']}, "
                      f"natural lighting, real estate showcase",
            "image_url": photo["url"],
            "duration": "5s",
            "webhook_url": f"{WEBHOOK}?listing_id={listing['id']}"
                           f"&photo_id={photo['id']}",
        })
        data = await resp.json()
        jobs.append({
            "listing_id": listing["id"],
            "photo_id": photo["id"],
            "job_id": data["id"],
        })
    return jobs

async def batch_listing_videos(listings: list):
    """Submit I2V jobs for all listings — results arrive via webhook."""
    async with aiohttp.ClientSession() as session:
        tasks = [listing_photo_to_video(session, l) for l in listings]
        return await asyncio.gather(*tasks)

# 20 agents × 5 listings × 3 rooms = 300 videos/batch
# ~$0.36/clip → $108/batch, results via signed webhook

import asyncio, aiohttp, json

API = "https://api.creativeai.run/v1/images/generations"
KEY = "YOUR_CREATIVEAI_KEY"

# One product image → multiple ad-creative variations
PRODUCT = "https://cdn.example.com/products/sneaker-hero.png"

VARIATIONS = [
    {"style": "Instagram Story",  "size": "1080x1920", "prompt": "Premium sneaker on gradient background, bold sale text overlay, vertical story format"},
    {"style": "Facebook Feed",    "size": "1200x628",  "prompt": "Lifestyle shot of sneaker on urban street, warm tones, horizontal ad banner"},
    {"style": "Google Display",   "size": "300x250",   "prompt": "Clean product shot of sneaker, white background, compact square layout"},
    {"style": "TikTok Cover",     "size": "1080x1920", "prompt": "Dynamic sneaker with motion blur, neon accents, vertical video thumbnail"},
    {"style": "Email Hero",       "size": "600x400",   "prompt": "Sneaker with seasonal props, soft lighting, email-friendly warm palette"},
]

async def generate_variation(session, variation):
    resp = await session.post(API, headers={
        "Authorization": f"Bearer {KEY}",
        "Content-Type": "application/json",
    }, json={
        "model": "gpt-image-1",
        "prompt": variation["prompt"],
        "size": variation["size"],
    })
    result = await resp.json()
    return {"style": variation["style"], "url": result["data"][0]["url"]}

async def bulk_ad_variations():
    async with aiohttp.ClientSession() as session:
        tasks = [generate_variation(session, v) for v in VARIATIONS]
        results = await asyncio.gather(*tasks)
    # 5 platform-targeted creatives from one product, ~10s total
    for r in results:
        print(f'{r["style"]}: {r["url"]}')
    return results

asyncio.run(bulk_ad_variations())

import asyncio, aiohttp

API_IMG = "https://api.creativeai.run/v1/images/generations"
API_VID = "https://api.creativeai.run/v1/video/generations"
KEY = "YOUR_CREATIVEAI_KEY"
WEBHOOK = "https://your-app.com/api/webhooks/content-ready"

async def generate_scheduled_post(session, post):
    """Generate image + optional video for a scheduled social post."""
    # 1. Generate the post image (~3 credits, ~$0.21)
    img_resp = await session.post(API_IMG, headers={
        "Authorization": f"Bearer {KEY}",
        "Content-Type": "application/json",
    }, json={
        "model": "gpt-image-1",
        "prompt": post["image_prompt"],
        "size": post["size"],  # "1080x1080" for feed, "1080x1920" for stories
    })
    img = await img_resp.json()
    result = {"post_id": post["id"], "image_url": img["data"][0]["url"]}

    # 2. If post needs a video clip, submit async job (~5 credits, ~$0.36)
    if post.get("video_prompt"):
        vid_resp = await session.post(API_VID, headers={
            "Authorization": f"Bearer {KEY}",
            "Content-Type": "application/json",
        }, json={
            "model": "kling-v3",
            "prompt": post["video_prompt"],
            "image_url": img["data"][0]["url"],  # I2V from the still
            "duration": "5s",
            "webhook_url": f"{WEBHOOK}?post_id={post['id']}",
        })
        vid = await vid_resp.json()
        result["video_job_id"] = vid["id"]

    return result

async def batch_content_pipeline(scheduled_posts):
    """Process a day's scheduled posts — images sync, videos async."""
    async with aiohttp.ClientSession() as session:
        tasks = [generate_scheduled_post(session, p) for p in scheduled_posts]
        return await asyncio.gather(*tasks)

# Example: 20 scheduled posts/day → ~$4.76-$6.80 total
# Images return immediately; videos arrive via webhook