Gemini 3 Pro Image Batch API 五折：批量生图费用与 Python 实现

文本输入：1,000 × 300 × 1 / 1,000,000 = 0.30 美元
参考图片：1,000 × 2 × 560 × 1 / 1,000,000 = 1.12 美元
图片输出：1,000 × 0.0672 = 67.20 美元
文本及思考：1,000 × 1,000 × 6 / 1,000,000 = 6.00 美元
合计：74.62 美元

# submit_batch.py
import json
import os
from pathlib import Path
from google import genai
from google.genai import types

MODEL = "gemini-3-pro-image"
manifest = Path("batch-manifest.json")
input_path = Path("image-requests.jsonl")

def save_manifest(value):
    temporary = manifest.with_suffix(".tmp")
    with temporary.open("w", encoding="utf-8") as handle:
        json.dump(value, handle, ensure_ascii=False, indent=2)
        handle.flush()
        os.fsync(handle.fileno())
    os.replace(temporary, manifest)

if manifest.exists():
    saved = json.loads(manifest.read_text(encoding="utf-8"))
    raise SystemExit(
        "已有提交记录，请恢复或核对，勿重复创建：" + str(saved)
    )

prompts = {
    "cup-main-v1": "生成白色陶瓷杯的电商主图，浅灰背景，柔和自然光。",
    "cup-scene-v1": "生成白色陶瓷杯放在木质餐桌上的场景图，晨光。",
}
with input_path.open("w", encoding="utf-8") as handle:
    for key, prompt in prompts.items():
        row = {
            "key": key,
            "request": {
                "contents": [{
                    "role": "user", "parts": [{"text": prompt}]
                }],
                "generationConfig": {
                    "responseModalities": ["TEXT", "IMAGE"],
                    "imageConfig": {"imageSize": "2K", "aspectRatio": "1:1"},
                },
            },
        }
        handle.write(json.dumps(row, ensure_ascii=False) + "\n")

client = genai.Client()
uploaded = client.files.upload(
    file=str(input_path),
    config=types.UploadFileConfig(
        display_name="cup-images-input", mime_type="jsonl"
    ),
)
# 先记录提交意图；若创建结果不明，保留这份记录并人工核对。
record = {
    "phase": "submitting", "model": MODEL,
    "input_file": uploaded.name, "keys": list(prompts),
}
save_manifest(record)
job = client.batches.create(
    model=MODEL, src=uploaded.name,
    config={"display_name": "cup-images-v1"},
)
record.update(phase="submitted", job_name=job.name)
save_manifest(record)
print("已保存任务名：", job.name)

# recover_images.py：纯本地 JSONL 解析，不调用 API
import base64
import json
import re
from pathlib import Path

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

def recover(raw_bytes, expected_keys, output_dir):
    expected = set(expected_keys)
    if len(expected) != len(expected_keys):
        raise ValueError("输入清单含重复键")
    if any(not re.fullmatch(r"[A-Za-z0-9_-]+", key) for key in expected):
        raise ValueError("本示例的键只允许字母、数字、下划线和连字符")
    output = Path(output_dir)
    output.mkdir(parents=True, exist_ok=True)
    rows, issues = {}, []
    for number, line in enumerate(raw_bytes.decode("utf-8").splitlines(), 1):
        if not line.strip():
            continue
        try:
            row = json.loads(line)
            if not isinstance(row, dict):
                raise ValueError("结果行不是对象")
            key = row.get("key")
            if not isinstance(key, str) or key not in expected:
                raise ValueError("结果键缺失或不在输入清单中")
            rows.setdefault(key, []).append(row)
        except (ValueError, TypeError) as error:
            issues.append({"line": number, "reason": str(error)})

    report = {}
    for key in sorted(expected):
        matching = rows.get(key, [])
        if len(matching) != 1:
            report[key] = {"state": "missing" if not matching else "duplicate"}
            continue
        row = matching[0]
        status = row.get("status") or {}
        if row.get("error") or status.get("code", 0) != 0:
            report[key] = {"state": "error", "detail": row.get("error") or status}
            continue
        response = row.get("response") or {}
        files, notes = [], []
        for candidate in response.get("candidates") or []:
            for part in (candidate.get("content") or {}).get("parts") or []:
                if part.get("thought"):
                    continue
                image = part.get("inlineData") or {}
                if not image:
                    continue
                extension = EXTENSIONS.get(image.get("mimeType"))
                if not extension:
                    notes.append("图片 MIME 类型未获本示例支持")
                    continue
                try:
                    data = base64.b64decode(image.get("data", ""), validate=True)
                    if not data:
                        raise ValueError("图片数据为空")
                    name = f"{key}-{len(files) + 1}{extension}"
                    (output / name).write_bytes(data)
                    files.append(name)
                except (ValueError, TypeError) as error:
                    notes.append(str(error))
        report[key] = {
            "state": "saved" if files else "no_image",
            "files": files, "notes": notes,
        }
    result = {"requests": report, "parse_issues": issues}
    (output / "report.json").write_text(
        json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    return result

# collect_batch.py
import json
import time
from pathlib import Path
from google import genai
from recover_images import recover

saved = json.loads(Path("batch-manifest.json").read_text(encoding="utf-8"))
if not saved.get("job_name"):
    raise SystemExit("缺少任务名，请先核对提交结果，勿重新创建")
client = genai.Client()
terminal = {
    "JOB_STATE_SUCCEEDED", "JOB_STATE_FAILED",
    "JOB_STATE_CANCELLED", "JOB_STATE_EXPIRED",
}
deadline = time.monotonic() + 30 * 60
job = None
while time.monotonic() < deadline:
    job = client.batches.get(name=saved["job_name"])
    state = job.state.name
    print(saved["job_name"], state)
    if state in terminal:
        break
    remaining = deadline - time.monotonic()
    if remaining > 0:
        time.sleep(min(60, remaining))
else:
    raise SystemExit("本轮停止等待，稍后用原任务名继续查询")

if state != "JOB_STATE_SUCCEEDED":
    raise SystemExit(f"任务终止：{state}；请保存错误与状态后检查")
if not job.dest or not job.dest.file_name:
    raise SystemExit("成功状态没有结果文件名，请保留任务名并核对响应")
raw = client.files.download(file=job.dest.file_name)
Path("batch-results.jsonl").write_bytes(raw)
result = recover(raw, saved["keys"], "images-output")
print(json.dumps(result, ensure_ascii=False, indent=2))