Nano Banana Pro RESOURCE_EXHAUSTED 429：排查与 Python、Node.js 恢复代码

python3 -m pip install Pillow

[
  {"id": "tea-001", "prompt": "生成一张白底产品照片：一杯透明玻璃杯装的热茶"},
  {"id": "tea-002", "prompt": "生成一张白底产品照片：一杯陶瓷杯装的热茶"}
]

import base64, hashlib, io, json, os, queue, random, re, sys, threading, time
from email.utils import parsedate_to_datetime
from pathlib import Path
from urllib.request import Request, urlopen
from urllib.error import HTTPError
from PIL import Image

MODEL = "gemini-3-pro-image"
URL = f"https://generativelanguage.googleapis.com/v1beta/models/{MODEL}:generateContent"

def post(payload, key, timeout):
    # urllib 无 SDK 重试；整个收响应过程有调用方等待期限。
    box = queue.Queue()
    def worker():
        try:
            request = Request(URL, data=json.dumps(payload).encode(), headers={
                "x-goog-api-key": key, "Content-Type": "application/json"})
            try:
                response = urlopen(request, timeout=timeout)
            except HTTPError as error:
                response = error
            with response:
                result = (response.code, dict(response.headers), response.read())
            box.put((True, result))
        except Exception as error:
            box.put((False, error))
    threading.Thread(target=worker, daemon=True).start()
    try:
        ok, value = box.get(timeout=timeout)
    except queue.Empty:
        raise TimeoutError("请求结果未知") from None
    if not ok:
        raise value
    return value

def decision(status, headers, body, wall=time.time):
    details = body.get("error", {}).get("details", [])
    minimum = 0.0
    header = next((v for k, v in headers.items() if k.lower() == "retry-after"), "")
    if re.fullmatch(r"\d+", header):
        minimum = float(header)
    elif header:
        try:
            minimum = max(0, parsedate_to_datetime(header).timestamp() - wall())
        except (ValueError, TypeError, OverflowError):
            pass
    for detail in details:
        if detail.get("@type", "").endswith("google.rpc.RetryInfo"):
            delay = detail.get("retryDelay", "")
            if re.fullmatch(r"\d+(?:\.\d+)?s", delay):
                minimum = max(minimum, float(delay[:-1]))
    if status == 429:
        message = body.get("error", {}).get("message", "").lower()
        if re.search(r"\blimit\s*:\s*0(?:\D|$)", message):
            return "blocked", minimum
        if "monthly" in message or "spending cap" in message:
            return "blocked", minimum
        for detail in details:
            if not detail.get("@type", "").endswith("google.rpc.QuotaFailure"):
                continue
            for violation in detail.get("violations", []):
                if "quotaValue" in violation and str(violation["quotaValue"]) == "0":
                    return "blocked", minimum
                label = " ".join(str(violation.get(k, "")) for k in
                                 ("quotaMetric", "quotaId", "description")).lower()
                if any(token in label for token in ("perday", "per_day", "daily")):
                    return "daily", minimum
    return ("retry" if status in (429, 503) else "blocked"), minimum

def image_bytes(body):
    for candidate in body.get("candidates", []):
        for part in candidate.get("content", {}).get("parts", []):
            if part.get("thought"):
                continue
            blob = part.get("inlineData", {})
            if blob.get("mimeType") not in ("image/png", "image/jpeg", "image/webp"):
                continue
            raw = base64.b64decode(blob.get("data", ""), validate=True)
            with Image.open(io.BytesIO(raw)) as image:
                image.load()
                actual = Image.MIME.get(image.format)
                if actual != blob["mimeType"]:
                    raise ValueError("图片 MIME 与内容不符")
                output = io.BytesIO()
                image.save(output, format="PNG")
            return output.getvalue()
    raise ValueError("200 响应没有可保存的图片")

def atomic(path, raw):
    temp = path.with_suffix(path.suffix + ".tmp")
    temp.write_bytes(raw)
    temp.replace(path)

def run(jobs, folder, key, send=post, clock=time.monotonic,
        sleep=time.sleep, jitter=random.random, budget=120.0, wall=time.time):
    folder = Path(folder)
    folder.mkdir(parents=True, exist_ok=True)
    ledger = folder / "state.json"
    state = json.loads(ledger.read_text()) if ledger.exists() else {}
    def save():
        atomic(ledger, json.dumps(state, ensure_ascii=False, indent=2).encode())
    ids = [job["id"] for job in jobs]
    if len(ids) != len(set(ids)) or any(not re.fullmatch(r"[a-zA-Z0-9_-]+", x) for x in ids):
        raise ValueError("job id 必须唯一，只用字母、数字、下划线或横线")
    for job in jobs:
        payload = {"contents": [{"parts": [{"text": job["prompt"]}]}],
                   "generationConfig": {"responseModalities": ["IMAGE"],
                   "imageConfig": {"aspectRatio": "1:1", "imageSize": "1K"}}}
        fingerprint = hashlib.sha256(json.dumps([MODEL, payload], sort_keys=True).encode()).hexdigest()
        record = state.setdefault(job["id"], {"fingerprint": fingerprint, "status": "pending"})
        if record["fingerprint"] != fingerprint:
            raise ValueError("同一 id 的请求已改变；请使用新 id")
        output = folder / (job["id"] + ".png")
        if record["status"] == "done":
            if not output.exists() or hashlib.sha256(output.read_bytes()).hexdigest() != record["sha256"]:
                record["status"] = "needs_check"
                save()
            continue
        if record["status"] != "pending" or wall() < record.get("notBefore", 0):
            continue
        deadline = clock() + budget
        for attempt in range(1, 5):
            remaining = deadline - clock()
            if remaining <= 0:
                break
            record["status"] = "unknown"  # 在发送前持久化，崩溃后不盲目重发。
            save()
            try:
                remaining = deadline - clock()
                if remaining <= 0:
                    record["status"] = "pending"
                    save()
                    break
                status, headers, raw = send(payload, key, min(45.0, remaining))
            except Exception:
                break  # 超时/断连，后台可能仍完成，保留 unknown。
            try:
                body = json.loads(raw)
                if not isinstance(body, dict):
                    raise ValueError("响应不是对象")
                if status == 200:
                    image = image_bytes(body)
                    atomic(output, image)
                    record.update(status="done", sha256=hashlib.sha256(image).hexdigest(),
                                  responseId=body.get("responseId"))
                    save()
                    break
                action, minimum = decision(status, headers, body, wall=wall)
            except Exception:
                record["status"] = "needs_check"
                save()
                break
            if action != "retry":
                record.update(status=action, http=status)
                save()
                break
            wait = max(minimum, min(2 ** attempt, 30) + jitter())
            record.update(status="pending", http=status, notBefore=wall() + wait)
            save()
            if attempt == 4:
                break
            if wait >= deadline - clock():
                break  # 放回待处理，不能把服务端 minimum 截短。
            sleep(wait)
    return state

if __name__ == "__main__":
    key = os.environ.get("GEMINI_API_KEY")
    if not key:
        raise SystemExit("请在运行环境配置 GEMINI_API_KEY")
    jobs = json.loads(Path(sys.argv[1]).read_text())
    result = run(jobs, sys.argv[2], key)
    print(json.dumps({k: v["status"] for k, v in result.items()}, ensure_ascii=False))

npm install sharp

import fs from 'node:fs';
import path from 'node:path';
import crypto from 'node:crypto';
import { pathToFileURL } from 'node:url';
import sharp from 'sharp';

const MODEL = 'gemini-3-pro-image';
const URL = `https://generativelanguage.googleapis.com/v1beta/models/${MODEL}:generateContent`;
const hash = value => crypto.createHash('sha256').update(value).digest('hex');
const atomic = (file, raw) => {
  fs.writeFileSync(file + '.tmp', raw);
  fs.renameSync(file + '.tmp', file);
};

export async function post(payload, key, timeout) {
  const response = await fetch(URL, {
    method: 'POST', headers: { 'Content-Type': 'application/json', 'x-goog-api-key': key },
    body: JSON.stringify(payload), signal: AbortSignal.timeout(Math.max(1, Math.floor(timeout * 1000))),
    redirect: 'error',
  });
  // timeout signal 同时覆盖读取响应体，不只覆盖响应头。
  return [response.status, Object.fromEntries(response.headers), await response.text()];
}

export function decision(status, headers, body, wall = Date.now) {
  const details = body.error?.details ?? [];
  const header = Object.entries(headers).find(([k]) => k.toLowerCase() === 'retry-after')?.[1];
  let minimum = /^\d+$/.test(header ?? '') ? Number(header) :
    Math.max(0, (Date.parse(header ?? '') - wall()) / 1000);
  if (!Number.isFinite(minimum)) minimum = 0;
  for (const detail of details) {
    if (detail['@type']?.endsWith('google.rpc.RetryInfo') && /^\d+(?:\.\d+)?s$/.test(detail.retryDelay ?? '')) {
      minimum = Math.max(minimum, Number(detail.retryDelay.slice(0, -1)));
    }
  }
  if (status === 429) {
    const message = (body.error?.message ?? '').toLowerCase();
    if (/\blimit\s*:\s*0(?:\D|$)/.test(message) || message.includes('monthly') || message.includes('spending cap')) {
      return ['blocked', minimum];
    }
    for (const detail of details) {
      if (!detail['@type']?.endsWith('google.rpc.QuotaFailure')) continue;
      for (const violation of detail.violations ?? []) {
        if (Object.hasOwn(violation, 'quotaValue') && String(violation.quotaValue) === '0') {
          return ['blocked', minimum];
        }
        const label = ['quotaMetric', 'quotaId', 'description'].map(k => violation[k] ?? '').join(' ').toLowerCase();
        if (['perday', 'per_day', 'daily'].some(token => label.includes(token))) return ['daily', minimum];
      }
    }
  }
  return [status === 429 || status === 503 ? 'retry' : 'blocked', minimum];
}

export async function imageBytes(body) {
  for (const candidate of body.candidates ?? []) {
    for (const part of candidate.content?.parts ?? []) {
      if (part.thought) continue;
      const blob = part.inlineData ?? {};
      const formats = { 'image/png': 'png', 'image/jpeg': 'jpeg', 'image/webp': 'webp' };
      if (!formats[blob.mimeType]) continue;
      const encoded = blob.data ?? '';
      if (!encoded || !/^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$/.test(encoded)) {
        throw Error('图片 base64 无效');
      }
      const raw = Buffer.from(encoded, 'base64');
      const metadata = await sharp(raw).metadata();
      if (metadata.format !== formats[blob.mimeType]) throw Error('图片 MIME 与内容不符');
      return sharp(raw).png().toBuffer(); // 实际解码，而不是只信 mimeType。
    }
  }
  throw Error('200 响应没有可保存的图片');
}

export async function run(jobs, folder, key, options = {}) {
  const { send = post, clock = () => performance.now() / 1000,
    sleep = seconds => new Promise(resolve => setTimeout(resolve, seconds * 1000)),
    jitter = Math.random, budget = 120, wall = () => Date.now() / 1000 } = options;
  fs.mkdirSync(folder, { recursive: true });
  const ledger = path.join(folder, 'state.json');
  const state = fs.existsSync(ledger) ? JSON.parse(fs.readFileSync(ledger, 'utf8')) : {};
  const save = () => atomic(ledger, JSON.stringify(state, null, 2));
  const ids = jobs.map(job => job.id);
  if (new Set(ids).size !== ids.length || ids.some(id => !/^[A-Za-z0-9_-]+$/.test(id))) {
    throw Error('job id 必须唯一，只用字母、数字、下划线或横线');
  }
  for (const job of jobs) {
    const payload = { contents: [{ parts: [{ text: job.prompt }] }], generationConfig: {
      responseModalities: ['IMAGE'], imageConfig: { aspectRatio: '1:1', imageSize: '1K' },
    } };
    const fingerprint = hash(JSON.stringify([MODEL, payload]));
    const record = state[job.id] ??= { fingerprint, status: 'pending' };
    if (record.fingerprint !== fingerprint) throw Error('同一 id 的请求已改变；请使用新 id');
    const output = path.join(folder, job.id + '.png');
    if (record.status === 'done') {
      if (!fs.existsSync(output) || hash(fs.readFileSync(output)) !== record.sha256) {
        record.status = 'needs_check'; save();
      }
      continue;
    }
    if (record.status !== 'pending' || wall() < (record.notBefore ?? 0)) continue;
    const deadline = clock() + budget;
    for (let attempt = 1; attempt <= 4; attempt++) {
      const remaining = deadline - clock();
      if (remaining <= 0) break;
      record.status = 'unknown'; save();
      let status, headers, raw;
      try {
        const available = deadline - clock();
        if (available <= 0) { record.status = 'pending'; save(); break; }
        [status, headers, raw] = await send(payload, key, Math.min(45, available));
      } catch { break; } // 结果未知，不自动再次发送。
      let action, minimum;
      try {
        const body = JSON.parse(raw);
        if (!body || typeof body !== 'object' || Array.isArray(body)) throw Error('响应不是对象');
        if (status === 200) {
          const image = await imageBytes(body);
          atomic(output, image);
          Object.assign(record, { status: 'done', sha256: hash(image), responseId: body.responseId });
          save(); break;
        }
        [action, minimum] = decision(status, headers, body, () => wall() * 1000);
      } catch { record.status = 'needs_check'; save(); break; }
      if (action !== 'retry') {
        Object.assign(record, { status: action, http: status }); save(); break;
      }
      const wait = Math.max(minimum, Math.min(2 ** attempt, 30) + jitter());
      Object.assign(record, { status: 'pending', http: status, notBefore: wall() + wait }); save();
      if (attempt === 4) break;
      if (wait >= deadline - clock()) break;
      await sleep(wait);
    }
  }
  return state;
}

if (process.argv[1] && import.meta.url === pathToFileURL(path.resolve(process.argv[1])).href) {
  const key = process.env.GEMINI_API_KEY;
  if (!key) throw Error('请在运行环境配置 GEMINI_API_KEY');
  const jobs = JSON.parse(fs.readFileSync(process.argv[2], 'utf8'));
  const result = await run(jobs, process.argv[3], key);
  console.log(Object.fromEntries(Object.entries(result).map(([id, value]) => [id, value.status])));
}