Gemini 3 Pro Image Batch API: 반값 요금과 이미지 회수까지

import json
from pathlib import Path

products = [
    ("mug-101-front", "흰색 도자기 머그잔"),
    ("mug-102-front", "남색 도자기 머그잔"),
]

with Path("catalog.jsonl").open("x", encoding="utf-8") as out:
    for key, product in products:
        row = {
            "key": key,
            "request": {
                "contents": [{
                    "role": "user",
                    "parts": [{"text": (
                        f"{product}의 정면 상품 사진을 만들어 주세요. "
                        "밝은 배경, 자연스러운 그림자, 글자 없이."
                    )}],
                }],
                "generationConfig": {
                    "responseModalities": ["TEXT", "IMAGE"],
                    "imageConfig": {
                        "aspectRatio": "1:1",
                        "imageSize": "2K",
                    },
                },
            },
        }
        out.write(json.dumps(row, ensure_ascii=False) + "\n")

from pathlib import Path
from google import genai
from google.genai import types

client = genai.Client()
job_path = Path("batch-job.txt")

# 같은 작업 디렉터리에서 제출을 다시 실행하지 않도록 예약합니다.
with job_path.open("x", encoding="utf-8") as saved:
    saved.write("SUBMISSION_IN_PROGRESS\n")
    saved.flush()
    uploaded = client.files.upload(
        file="catalog.jsonl",
        config=types.UploadFileConfig(
            display_name="catalog-ko",
            mime_type="jsonl",
        ),
    )
    job = client.batches.create(
        model="gemini-3-pro-image",
        src=uploaded.name,
        config={"display_name": "catalog-ko-20261006"},
    )
    saved.seek(0)
    saved.truncate()
    saved.write(job.name + "\n")
    saved.flush()

print(job.name)

import time
from pathlib import Path
from google import genai

client = genai.Client()
job_name = Path("batch-job.txt").read_text(encoding="utf-8").strip()
if not job_name.startswith("batches/"):
    raise RuntimeError("먼저 제출 결과와 실제 작업 이름을 확인하세요.")

terminal = {
    "JOB_STATE_SUCCEEDED", "JOB_STATE_FAILED",
    "JOB_STATE_CANCELLED", "JOB_STATE_EXPIRED",
}
deadline = time.monotonic() + 30 * 60

while True:
    job = client.batches.get(name=job_name)
    state = job.state.name
    if state in terminal:
        break
    remaining = deadline - time.monotonic()
    if remaining <= 0:
        raise TimeoutError(f"로컬 대기 종료. 나중에 같은 작업 조회: {job_name}")
    time.sleep(min(60, remaining))

if state != "JOB_STATE_SUCCEEDED":
    raise RuntimeError(f"종료 상태: {state}; 오류: {job.error}")
if not job.dest or not job.dest.file_name:
    raise RuntimeError("성공 상태지만 결과 파일 정보가 없습니다.")

payload = client.files.download(file=job.dest.file_name)
Path("results.jsonl").write_bytes(payload)
print("results.jsonl 저장 완료")

import base64
import binascii
import json
from pathlib import Path

EXT = {"image/png": ".png", "image/jpeg": ".jpg", "image/webp": ".webp"}

def parse_results(text, expected_keys):
    grouped, issues = {}, []
    for number, line in enumerate(text.splitlines(), 1):
        if not line.strip():
            continue
        try:
            row = json.loads(line)
            if not isinstance(row, dict):
                raise ValueError("object가 아닙니다")
        except (ValueError, TypeError) as exc:
            issues.append({"line": number, "reason": "invalid_json", "detail": str(exc)})
            continue
        key = row.get("key")
        if not isinstance(key, str) or key not in expected_keys:
            issues.append({"line": number, "reason": "unknown_key"})
            continue
        grouped.setdefault(key, []).append(row)

    images = {}
    for key in expected_keys:
        rows = grouped.get(key, [])
        if len(rows) != 1:
            reason = "missing_result" if not rows else "duplicate_key"
            issues.append({"key": key, "reason": reason})
            continue
        row = rows[0]
        status = row.get("status") or {}
        if row.get("error") or status.get("code", 0):
            issues.append({"key": key, "reason": "request_error", "detail": row})
            continue
        response = row.get("response") or {}
        found, invalid = [], False
        for candidate in response.get("candidates") or []:
            content = candidate.get("content") or {}
            for part in content.get("parts") or []:
                if part.get("thought"):
                    continue
                data = part.get("inlineData") or {}
                mime, encoded = data.get("mimeType"), data.get("data")
                if not data:
                    continue
                if mime not in EXT or not isinstance(encoded, str):
                    invalid = True
                    continue
                try:
                    blob = base64.b64decode(encoded, validate=True)
                except (binascii.Error, ValueError):
                    invalid = True
                    continue
                if blob:
                    found.append((EXT[mime], blob))
                else:
                    invalid = True
        if found:
            images[key] = found
        if invalid or not found:
            reason = "invalid_image_data" if invalid else "no_final_image"
            issues.append({"key": key, "reason": reason})
    return images, issues

if __name__ == "__main__":
    requests = [json.loads(line) for line in Path("catalog.jsonl").read_text(
        encoding="utf-8").splitlines() if line.strip()]
    keys = [row["key"] for row in requests]
    if len(keys) != len(set(keys)):
        raise ValueError("입력 key가 중복됩니다.")
    images, issues = parse_results(
        Path("results.jsonl").read_bytes().decode("utf-8"), set(keys))
    output = Path("saved-images")
    output.mkdir(exist_ok=False)
    manifest = {}
    for key, parts in images.items():
        # 입력 key를 파일 경로로 직접 쓰지 않습니다.
        index = keys.index(key)
        manifest[key] = []
        for n, (extension, blob) in enumerate(parts, 1):
            name = f"item-{index:06d}-{n}{extension}"
            (output / name).write_bytes(blob)
            manifest[key].append(name)
    (output / "manifest.json").write_text(
        json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
    (output / "issues.json").write_text(
        json.dumps(issues, ensure_ascii=False, indent=2), encoding="utf-8")