AIGC 全景与生成模型谱系

"""最小可跑的文生图：把一句中文描述换成一张 PNG。

依赖：只用标准库（urllib / json / base64），不需要 pip 装任何东西。
环境变量：
    IMG_API_BASE   形如 https://xxx.example.com  （不带结尾斜杠）
    IMG_API_KEY    你的密钥，绝不写进源码
    IMG_MODEL      模型名，例如 gpt-image-2.5-sunburst
运行：
    export IMG_API_BASE=... IMG_API_KEY=... IMG_MODEL=...
    python3 min_text2img.py
"""
import base64
import json
import os
import urllib.request

# ① 三个配置项全部从环境变量读，源码里不出现任何密钥
API_BASE = os.environ.get("IMG_API_BASE", "").rstrip("/")
API_KEY = os.environ.get("IMG_API_KEY", "")
MODEL = os.environ.get("IMG_MODEL", "")

# ② 提示词：说清「画什么 + 什么风格 + 什么构图」，三要素缺一张图就飘
PROMPT = "一只橘猫坐在窗台上看雪，扁平插画风格，暖色调，构图居中"


def main():
    if not (API_BASE and API_KEY and MODEL):
        raise SystemExit("缺少环境变量：IMG_API_BASE / IMG_API_KEY / IMG_MODEL")

    # ③ 组请求体。OpenAI 兼容的图片接口只认这四个字段
    body = json.dumps({
        "model": MODEL,
        "prompt": PROMPT,
        "size": "1024x1024",
        "n": 1,
    }).encode("utf-8")

    req = urllib.request.Request(
        f"{API_BASE}/v1/images/generations",
        data=body,
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
        },
        method="POST",
    )

    # ④ 出图是重活，超时给足。300 秒不算夸张
    with urllib.request.urlopen(req, timeout=300) as resp:
        payload = json.load(resp)

    # ⑤ 返回的是 base64 字符串，不是图片二进制，要自己解码落盘
    b64 = payload["data"][0]["b64_json"]
    with open("out.png", "wb") as fp:
        fp.write(base64.b64decode(b64))
    print("已保存 out.png，字节数：", os.path.getsize("out.png"))


if __name__ == "__main__":
    main()

"""批量出图流水线：把一份「题材 × 风格」的组合表跑成一批图，并留下可追查的记录。

解决四个真实问题：
    1. 出图接口是慢接口，串行跑一百张要等到天亮 —— 用线程池并发；
    2. 出图接口会偶发 502 / 限流 —— 指数退避重试，别一失败就整批崩；
    3. 跑一半断了不该从头再来 —— 已存在的文件直接跳过（断点续跑）；
    4. 事后没人说得清某张图是怎么来的 —— 每张图写一条 manifest 记录。

运行：
    export IMG_API_BASE=... IMG_API_KEY=... IMG_MODEL=...
    python3 batch_pipeline.py
"""
import base64
import hashlib
import json
import os
import random
import threading
import time
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed

API_BASE = os.environ.get("IMG_API_BASE", "").rstrip("/")
API_KEY = os.environ.get("IMG_API_KEY", "")
MODEL = os.environ.get("IMG_MODEL", "")

OUT_DIR = "out_images"
MANIFEST = os.path.join(OUT_DIR, "manifest.jsonl")

# ① 题材与风格分开列，笛卡尔积展开成任务表。
#    这样加一个风格是加一行，不是复制粘贴一整段提示词。
SUBJECTS = [
    "一只橘猫坐在窗台上看雪",
    "雨后的老城巷口，青石板泛着水光",
    "深夜便利店的暖黄灯光",
]
STYLES = [
    ("flat", "扁平插画风格，纯色块，描边干净"),
    ("ink", "中国水墨风格，留白多，淡墨渐变"),
]

MAX_WORKERS = 3      # 并发数：太高会被限流，3~5 是常见的安全区间
MAX_RETRY = 4        # 每个任务最多重试次数
TIMEOUT = 300        # 单次请求超时（秒），出图是慢活

_lock = threading.Lock()


def build_tasks():
    """展开成任务列表，每个任务自带一个稳定的 id。"""
    tasks = []
    for si, subject in enumerate(SUBJECTS):
        for style_key, style_desc in STYLES:
            prompt = f"{subject}，{style_desc}"
            # ② 用提示词的哈希做文件名前缀：同样的提示词永远落在同一个文件上，
            #    这是断点续跑和去重的基础。
            digest = hashlib.sha256(prompt.encode("utf-8")).hexdigest()[:10]
            tasks.append({
                "task_id": f"{si:02d}-{style_key}-{digest}",
                "prompt": prompt,
                "subject": subject,
                "style": style_key,
                "size": "1024x1024",
            })
    return tasks


def call_api(prompt, size):
    """调一次出图接口，返回 base64 字符串。失败直接抛异常，由上层决定要不要重试。"""
    body = json.dumps({
        "model": MODEL, "prompt": prompt, "size": size, "n": 1,
    }).encode("utf-8")
    req = urllib.request.Request(
        f"{API_BASE}/v1/images/generations",
        data=body,
        headers={"Authorization": f"Bearer {API_KEY}",
                 "Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
        payload = json.load(resp)
    return payload["data"][0]["b64_json"]


def run_one(task):
    """跑一个任务：已存在就跳过，失败就指数退避重试。"""
    path = os.path.join(OUT_DIR, task["task_id"] + ".png")
    if os.path.exists(path) and os.path.getsize(path) > 0:
        return {"task_id": task["task_id"], "status": "skipped", "path": path}

    last_err = None
    for attempt in range(1, MAX_RETRY + 1):
        try:
            b64 = call_api(task["prompt"], task["size"])
            data = base64.b64decode(b64)
            # ③ 先写临时文件再改名：中途被 Ctrl-C 也不会留下半张坏图
            tmp = path + ".part"
            with open(tmp, "wb") as fp:
                fp.write(data)
            os.replace(tmp, path)

            record = {
                "task_id": task["task_id"],
                "prompt": task["prompt"],
                "subject": task["subject"],
                "style": task["style"],
                "size": task["size"],
                "model": MODEL,
                "bytes": len(data),
                "sha256": hashlib.sha256(data).hexdigest(),
                "created_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
                "attempt": attempt,
            }
            # ④ manifest 用 jsonl 追加写，多线程下加锁，别写串行
            with _lock:
                with open(MANIFEST, "a", encoding="utf-8") as fp:
                    fp.write(json.dumps(record, ensure_ascii=False) + "\n")
            return {"task_id": task["task_id"], "status": "ok", "path": path}

        except urllib.error.HTTPError as exc:
            last_err = f"HTTP {exc.code}"
            # 4xx 里只有 429（限流）值得重试，其余是请求本身错了，重试多少次都一样
            if exc.code < 500 and exc.code != 429:
                break
        except Exception as exc:                      # noqa: BLE001
            last_err = repr(exc)

        if attempt < MAX_RETRY:
            # ⑤ 指数退避 + 随机抖动：避免多个 worker 同时醒来再次撞在一起
            backoff = min(60, 2 ** attempt) + random.uniform(0, 3)
            time.sleep(backoff)

    return {"task_id": task["task_id"], "status": "failed", "error": last_err}


def main():
    if not (API_BASE and API_KEY and MODEL):
        raise SystemExit("缺少环境变量：IMG_API_BASE / IMG_API_KEY / IMG_MODEL")
    os.makedirs(OUT_DIR, exist_ok=True)

    tasks = build_tasks()
    print(f"任务总数 {len(tasks)}，并发 {MAX_WORKERS}")

    done = []
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
        futures = {pool.submit(run_one, t): t for t in tasks}
        for fut in as_completed(futures):
            result = fut.result()
            done.append(result)
            print(f"[{len(done)}/{len(tasks)}] {result['status']:<7} {result['task_id']}")

    ok = sum(1 for r in done if r["status"] == "ok")
    skipped = sum(1 for r in done if r["status"] == "skipped")
    failed = [r for r in done if r["status"] == "failed"]
    print(f"\n完成 {ok} 张，跳过 {skipped} 张，失败 {len(failed)} 张")
    for r in failed:
        print("  失败：", r["task_id"], r.get("error"))


if __name__ == "__main__":
    main()

任务总数 6，并发 3
[1/6] ok      00-flat-3f9c2a1b7e
[2/6] ok      00-ink-8b1d4e6a02
[3/6] skipped 01-flat-c47a90e155
[4/6] ok      01-ink-2e8f71b3d9
[5/6] failed  02-flat-7a3c5d0f81
[6/6] ok      02-ink-6d92b4e738

完成 4 张，跳过 1 张，失败 1 张
  失败： 02-flat-7a3c5d0f81 HTTP 502

"""多模态统一抽象层：一个 generate() 管住文本、图像、音频三种产出。

为什么要抽这一层：
    每接一家服务就在业务代码里写一遍 if provider == "xxx"，接第三家的时候
    代码就烂透了。把「模态 → 能力 → 后端」三件事分开，业务只认能力名。

结构：
    Capability   一种能力（text / image / audio），声明入参与出参形状
    Backend      一个具体后端的实现，注册到某个能力上
    Router       按能力名派发，带兜底与降级

运行（不配任何密钥也能跑，会走 DryRun 后端打印请求）：
    python3 modal_router.py
"""
import json
import os
import time
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional


# ---------------------------------------------------------------- 数据结构

@dataclass
class GenRequest:
    """一次生成请求。三种模态共用同一个请求对象，差异放在 options 里。"""
    capability: str                      # "text" / "image" / "audio"
    prompt: str
    options: Dict[str, Any] = field(default_factory=dict)


@dataclass
class GenResult:
    """一次生成的结果。content 的含义由 kind 决定，调用方据此处理。"""
    kind: str                            # "text" / "image_b64" / "audio_b64"
    content: str
    backend: str
    elapsed_ms: int
    meta: Dict[str, Any] = field(default_factory=dict)


class BackendError(RuntimeError):
    """后端失败。抛这个异常表示「这个后端不行，可以换下一个」。"""


# ---------------------------------------------------------------- 后端实现

class Backend:
    """所有后端的基类。加一家新服务 = 写一个子类 + 注册，业务代码一行不改。"""

    name = "base"
    capabilities: List[str] = []

    def available(self) -> bool:
        """环境变量齐不齐。不齐就别参与派发，省得每次都失败一遍。"""
        return True

    def run(self, req: GenRequest) -> GenResult:       # pragma: no cover
        raise NotImplementedError


class DryRunBackend(Backend):
    """不连网的占位后端：把请求原样打印出来。

    它的价值不是演示，是让整条链路在没有密钥的机器上也能跑通，
    这样写业务逻辑的人不必等密钥审批。
    """

    name = "dryrun"
    capabilities = ["text", "image", "audio"]

    def run(self, req: GenRequest) -> GenResult:
        start = time.time()
        payload = {"capability": req.capability,
                   "prompt": req.prompt,
                   "options": req.options}
        body = json.dumps(payload, ensure_ascii=False)
        kind = {"text": "text", "image": "image_b64", "audio": "audio_b64"}[req.capability]
        return GenResult(
            kind=kind,
            content=f"[dryrun] {body}",
            backend=self.name,
            elapsed_ms=int((time.time() - start) * 1000),
            meta={"note": "未真实调用任何服务"},
        )


class OpenAICompatImageBackend(Backend):
    """任何 OpenAI 兼容的图片接口。换厂商只改 base_url，业务代码不动。"""

    name = "openai-compat-image"
    capabilities = ["image"]

    def __init__(self):
        self.base = os.environ.get("IMG_API_BASE", "").rstrip("/")
        self.key = os.environ.get("IMG_API_KEY", "")
        self.model = os.environ.get("IMG_MODEL", "")

    def available(self) -> bool:
        return bool(self.base and self.key and self.model)

    def run(self, req: GenRequest) -> GenResult:
        import urllib.request                       # 局部导入：不走这条路就不加载

        start = time.time()
        body = json.dumps({
            "model": self.model,
            "prompt": req.prompt,
            "size": req.options.get("size", "1024x1024"),
            "n": 1,
        }).encode("utf-8")
        request = urllib.request.Request(
            f"{self.base}/v1/images/generations",
            data=body,
            headers={"Authorization": f"Bearer {self.key}",
                     "Content-Type": "application/json"},
            method="POST",
        )
        try:
            with urllib.request.urlopen(request, timeout=300) as resp:
                payload = json.load(resp)
            b64 = payload["data"][0]["b64_json"]
        except Exception as exc:                    # noqa: BLE001
            raise BackendError(f"{self.name} 失败：{exc!r}") from exc

        return GenResult(
            kind="image_b64",
            content=b64,
            backend=self.name,
            elapsed_ms=int((time.time() - start) * 1000),
            meta={"model": self.model},
        )


# ---------------------------------------------------------------- 路由器

class Router:
    """按能力派发，失败自动换下一个后端。"""

    def __init__(self):
        self._by_capability: Dict[str, List[Backend]] = {}
        self._hooks: List[Callable[[GenRequest, Optional[GenResult], Optional[Exception]], None]] = []

    def register(self, backend: Backend) -> "Router":
        for cap in backend.capabilities:
            self._by_capability.setdefault(cap, []).append(backend)
        return self

    def on_finish(self, hook) -> "Router":
        """注册一个回调，用来记账、埋点、写溯源清单。"""
        self._hooks.append(hook)
        return self

    def generate(self, req: GenRequest) -> GenResult:
        chain = [b for b in self._by_capability.get(req.capability, []) if b.available()]
        if not chain:
            raise BackendError(f"没有可用后端支持能力：{req.capability}")

        last_exc: Optional[Exception] = None
        for backend in chain:
            try:
                result = backend.run(req)
            except BackendError as exc:
                last_exc = exc
                for hook in self._hooks:
                    hook(req, None, exc)
                continue                              # 换下一家，不让整条链路死在一家身上
            for hook in self._hooks:
                hook(req, result, None)
            return result

        raise BackendError(f"能力 {req.capability} 的后端全部失败：{last_exc!r}")


def main():
    router = (Router()
              .register(OpenAICompatImageBackend())   # 优先走真实服务
              .register(DryRunBackend()))             # 兜底：密钥没配也能跑通链路

    def audit(req, result, exc):
        status = "ok" if result else "fail"
        who = result.backend if result else "-"
        print(f"  [{status}] cap={req.capability} backend={who} err={exc}")

    router.on_finish(audit)

    for req in [
        GenRequest("image", "一只橘猫坐在窗台上看雪，扁平插画风格", {"size": "1024x1024"}),
        GenRequest("text", "给这张图写一句二十字以内的中文标题"),
        GenRequest("audio", "把上面那句标题读出来，语速偏慢"),
    ]:
        print(f"请求：{req.capability}")
        result = router.generate(req)
        preview = result.content[:60].replace("\n", " ")
        print(f"  → kind={result.kind} backend={result.backend} "
              f"{result.elapsed_ms}ms content={preview}...\n")


if __name__ == "__main__":
    main()

"""生成溯源清单：让每一张出图都能回答「谁、用什么、怎么生成的」。

为什么必须做：
    出图一多，三个月后没人说得清某张图的提示词、模型版本、随机种子。
    想复现复现不了，想排查侵权风险也查不到源头。
    这份清单写的是「生成参数」，不是「内容真伪」—— 它证明不了图里的东西是真的。

功能：
    record()   追加一条生成记录（含文件指纹）
    verify()   按指纹校验文件有没有被换掉或损坏
    report()   汇总成 Markdown 表，交付时随包给出

运行：
    python3 provenance.py demo      # 造两条演示记录并校验
"""
import hashlib
import json
import os
import sys
import time
from typing import Dict, Iterator, List

LEDGER = "provenance.jsonl"


def file_sha256(path: str, chunk: int = 1 << 20) -> str:
    """分块算哈希：图片动辄几 MB，别一次性读进内存。"""
    digest = hashlib.sha256()
    with open(path, "rb") as fp:
        while True:
            block = fp.read(chunk)
            if not block:
                break
            digest.update(block)
    return digest.hexdigest()


def record(path: str, *, prompt: str, model: str, seed=None,
           extra: Dict = None, ledger: str = LEDGER) -> Dict:
    """给一个已生成的文件登记一条记录。"""
    if not os.path.exists(path):
        raise FileNotFoundError(path)

    entry = {
        "path": os.path.relpath(path),
        "bytes": os.path.getsize(path),
        "sha256": file_sha256(path),
        "prompt": prompt,
        "model": model,
        "seed": seed,
        "created_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "generator": "ai",          # 明确标注这是机器生成的产物
    }
    if extra:
        entry.update(extra)

    with open(ledger, "a", encoding="utf-8") as fp:
        fp.write(json.dumps(entry, ensure_ascii=False) + "\n")
    return entry


def load(ledger: str = LEDGER) -> Iterator[Dict]:
    if not os.path.exists(ledger):
        return iter(())
    def _gen():
        with open(ledger, encoding="utf-8") as fp:
            for line in fp:
                line = line.strip()
                if line:
                    yield json.loads(line)
    return _gen()


def verify(ledger: str = LEDGER) -> List[Dict]:
    """逐条校验文件是否仍与登记时一致。返回有问题的条目。"""
    problems = []
    for entry in load(ledger):
        path = entry["path"]
        if not os.path.exists(path):
            problems.append({**entry, "issue": "文件已丢失"})
            continue
        if file_sha256(path) != entry["sha256"]:
            problems.append({**entry, "issue": "内容与登记时不一致"})
    return problems


def report(ledger: str = LEDGER) -> str:
    """汇总成 Markdown 表，直接贴进交付说明。"""
    rows = list(load(ledger))
    if not rows:
        return "（暂无记录）"
    lines = ["| 文件 | 模型 | seed | 生成时间 | 指纹前 12 位 |",
             "| --- | --- | --- | --- | --- |"]
    for r in rows:
        lines.append("| {path} | {model} | {seed} | {created_at} | `{short}` |".format(
            path=r["path"], model=r["model"],
            seed=r.get("seed") if r.get("seed") is not None else "-",
            created_at=r["created_at"], short=r["sha256"][:12]))
    return "\n".join(lines)


def _demo():
    os.makedirs("demo_assets", exist_ok=True)
    for name, text in [("a.txt", "第一张图的占位内容"), ("b.txt", "第二张图的占位内容")]:
        with open(os.path.join("demo_assets", name), "w", encoding="utf-8") as fp:
            fp.write(text)

    record("demo_assets/a.txt", prompt="一只橘猫坐在窗台上看雪，扁平插画",
           model="demo-image-model", seed=20260127)
    record("demo_assets/b.txt", prompt="雨后老城巷口，水墨风格",
           model="demo-image-model", seed=20260129)

    print(report())
    print("\n校验：", verify() or "全部一致 ✅")

    # 故意改动一个文件，验证校验确实能抓出来
    with open("demo_assets/b.txt", "a", encoding="utf-8") as fp:
        fp.write("（被人改过）")
    print("改动后校验：", verify())


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "demo":
        _demo()
    else:
        print(report())

| 文件 | 模型 | seed | 生成时间 | 指纹前 12 位 |
| --- | --- | --- | --- | --- |
| demo_assets/a.txt | demo-image-model | 20260127 | 2026-02-03T10:12:41 | `9f2c7a1b4d08` |
| demo_assets/b.txt | demo-image-model | 20260129 | 2026-02-03T10:12:41 | `4e81d35c6a97` |

校验： 全部一致 ✅
改动后校验： [{'path': 'demo_assets/b.txt', ..., 'issue': '内容与登记时不一致'}]

"""AIGC 统一客户端骨架 —— 复制后只改 TODO 处。

这份骨架把「接哪家服务」和「业务要什么」彻底分开：
    业务代码只说 client.image("一只橘猫...")，
    换厂商、加降级、加重试、加审计都在这个文件里改，业务代码一行不动。

依赖：标准库即可。
"""
import base64
import json
import os
import random
import time
import urllib.error
import urllib.request
from typing import Any, Callable, Dict, List, Optional

# ---------------------------------------------------------------------------
# TODO 1：把你要接的服务填在这里。
#   key      给业务代码用的短名字
#   base     服务根地址，不带结尾斜杠
#   env_key  存放密钥的环境变量名（密钥本身绝不写进源码）
#   model    默认模型名
#   caps     这家服务支持哪些能力
# ---------------------------------------------------------------------------
PROVIDERS: List[Dict[str, Any]] = [
    {
        "key": "primary",
        "base": os.environ.get("IMG_API_BASE", "").rstrip("/"),
        "env_key": "IMG_API_KEY",
        "model": os.environ.get("IMG_MODEL", ""),
        "caps": ["image"],
    },
    # TODO 2：要加第二家做降级就再补一段，顺序即优先级
    # {
    #     "key": "backup",
    #     "base": os.environ.get("BACKUP_API_BASE", "").rstrip("/"),
    #     "env_key": "BACKUP_API_KEY",
    #     "model": os.environ.get("BACKUP_MODEL", ""),
    #     "caps": ["image", "text"],
    # },
]

MAX_RETRY = 4        # TODO 3：重试次数与超时按你的服务实际情况调
TIMEOUT = 300


class AigcError(RuntimeError):
    pass


class AigcClient:
    def __init__(self, providers=None, on_event: Optional[Callable] = None):
        self.providers = providers if providers is not None else PROVIDERS
        # TODO 4：把埋点/审计/写溯源清单挂在这个回调上
        self.on_event = on_event or (lambda **kw: None)

    # ------------------------------------------------------------- 对外能力

    def image(self, prompt: str, size: str = "1024x1024", **opts) -> bytes:
        """文生图，返回图片二进制。"""
        payload = self._dispatch("image", prompt, {"size": size, **opts})
        return base64.b64decode(payload["data"][0]["b64_json"])

    def text(self, prompt: str, **opts) -> str:
        """文本生成。TODO 5：不同厂商的文本接口字段不一致，在这里对齐。"""
        payload = self._dispatch("text", prompt, opts)
        return payload["choices"][0]["message"]["content"]

    # ------------------------------------------------------------- 内部实现

    def _endpoint(self, capability: str) -> str:
        return {
            "image": "/v1/images/generations",
            "text": "/v1/chat/completions",
        }[capability]

    def _body(self, capability: str, provider: Dict, prompt: str, opts: Dict) -> bytes:
        if capability == "image":
            body = {"model": provider["model"], "prompt": prompt,
                    "size": opts.get("size", "1024x1024"), "n": 1}
        else:
            body = {"model": provider["model"],
                    "messages": [{"role": "user", "content": prompt}]}
            body.update({k: v for k, v in opts.items() if k != "size"})
        return json.dumps(body).encode("utf-8")

    def _dispatch(self, capability: str, prompt: str, opts: Dict) -> Dict:
        usable = [p for p in self.providers
                  if capability in p["caps"] and p["base"] and os.environ.get(p["env_key"])]
        if not usable:
            raise AigcError(f"没有可用服务支持能力 {capability}；检查环境变量是否配齐")

        last_err = None
        for provider in usable:
            try:
                return self._call_with_retry(capability, provider, prompt, opts)
            except AigcError as exc:
                last_err = exc
                self.on_event(event="provider_failed", provider=provider["key"], error=str(exc))
        raise AigcError(f"全部服务失败：{last_err}")

    def _call_with_retry(self, capability, provider, prompt, opts) -> Dict:
        url = provider["base"] + self._endpoint(capability)
        key = os.environ[provider["env_key"]]
        body = self._body(capability, provider, prompt, opts)

        for attempt in range(1, MAX_RETRY + 1):
            req = urllib.request.Request(
                url, data=body,
                headers={"Authorization": f"Bearer {key}",
                         "Content-Type": "application/json"},
                method="POST")
            try:
                started = time.time()
                with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
                    payload = json.load(resp)
                self.on_event(event="ok", provider=provider["key"],
                              capability=capability, attempt=attempt,
                              elapsed_ms=int((time.time() - started) * 1000))
                return payload
            except urllib.error.HTTPError as exc:
                # 只有 5xx 和 429 值得重试；其余是请求本身写错了
                if exc.code < 500 and exc.code != 429:
                    raise AigcError(f"HTTP {exc.code}，请求有问题，不重试") from exc
                err = f"HTTP {exc.code}"
            except Exception as exc:                      # noqa: BLE001
                err = repr(exc)

            if attempt == MAX_RETRY:
                raise AigcError(f"{provider['key']} 重试 {MAX_RETRY} 次仍失败：{err}")
            time.sleep(min(60, 2 ** attempt) + random.uniform(0, 3))

        raise AigcError("不可达")


if __name__ == "__main__":
    # TODO 6：换成你自己的调用
    client = AigcClient(on_event=lambda **kw: print("  事件：", kw))
    data = client.image("一只橘猫坐在窗台上看雪，扁平插画风格")
    with open("out.png", "wb") as fp:
        fp.write(data)
    print("已保存 out.png", len(data), "字节")

{
  "_说明": "批量出图的任务配置。题材与风格分开列，程序做笛卡尔积展开；加一个风格是加一行，不是复制一整段提示词。",
  "output_dir": "out_images",
  "manifest": "out_images/manifest.jsonl",
  "concurrency": 3,
  "max_retry": 4,
  "timeout_seconds": 300,
  "default_size": "1024x1024",
  "subjects": [
    {"key": "cat-snow", "text": "TODO：写清楚画什么，主体 + 场景 + 动作"},
    {"key": "old-alley", "text": "TODO：第二个题材"}
  ],
  "styles": [
    {"key": "flat", "text": "TODO：风格描述，例如 扁平插画风格，纯色块，描边干净"},
    {"key": "ink", "text": "TODO：第二种风格"}
  ],
  "negative_prompt": "TODO：不想要的东西，例如 文字、水印、多余的手指",
  "seed": null,
  "_seed说明": "填一个整数可复现同一张图；填 null 表示每次随机。要复现就必须记进 manifest。"
}