One API for Shopify Apps & E-commerce Platforms

import requests
import concurrent.futures
import json

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

# Full Shopify pipeline:
# Step 1: Product photos → AI lifestyle images (Seedream V1.5)
# Step 2: Best image → product video (Kling V3)
# One API, one set of credentials, one billing account.

catalog = [
    {"sku": "LAMP-BRASS", "image_url": "https://cdn.shopify.com/s/.../lamp.jpg"},
    {"sku": "VASE-CERAMIC", "image_url": "https://cdn.shopify.com/s/.../vase.jpg"},
    {"sku": "CHAIR-WALNUT", "image_url": "https://cdn.shopify.com/s/.../chair.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": "seedream-3",
            "prompt": SCENE_PROMPT,
            "image_url": product["image_url"],
            "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"]

    # --- 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-shopify-app.com/webhooks/video-ready",
        },
        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
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(run_pipeline, catalog))

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