Tutorial Batch Image Generation via REST API AiStudio untuk Katalog Ribuan Produk

[Katalog CSV/ERP] 
       │
       ▼
[Manual Web UI Prompting] ──(Bottleneck: 1 gambar / 2 menit + Human Error)──► [Katalog Terlambat]
       │
   (Solusi API)
       ▼
[AiStudio REST API Gateway] ──(Concurrency: 50 req/sec via Banana Pro)─────► [Ribuan Aset Siap S3/CDN]

+-----------------------------------------------------------------------+
|                         KATALOG PRODUK(CSV/DB)                       |
|   - SKU: "SKU-BEAUTY-091"                                             |
|   - Image URL: "https://storage.brand.id/raw/serum_transparan.png"    |
|   - Theme: "Warm minimalist bathroom, morning sunlight, terrazzo"     |
+-----------------------------------┬-----------------------------------+
                                    │
                                    ▼
+-----------------------------------------------------------------------+
|                    WORKER ENGINE(Python AsyncIO)                     |
|   - Worker Pool(Semaphore limit = 10)                                |
|   - Dynamic Prompt Interpolation Engine                               |
|   - Exponential Backoff & Retry Mechanism                             |
+-----------------------------------┬-----------------------------------+
                                    │  REST POST /v1/images/generations
                                    ▼  (Model: banana-pro)
+-----------------------------------------------------------------------+
|                       AISTUDIO.ID API GATEWAY                         |
|   - Authentication & Rate Limiting                                    |
|   - Routing ke Banana Pro Engine                                      |
+-----------------------------------┬-----------------------------------+
                                    │
                                    ▼
+-----------------------------------------------------------------------+
|                            OUTPUT PIPELINE                            |
|   - Structured JSON Response Validation                               |
|   - Asset Downloader & S3/R2 Pipeline Auto-Sync                       |
|   - Output Catalog Manifest Generation(SKU-to-Render.json)           |
+-----------------------------------------------------------------------+

pip install aiohttp asyncio tenacity pydantic

import asyncio
import csv
import json
import logging
import os
from pathlib import Path
from typing import Dict, List, Optional
import aiohttp
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

# --- KONFIGURASI LOGGING ---
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler()]
)
logger = logging.getLogger("AiStudioBatch")

# --- KONFIGURASI API & ENVIRONMENT ---
AISTUDIO_API_KEY = os.getenv("AISTUDIO_API_KEY", "YOUR_API_KEY_HERE")
AISTUDIO_BASE_URL = "https://api.aistudio.id/v1/images/generations"
MODEL_NAME = "banana-pro"
MAX_CONCURRENT_REQUESTS = 10  # Sesuaikan dengan tier rate limit Anda
OUTPUT_DIR = Path("./generated_catalog")
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
CHECKPOINT_FILE = OUTPUT_DIR / "checkpoint.json"

# --- TEMPLATE PROMPT KHUSUS PRODUK ---
PROMPT_TEMPLATE = (
    "Commercial product photography of {product_name}. "
    "Placed naturally on a {surface_material}, background is a {scene_context}. "
    "Natural ambient lighting, soft directional window light, realistic contact shadows, "
    "clean modern aesthetic, hyper-detailed texture, 8k resolution, editorial composition, "
    "no text, no watermark, no extra unwanted objects."
)

def load_checkpoint() -> set:
    """Memuat daftar SKU yang sudah sukses diproses agar proses idempotent."""
    if CHECKPOINT_FILE.exists():
        try:
            with open(CHECKPOINT_FILE, "r") as f:
                return set(json.load(f))
        except Exception as e:
            logger.warning(f"Gagal membaca checkpoint: {e}. Mengulang dari awal.")
    return set()

def save_checkpoint(completed_skus: set):
    """Menyimpan SKU yang telah selesai ke checkpoint file."""
    with open(CHECKPOINT_FILE, "w") as f:
        json.dump(list(completed_skus), f, indent=2)

@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError)),
    reraise=True
)
async def generate_product_image(
    session: aiohttp.ClientSession,
    semaphore: asyncio.Semaphore,
    row: Dict[str, str]
) -> Optional[Dict[str, str]]:
    """Mengirim payload ke endpoint Banana Pro AiStudio dengan kontrol konkurensi."""
    sku = row["sku"]
    product_name = row["product_name"]
    surface = row.get("surface", "minimalist matte stone pedestal")
    scene = row.get("scene", "high-end boutique showroom, warm blurred ambient")
    
    prompt = PROMPT_TEMPLATE.format(
        product_name=product_name,
        surface_material=surface,
        scene_context=scene
    )

    payload = {
        "model": MODEL_NAME,
        "prompt": prompt,
        "n": 1,
        "size": "1024x1024",
        "response_format": "url",
        "parameters": {
            "negative_prompt": "blurry, low quality, distorted geometry, fake plastic look, bad shadows, low resolution",
            "guidance_scale": 7.5,
            "steps": 30
        }
    }

    headers = {
        "Authorization": f"Bearer {AISTUDIO_API_KEY}",
        "Content-Type": "application/json"
    }

    async with semaphore:
        logger.info(f"Memproses SKU: {sku} - {product_name}")
        async with session.post(AISTUDIO_BASE_URL, json=payload, headers=headers, timeout=60) as response:
            if response.status == 429:
                logger.warning(f"Rate limited pada SKU: {sku}. Tenacity akan melakukan retry.")
                raise aiohttp.ClientError("Rate limited(HTTP 429)")

            if response.status != 200:
                error_body = await response.text()
                logger.error(f"Gagal memproses SKU {sku} (HTTP {response.status}): {error_body}")
                return None

            data = await response.json()
            image_url = data["data"][0]["url"]
            
            # Unduh dan simpan gambar secara asinkron
            return await save_image(session, sku, image_url)

async def save_image(session: aiohttp.ClientSession, sku: str, image_url: str) -> Optional[Dict[str, str]]:
    """Mengunduh gambar hasil generasi dan menyimpannya ke disk lokal."""
    target_path = OUTPUT_DIR / f"{sku}.png"
    try:
        async with session.get(image_url, timeout=30) as img_resp:
            if img_resp.status == 200:
                content = await img_resp.read()
                with open(target_path, "wb") as f:
                    f.write(content)
                logger.info(f"Tersimpan: {target_path}")
                return {"sku": sku, "status": "SUCCESS", "path": str(target_path)}
            else:
                logger.error(f"Gagal mengunduh gambar SKU {sku} dari URL: {image_url}")
                return None
    except Exception as e:
        logger.error(f"Terjadi error saat menulis file untuk SKU {sku}: {e}")
        return None

async def orchestrate_batch(csv_filepath: str):
    """Fungsi utama pengelola batch pipeline."""
    if not Path(csv_filepath).exists():
        logger.error(f"File katalog {csv_filepath} tidak ditemukan!")
        return

    # Baca CSV
    with open(csv_filepath, mode="r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        all_items = list(reader)

    completed_skus = load_checkpoint()
    items_to_process = [item for item in all_items if item["sku"] not in completed_skus]

    logger.info(f"Total katalog: {len(all_items)} SKU. Sudah selesai: {len(completed_skus)}. Tersisa: {len(items_to_process)} SKU.")

    if not items_to_process:
        logger.info("Semua SKU sudah selesai diproses.")
        return

    semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)
    timeout = aiohttp.ClientTimeout(total=120)

    async with aiohttp.ClientSession(timeout=timeout) as session:
        tasks = [
            generate_product_image(session, semaphore, item)
            for item in items_to_process
        ]

        # Jalankan tasks dengan logging progres berkala
        results = await asyncio.gather(*tasks, return_exceptions=True)

        for res in results:
            if isinstance(res, dict) and res.get("status") == "SUCCESS":
                completed_skus.add(res["sku"])

        save_checkpoint(completed_skus)
        logger.info(f"Pipeline selesai. Checkpoint diperbarui. Total sukses: {len(completed_skus)}/{len(all_items)}")

if __name__ == "__main__":
    # Contoh penggunaan
    CSV_CATALOG_PATH = "katalog_produk.csv"
    
    # Jalankan event loop
    asyncio.run(orchestrate_batch(CSV_CATALOG_PATH))

sku,product_name,surface,scene
SKU-BOT-001,Matte Black Stainless Steel Tumbler,oak wood kitchen island,minimalist kitchen with morning soft sun rays
SKU-SKIN-002,Organic Rose Face Serum 30ml,white Carrara marble tray,luxury bathroom with botanical plants in soft focus
SKU-SNK-003,Premium Roasted Almonds Pouch,dark slate counter,artisan cafe table setting with warm spotlight
SKU-SHO-004,Retro High-Top Canvas Sneakers,polished concrete floor,urban streetwear retail showroom with neon ambiance

[Prompt Layer 1: Foreground Core] ──► "Commercial product photography of Matte Black Tumbler..."
               │
[Prompt Layer 2: Contact Plane]   ──► "...placed on polished concrete floor with soft contact shadows..."
               │
[Prompt Layer 3: Depth & Vibe]     ──► "...modern studio background, 50mm lens, f/2.8 bokeh, soft rim light"