Generate Thousands of Product Images & Videos at Scale

import requests
import concurrent.futures
import json

API_KEY = "your_api_key"
IMAGE_URL = "https://api.creativeai.run/v1/images/generations"
VIDEO_URL = "https://api.creativeai.run/v1/video/generations"

# Step 1: Raw product photos → AI-generated lifestyle images
# Step 2: Best lifestyle image → short product video
# One API, one set of credentials, one pipeline.

catalog = [
    {"sku": "LAMP-BRASS", "image_url": "https://cdn.example.com/catalog/lamp-brass.jpg"},
    {"sku": "VASE-CERAMIC", "image_url": "https://cdn.example.com/catalog/vase-ceramic.jpg"},
    {"sku": "CHAIR-WALNUT", "image_url": "https://cdn.example.com/catalog/chair-walnut.jpg"},
]

SCENE_PROMPT = "Product in modern living room, natural light, clean composition"
VIDEO_PROMPT = "5-second product hero shot, slow camera orbit, soft shadows"

def run_pipeline(product):
    """Image generation → best pick → video submission for one SKU."""
    sku = product["sku"]

    # --- Stage 1: Generate 4 lifestyle image variants ---
    img_resp = requests.post(
        IMAGE_URL,
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "model": "gpt-image-1",
            "prompt": SCENE_PROMPT,
            "image_url": product["image_url"],  # reference fidelity
            "n": 4,
            "size": "1024x1024",
            "quality": "high",
        },
        timeout=120,
    )
    img_resp.raise_for_status()
    images = img_resp.json().get("data", [])
    best_image_url = images[0]["url"]  # pick first; swap for your own scorer

    # --- Stage 2: Best image → async product video ---
    vid_resp = requests.post(
        VIDEO_URL,
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "model": "kling-v3",
            "prompt": VIDEO_PROMPT,
            "image_url": best_image_url,
            "duration": "5s",
            "aspect_ratio": "16:9",
            "webhook_url": "https://your-app.com/webhooks/creativeai",
        },
        timeout=60,
    )
    vid_resp.raise_for_status()
    job = vid_resp.json()

    return {
        "sku": sku,
        "images_generated": len(images),
        "video_job_id": job["id"],
        "video_status": job["status"],
    }

# Run full pipeline for every SKU (bounded concurrency)
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(run_pipeline, catalog))

# Write manifest for downstream ingestion
with open("pipeline_manifest.jsonl", "w") as f:
    for r in results:
        f.write(json.dumps(r) + "\n")

for r in results:
    print(f"{r['sku']}: {r['images_generated']} images + video job {r['video_job_id']}")
print("Videos will arrive at your webhook as each render completes.")