Gamma：批量生成时，轮询本身就会消耗限流额度，导出链接也容易被泄露

批量生成时，轮询本身就会消耗限流额度，导出链接也容易被泄露。这篇做两件事：

• 自适应轮询：读取每次响应里的 x-ratelimit-remaining-burst，低于 100 时从 5 秒放慢到 15 秒（这是官方文档示例的做法）；遇到 429 按 Retry-After 等待，遇到 5xx 指数退避。
• 导出链接保管：exportUrl 只存在权限 600 的 exports.json 里，并记录大约一周后的过期时间；list 命令只显示脱敏后的链接，purge 命令清理过期记录。

先看清成品（box 上对着模拟服务器运行；模拟器开启了“剩余配额很少”和“先返回一次 429”）

• 日志依次显示：429 等待 2 秒 → 剩余 100 时 1 秒后再查 → 剩余 99 时放慢到 3 秒（演示时把 5 秒和 15 秒缩短成了 1 秒和 3 秒）。
• 保管文件的权限是 -rw-------。
• pytest 2 项通过。

合规提醒：调用 Gamma API 需要 Pro、Ultra、Teams、Business 或 Enterprise 计划的 API Key（sk-gamma-…），Key 只放在服务器端。exportUrl 在大约一周内任何拿到链接的人都能下载，要当作密钥来保管。输入的文字和图片必须是你有权使用的；开启网络图片来源时，要注意图片的授权条款，商用场景优先选 webFreeToUseCommercially 或 noImages。在中国境内公开发布 AI 生成的内容，要遵守《生成式人工智能服务管理暂行办法》，并按《人工智能生成合成内容标识办法》加上 AI 标识。Gamma 按额度（credits）计费，本文不列价格。

运行说明：box 上没有 Gamma API Key，所以没有真正生成过文稿。流程跑在我自己写的模拟服务器 fake_gamma.py 上，导出的 PDF 是 Pillow 画的，额度数字也是模拟器编的。用错误的 Key 请求真实接口确实运行过，返回 401 {"message":"Invalid API key","statusCode":401}。

做完之后你会得到什么？

需要准备哪些环境？

和 A 篇相同。

限流规则（来自官方文档）

• 每个响应都带有这些头：x-ratelimit-limit/remaining-burst（短时间窗口）、x-ratelimit-limit/remaining（每小时）、x-ratelimit-limit/remaining-daily（每天）。
• 官方的自适应示例：remaining-burst 低于 100 时，轮询间隔从 5 秒改为 15 秒。
• 收到 429 时按 Retry-After 等待；Agent 接口可能返回 workspace_concurrency_limit_exceeded。
• 出现 500 错误时，联系客服要提供 x-request-id 响应头。

文件结构

53-gamma-C-src/
├── poller.py
├── .gitignore
├── gamma_deck.py
├── fake_gamma.py
├── outline.md
└── test_poller.py

第 1 步：代码

"""自适应轮询 + 导出链接保管。
轮询：默认每 5 秒一次；x-ratelimit-remaining-burst 低于 100 时放慢到 15 秒（官方文档示例的做法）；
     429 按 Retry-After 等待；5xx/502 指数退避（最长 60 秒）。
保管：exportUrl 约一周内任何人都能下载，所以只存在权限 600 的 exports.json 里，记录过期时间，日志里一律脱敏；
     拿到后立即下载，过期的记录可以清理。"""
import json, os, pathlib, sys, time
import requests
import gamma_deck as g

VAULT = pathlib.Path(os.environ.get("GAMMA_VAULT", "exports.json"))
WEEK = 7 * 24 * 3600


def next_interval(resp, base=5.0, slow=15.0, low=100):
    try:
        burst = int(resp.headers.get("x-ratelimit-remaining-burst", "9999"))
    except ValueError:
        burst = 9999
    return slow if burst < low else base, burst


def poll(s, gid, base=5.0, slow=15.0, timeout=900, sleep=time.sleep, log=print):
    t0, backoff = time.time(), base
    while time.time() - t0 < timeout:
        r = s.get(f"{g.API}/v1.0/generations/{gid}", timeout=60)
        if r.status_code == 429:
            wait = int(r.headers.get("Retry-After", "10")); log(f"429 → 按 Retry-After 等 {wait}s"); sleep(wait); continue
        if r.status_code >= 500:
            backoff = min(backoff * 2, 60); log(f"{r.status_code} → 退避 {backoff:.0f}s"); sleep(backoff); continue
        if r.status_code >= 400:
            raise SystemExit(f"HTTP {r.status_code}: {r.text[:200]}")
        d = r.json()
        if d["status"] in ("completed", "failed"):
            return d
        iv, burst = next_interval(r, base, slow)
        log(f"pending｜remaining-burst={burst}｜{iv:.0f}s 后再查"); sleep(iv); backoff = base
    raise TimeoutError(gid)


def vault_load():
    return json.loads(VAULT.read_text()) if VAULT.exists() else {}


def vault_save(v):
    VAULT.write_text(json.dumps(v, ensure_ascii=False, indent=1)); os.chmod(VAULT, 0o600)


def keep(gid, d, outdir="outputs", ext="pdf"):
    v = vault_load()
    v[gid] = {"exportUrl": d["exportUrl"], "gammaUrl": d.get("gammaUrl"), "fetched_at": int(time.time()),
              "expires_at": int(time.time()) + WEEK}
    vault_save(v)
    out = pathlib.Path(outdir); out.mkdir(exist_ok=True)
    f = out / f"{gid}.{ext}"; f.write_bytes(requests.get(d["exportUrl"], timeout=300).content)
    return f


def purge(now=None):
    now = now or time.time(); v = vault_load()
    gone = [k for k, x in v.items() if x["expires_at"] < now]
    for k in gone: del v[k]
    vault_save(v); return gone


def main():
    cmd = sys.argv[1]
    if cmd == "run":
        s = g.session()
        gid = g.create(s, g.body_from_outline(pathlib.Path(sys.argv[2]).read_text(encoding="utf-8")))
        d = poll(s, gid, base=float(os.environ.get("POLL_BASE", 5)), slow=float(os.environ.get("POLL_SLOW", 15)))
        if d["status"] != "completed":
            raise SystemExit("失败：" + json.dumps(d.get("error"), ensure_ascii=False))
        f = keep(gid, d)
        print(f"✓ {gid} 已下载 {f}；导出链接已存入 {VAULT}（600），日志只显示 {g.mask(d['exportUrl'])}")
    elif cmd == "list":
        for k, x in vault_load().items():
            left = (x["expires_at"] - time.time()) / 86400
            print(f"{k}  {g.mask(x['exportUrl'])}  约 {left:.1f} 天后过期")
    elif cmd == "purge":
        print("已清理：", purge() or "无")


if __name__ == "__main__":
    main()

第 2 步：测试

import os, stat, time, pytest
import poller


class R:
    def __init__(self, code, body=None, headers=None):
        self.status_code, self._b, self.headers, self.text = code, body or {}, headers or {}, ""
    def json(self): return self._b


class S:
    def __init__(self, seq): self.seq = list(seq)
    def get(self, url, **kw): return self.seq.pop(0)


def test_adaptive_and_retry():
    seq = [R(200, {"status": "pending"}, {"x-ratelimit-remaining-burst": "5000"}),
           R(200, {"status": "pending"}, {"x-ratelimit-remaining-burst": "42"}),
           R(429, headers={"Retry-After": "3"}), R(502), R(200, {"status": "completed"})]
    waits = []
    d = poller.poll(S(seq), "g", sleep=waits.append, log=lambda *a: None)
    assert d["status"] == "completed" and waits == [5.0, 15.0, 3, 10.0]


def test_vault_perms_and_purge(tmp_path, monkeypatch):
    monkeypatch.setattr(poller, "VAULT", tmp_path / "v.json")
    poller.vault_save({"old": {"exportUrl": "u", "expires_at": 1}, "new": {"exportUrl": "u", "expires_at": time.time() + 99}})
    assert stat.S_IMODE(os.stat(tmp_path / "v.json").st_mode) == 0o600
    assert poller.purge() == ["old"] and list(poller.vault_load()) == ["new"]

第 3 步：运行

$ FAKE_BURST=103 FAKE_429=1 python fake_gamma.py &
$ export GAMMA_API_BASE=http://127.0.0.1:18531 GAMMA_API_KEY=sk-gamma-demo POLL_BASE=1 POLL_SLOW=3
$ python poller.py run outline.md
429 → 按 Retry-After 等 2s
pending｜remaining-burst=100｜1s 后再查
pending｜remaining-burst=99｜3s 后再查
✓ fcd518a7c7d04c8caadf 已下载 outputs/fcd518a7c7d04c8caadf.pdf；导出链接已存入 exports.json（600），日志只显示 http://127.0.0.1:18531/…(已隐藏)

$ python poller.py list
fcd518a7c7d04c8caadf  http://127.0.0.1:18531/…(已隐藏)  约 7.0 天后过期
$ ls -l exports.json
-rw------- …
$ python poller.py purge
已清理： 无

$ python -m pytest -q
2 passed in 0.08s

调试示例

为什么剩余 100 时没有放慢？ 日志中 remaining-burst=100 那次仍然是 1 秒，到 99 才放慢。这和官方示例的写法 if remaining < 100 一致：阈值本身不包括在内。如果你希望留出更多余量（比如同时跑多个任务），可以在调用 next_interval() 时调大 low 参数。

单元测试 test_adaptive_and_retry 固定了整个等待序列 [5, 15, 3, 10]：正常间隔 → 配额低时放慢 → 按 Retry-After 等待 → 502 退避一次（5 秒翻倍成 10 秒）。以后修改策略时，这个测试就是回归基准。

常见问题

模板规则

• 轮询间隔由响应头决定，不要写死。
• 429 按 Retry-After 等待，5xx 指数退避。
• 导出链接按密钥对待：文件权限 600、日志脱敏、记录过期时间、不提交进 git。

项目资源

轮询有节制、链接有保管，批量生成文稿就可以放心交给定时任务了。