Leonardo AI：一次完整的 Leonardo 生成流程：创建任务 → 轮询状态 → 下载图片 → 写清单

一次完整的 Leonardo 生成流程：创建任务 → 轮询状态 → 下载图片 → 写清单。 几个要点：

• 轮询间隔逐渐加长，并设置总超时；
• 只有 COMPLETE 状态才读取图片（官方 FAQ：PENDING 时 generated_images 是空的）；
• nsfw: true 的图片不下载；
• 每张图片记录 sha256。 另外，v1 和 v2 的创建接口返回的结构不一样，代码对两种都做了兼容。

先看清成品

• 本地替身上：轮询 3 次后完成，等待间隔是 2 秒和 3 秒；下载 1 张图片，另一张因为 nsfw 被跳过；清单里记录了提示词和 sha256。失败状态、超时、内容审核拦截（400）、鉴权失败（401）都有测试。pytest 5 项通过。
• 用无效 Key 请求真实接口：✗ HTTP 401 [access-denied] Authentication hook unauthorized this request，退出码为 1。

合规提醒：Leonardo API 需要单独购买 API 额度，和网页版订阅是分开的。API Key 只放在服务端环境变量里，不要放进前端代码，也不要共享账号或 Key。生成内容要遵守 Leonardo 的条款和内容政策：官方会默认拦截 NSFW 提示词，并在结果里用 nsfw 字段标记图片。不要尝试绕过审核，也不要生成侵犯他人肖像权、商标或版权的内容。请求体里的 public 字段决定图片是否出现在社区公开流中，处理敏感素材时设为 false。本文不列价格。

运行说明：box 上没有 Leonardo API Key 和 API 额度，所以没有真正生成图片。真实运行过的部分是：用无效 Key 请求 v1 和 v2 接口，都返回 HTTP 401 {{"error":"Authentication hook unauthorized this request","path":"$","code":"access-denied"}}。成功路径用的是我自己写的本地替身；替身的格式依据的是官方文档中的示例（FAQ、NSFW 指南、webhook 指南、v2 OpenAPI 规范）。

做完之后你会得到什么？

需要准备哪些环境？

Python 3.9 以上（只用标准库）；测试需要 pip install pytest。真实运行需要 LEONARDO_API_KEY 和 API 额度。

关键事实

文件结构

63-leonardo-ai-B-src/
├── leo_run.py
├── request.json
├── fake_leo.py
└── test_leo_run.py

第 1 步：主流程

"""Leonardo 生成任务：创建 → 轮询 → 下载 → 写清单。
- 创建：POST {BASE}/generations（v1 快速入门的请求体），响应里取 generationId
  （v1 的响应是 {"sdGenerationJob": {"generationId": ...}}，v2 参考页的响应是顶层 {"generationId": ...}，两种都兼容）
- 轮询：GET {BASE}/generations/{id}，看 generations_by_pk.status：PENDING 继续等，COMPLETE 或 FAILED 结束
  （官方 FAQ：PENDING 时 generated_images 为空）。轮询间隔逐渐加长，并设总超时
- 下载：nsfw 为 true 的图片不下载，记入清单；每张图记录 sha256
- 官方更推荐用 webhook 代替轮询，见 63C"""
import hashlib, json, os, pathlib, sys, time, urllib.error, urllib.request

BASE = os.environ.get("LEONARDO_BASE", "https://cloud.leonardo.ai/api/rest/v1")
UA = "binbho-tutorial/1.0"

class LeoError(RuntimeError):
    pass

def _req(method, url, key, body=None, timeout=60):
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(url, data=data, method=method, headers={
        "authorization": f"Bearer {key}", "accept": "application/json", "content-type": "application/json", "User-Agent": UA})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return json.load(r)
    except urllib.error.HTTPError as e:
        try:
            err = json.loads(e.read())
        except ValueError:
            err = {}
        raise LeoError(f"HTTP {e.code} [{err.get('code')}] {err.get('error')}") from None

def create(key, body, base=BASE):
    r = _req("POST", f"{base}/generations", key, body)
    gid = r.get("generationId") or (r.get("sdGenerationJob") or {}).get("generationId")
    if not gid:
        raise LeoError(f"响应里没有 generationId：{str(r)[:200]}")
    return gid

def wait(key, gid, base=BASE, timeout=300, sleep=time.sleep, clock=time.monotonic):
    start, delay = clock(), 2.0
    while True:
        g = _req("GET", f"{base}/generations/{gid}", key)["generations_by_pk"]
        if g["status"] == "COMPLETE":
            return g
        if g["status"] == "FAILED":
            raise LeoError(f"生成失败：{gid}")
        if clock() - start + delay > timeout:
            raise LeoError(f"等待超时（{timeout}s），任务仍为 {g['status']}，可稍后用 id {gid} 再查")
        sleep(delay); delay = min(delay * 1.5, 15)

def download(g, outdir, opener=urllib.request.urlopen):
    out = pathlib.Path(outdir); out.mkdir(parents=True, exist_ok=True)
    manifest = {"generation_id": g["id"], "prompt": g.get("prompt"), "images": []}
    for i, img in enumerate(g["generated_images"]):
        item = {"id": img["id"], "url": img["url"], "nsfw": img.get("nsfw", False)}
        if item["nsfw"]:
            item["skipped"] = "nsfw 标记为 true，未下载"
        else:
            ext = pathlib.Path(img["url"].split("?")[0]).suffix or ".jpg"
            p = out / f"{g['id']}_{i}{ext}"
            with opener(urllib.request.Request(img["url"], headers={"User-Agent": UA}), timeout=60) as r:
                data = r.read()
            p.write_bytes(data)
            item.update(file=p.name, sha256=hashlib.sha256(data).hexdigest(), bytes=len(data))
        manifest["images"].append(item)
    (out / f"{g['id']}.manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=1), encoding="utf-8")
    return manifest

if __name__ == "__main__":
    key = os.environ["LEONARDO_API_KEY"]
    body = json.load(open(sys.argv[1], encoding="utf-8"))
    try:
        gid = create(key, body); print("任务已创建：", gid)
        m = download(wait(key, gid), sys.argv[2] if len(sys.argv) > 2 else "out")
        print(f"下载 {sum('file' in x for x in m['images'])} 张，跳过 {sum('skipped' in x for x in m['images'])} 张")
    except LeoError as e:
        print("✗", e); sys.exit(1)

第 2 步：替身和测试

"""本地替身（我自己写的，不是 Leonardo 官方程序）。格式依据：
- 401 的错误体照搬真实接口：{"error":"Authentication hook unauthorized this request","path":"$","code":"access-denied"}
- 内容审核拦截（400）照搬官方 NSFW 指南：{"error":"content moderation filter: nude","path":"$","code":"unexpected"}
- generations_by_pk / status / generated_images[].nsfw 依据官方 FAQ 和 NSFW 指南"""
import json, threading, urllib.parse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

STATE = {"polls_until_done": 2, "polls": {}, "final": "COMPLETE", "nsfw_second": True, "v2_shape": False}
PNG = bytes.fromhex("89504e470d0a1a0a0000000d4948445200000001000000010806000000"
                    "1f15c4890000000d49444154789c6360f8cfc000000301010018dd8db40000000049454e44ae426082")

class H(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass
    def _j(self, code, obj):
        raw = json.dumps(obj).encode()
        self.send_response(code); self.send_header("Content-Type", "application/json"); self.end_headers(); self.wfile.write(raw)
    def _auth(self):
        if self.headers.get("authorization") != "Bearer good":
            self._j(401, {"error": "Authentication hook unauthorized this request", "path": "$", "code": "access-denied"}); return False
        return True
    def do_POST(self):
        if not self._auth():
            return
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        if "nude" in body.get("prompt", ""):
            return self._j(400, {"error": "content moderation filter: nude", "path": "$", "code": "unexpected"})
        gid = "11111111-2222-4333-8444-555555555555"
        STATE["polls"][gid] = 0; STATE["prompt"] = body.get("prompt")
        self._j(200, {"generationId": gid, "apiCreditCost": 1} if STATE["v2_shape"] else {"sdGenerationJob": {"generationId": gid, "apiCreditCost": 1}})
    def do_GET(self):
        path = urllib.parse.urlsplit(self.path).path
        if path.startswith("/img/"):
            self.send_response(200); self.send_header("Content-Type", "image/png"); self.end_headers(); return self.wfile.write(PNG)
        if not self._auth():
            return
        gid = path.rsplit("/", 1)[-1]
        STATE["polls"][gid] += 1
        done = STATE["polls"][gid] > STATE["polls_until_done"]
        status = STATE["final"] if done else "PENDING"
        host = f"http://{self.headers['Host']}"
        imgs = [{"id": f"img{i}", "url": f"{host}/img/Leonardo_{i}.png", "nsfw": STATE["nsfw_second"] and i == 1, "likeCount": 0,
                 "generated_image_variation_generics": []} for i in range(2)] if status == "COMPLETE" else []
        self._j(200, {"generations_by_pk": {"id": gid, "status": status, "prompt": STATE.get("prompt"), "generated_images": imgs}})

def start():
    s = ThreadingHTTPServer(("127.0.0.1", 0), H)
    threading.Thread(target=s.serve_forever, daemon=True).start()
    return s, f"http://127.0.0.1:{s.server_address[1]}"

import json
import pytest
import fake_leo as F, leo_run as L

@pytest.fixture
def base():
    F.STATE.update(polls_until_done=2, polls={}, final="COMPLETE", nsfw_second=True, v2_shape=False)
    s, u = F.start(); yield u; s.shutdown()

BODY = json.load(open("request.json"))

def test_full_flow(base, tmp_path):
    waits = []
    gid = L.create("good", BODY, base)
    g = L.wait("good", gid, base, sleep=waits.append)
    m = L.download(g, tmp_path)
    assert waits == [2.0, 3.0] and F.STATE["polls"][gid] == 3
    assert m["images"][0]["file"].endswith("_0.png") and len(m["images"][0]["sha256"]) == 64
    assert m["images"][1]["skipped"].startswith("nsfw") and "file" not in m["images"][1]
    assert json.load(open(tmp_path / f"{gid}.manifest.json"))["prompt"] == BODY["prompt"]

def test_v2_response_shape(base):
    F.STATE["v2_shape"] = True
    assert L.create("good", BODY, base) == "11111111-2222-4333-8444-555555555555"

def test_failed(base):
    F.STATE["final"] = "FAILED"
    with pytest.raises(L.LeoError, match="生成失败"):
        L.wait("good", L.create("good", BODY, base), base, sleep=lambda s: None)

def test_timeout(base):
    F.STATE["polls_until_done"] = 99
    t = iter(range(0, 1000, 10))
    with pytest.raises(L.LeoError, match="等待超时"):
        L.wait("good", L.create("good", BODY, base), base, timeout=30, sleep=lambda s: None, clock=lambda: next(t))

def test_moderation_and_auth(base):
    with pytest.raises(L.LeoError, match=r"HTTP 400 \[unexpected\] content moderation filter"):
        L.create("good", {**BODY, "prompt": "nude"}, base)
    with pytest.raises(L.LeoError, match=r"HTTP 401 \[access-denied\]"):
        L.create("bad", BODY, base)

第 3 步：运行

$ LEONARDO_API_KEY=invalid-key python leo_run.py request.json
✗ HTTP 401 [access-denied] Authentication hook unauthorized this request
$ python -m pytest -q
5 passed

调试示例

1. v1 和 v2 返回的 generationId 位置不同。 快速入门（v1）只说“返回一个 generationId”。从官方错误示例中的路径 $.selectionSet.sdGenerationJob... 可以看出，v1 的结果包在 sdGenerationJob 里；而 63A 抓取到的 v2 规范中，generationId 在响应的顶层。如果只按其中一种写法取值，换接口版本后就会拿到 None。create() 两种都会尝试，取不到时直接报错，test_v2_response_shape 覆盖了 v2 的情况。

2. 查询接口的文档链接已经失效。 官方 webhook 指南里链接的 /reference/getgenerationbyid 返回 404，仍然可以访问的是 /v1.0/reference/getgenerationbyid。v2 规范里只有 POST /generations，我没有找到 v2 的查询接口文档，所以轮询使用 v1 的 GET。用 v2 创建的任务能否用 v1 查询，我没有验证过（需要真实 Key），这是一个未确认的假设。

常见问题

模板规则

• 轮询要有间隔递增和总超时。
• 只读取 COMPLETE 状态的结果。
• nsfw 图片不对外展示。
• 每张图片都记录 sha256 和提示词。

项目资源

生成任务是异步的，要按状态来处理，不能拿到 ID 就马上读取结果。