Gemini 3 Pro ImageのBatch APIは半額：料金の見積もりと画像の回収手順

画像出力：1,000 × 1,120 × 60 / 1,000,000 = 67.20米ドル
入力：1,000 × (200 + 560) × 1 / 1,000,000 = 0.76米ドル
小計：67.96米ドル
ここに出力テキスト・思考・利用した有料ツール等の費用を加える

{"key":"mug-blue-01","request":{"contents":[{"role":"user","parts":[{"text":"青い陶器のマグカップを、白い背景で商品写真風に描いてください。文字やロゴは入れないでください。"}]}],"generationConfig":{"responseModalities":["TEXT","IMAGE"],"imageConfig":{"aspectRatio":"1:1","imageSize":"2K"}}}}
{"key":"mug-cream-01","request":{"contents":[{"role":"user","parts":[{"text":"クリーム色の陶器のマグカップを、白い背景で商品写真風に描いてください。文字やロゴは入れないでください。"}]}],"generationConfig":{"responseModalities":["TEXT","IMAGE"],"imageConfig":{"aspectRatio":"1:1","imageSize":"2K"}}}}

import json
import os
import sys
import time
import uuid
from pathlib import Path
from google import genai
from google.genai import types

ROOT = Path("mug-batch")
ROOT.mkdir(exist_ok=True)
STATE = ROOT / "job.json"
TERMINAL = {
    "JOB_STATE_SUCCEEDED", "JOB_STATE_FAILED",
    "JOB_STATE_CANCELLED", "JOB_STATE_EXPIRED",
}

def save_state(data):
    tmp = ROOT / "job.tmp"
    with tmp.open("w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False)
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp, STATE)

mode = sys.argv[1] if len(sys.argv) == 2 else ""
if mode not in {"submit", "collect"}:
    raise SystemExit("使い方: python batch_images.py submit | collect")
client = genai.Client()

if mode == "submit":
    if STATE.exists():
        raise SystemExit("job.jsonが既にあります。再送せず状態を確認してください。")
    source = Path("requests.jsonl")
    keys = [json.loads(line)["key"] for line in
            source.read_text(encoding="utf-8").splitlines() if line.strip()]
    if not keys or len(keys) != len(set(keys)):
        raise SystemExit("要求が空、またはキーが重複しています。")
    data = {"display_name": "mug-" + uuid.uuid4().hex, "keys": keys}
    save_state(data)  # 登録中に中断しても、無条件の再送を防ぐ
    uploaded = client.files.upload(
        file=str(source),
        config=types.UploadFileConfig(mime_type="jsonl"),
    )
    data["input_file"] = uploaded.name
    save_state(data)
    job = client.batches.create(
        model="gemini-3-pro-image",
        src=uploaded.name,
        config={"display_name": data["display_name"]},
    )
    data["job_name"] = job.name
    save_state(data)
    print(job.name)

else:
    data = json.loads(STATE.read_text(encoding="utf-8"))
    if not data.get("job_name"):
        raise SystemExit("登録結果が不明です。display_nameで既存ジョブを照合してください。")
    deadline = time.monotonic() + 1800
    while True:
        job = client.batches.get(name=data["job_name"])
        state = job.state.name
        print(state)
        if state in TERMINAL:
            break
        if time.monotonic() >= deadline:
            raise SystemExit("30分の待機を終了しました。後でcollectを再実行できます。")
        time.sleep(60)
    if state != "JOB_STATE_SUCCEEDED":
        raise SystemExit("終了状態を確認してください: " + state)
    if not job.dest or not job.dest.file_name:
        raise SystemExit("出力ファイルが見つかりません。ジョブ情報を確認してください。")
    raw = client.files.download(file=job.dest.file_name)
    (ROOT / "results.jsonl").write_bytes(raw)
    print("保存先:", ROOT / "results.jsonl")

import base64
import binascii
import json
from pathlib import Path

root = Path("mug-batch")
job = json.loads((root / "job.json").read_text(encoding="utf-8"))
expected = job["keys"]
index = {key: n for n, key in enumerate(expected, 1)}
out = root / "images"
out.mkdir(exist_ok=True)
extensions = {"image/png": ".png", "image/jpeg": ".jpg", "image/webp": ".webp"}
seen, report, anomalies = set(), {}, []

for line_no, line in enumerate(
    (root / "results.jsonl").read_bytes().decode("utf-8").splitlines(), 1
):
    if not line.strip():
        continue
    try:
        row = json.loads(line)
    except json.JSONDecodeError:
        anomalies.append({"line": line_no, "reason": "invalid_json"})
        continue
    key = row.get("key")
    if key not in index or key in seen:
        anomalies.append({"line": line_no, "reason": "unknown_or_duplicate_key", "key": key})
        continue
    seen.add(key)
    response = row.get("response") or {}
    item = {"files": [], "text": [], "finish_reasons": [],
            "usage": response.get("usageMetadata"),
            "error": row.get("error"), "status": row.get("status")}
    for candidate in response.get("candidates", []):
        item["finish_reasons"].append(candidate.get("finishReason"))
        for part in (candidate.get("content") or {}).get("parts", []):
            if part.get("thought"):
                continue
            if "text" in part:
                item["text"].append(part["text"])
            blob = part.get("inlineData") or {}
            ext = extensions.get(blob.get("mimeType"))
            if not ext or not blob.get("data"):
                continue
            try:
                image = base64.b64decode(blob["data"], validate=True)
            except (binascii.Error, ValueError):
                anomalies.append({"key": key, "reason": "invalid_base64"})
                continue
            if not image:
                continue
            name = f"{index[key]:05d}-{len(item['files']) + 1}{ext}"
            (out / name).write_bytes(image)
            item["files"].append(name)
    item["result"] = "saved" if item["files"] else "no_image"
    report[key] = item

for key in expected:
    if key not in seen:
        report[key] = {"result": "missing", "files": []}
needs_check = [key for key, item in report.items()
               if item["result"] != "saved" or item.get("error") or item.get("status")]
(root / "report.json").write_text(
    json.dumps({"items": report, "needs_check": needs_check, "anomalies": anomalies},
               ensure_ascii=False, indent=2), encoding="utf-8"
)
print("画像保存済み要求:", sum(bool(item["files"]) for item in report.values()))
print("確認が必要なキー:", needs_check)
print("行・データの異常:", len(anomalies))