FLUX.2 [klein] 4B 图片生成

export WILDFLOW_API_KEY="替换为控制台生成的密钥"
export WILDFLOW_IDEMPOTENCY_KEY="$(python3 -c 'import uuid; print(uuid.uuid4())')"

curl --fail-with-body --silent --show-error -D response-headers.txt -X POST https://api.wildflow.cn/v1/jobs \
  -H "Authorization: Bearer $WILDFLOW_API_KEY" \
  -H "Idempotency-Key: $WILDFLOW_IDEMPOTENCY_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "FLUX.2 [klein] 4B",
    "parameters": {
      "prompt": "一只熊猫在竹林中阅读，柔和自然光，电影感构图",
      "width": 1024,
      "height": 1024,
      "num_inference_steps": 4,
      "guidance_scale": 1.0,
      "seed": 42
    }
  }' \
  --output operation.json

export RESULT_FILE="result.png"
export RESULT_MEDIA_TYPE="image/png"

python3 - <<'PY'
import hashlib
import json
import os
import time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from pathlib import Path
from urllib.error import HTTPError
from urllib.parse import quote
from urllib.request import Request, urlopen

base = "https://api.wildflow.cn"
headers = {"Authorization": "Bearer " + os.environ["WILDFLOW_API_KEY"]}
operation = json.loads(Path("operation.json").read_text())
operation_id = operation["id"]
status_url = base + "/v1/jobs/" + quote(operation_id, safe="")
output = Path(os.environ["RESULT_FILE"])
expected_media_type = os.environ["RESULT_MEDIA_TYPE"]
deadline = time.monotonic() + 7200

def pause(value):
    try:
        seconds = float(value)
    except (TypeError, ValueError):
        try:
            seconds = (parsedate_to_datetime(value) - datetime.now(timezone.utc)).total_seconds()
        except (TypeError, ValueError, OverflowError):
            seconds = 3
    seconds = max(1, seconds)
    if time.monotonic() + seconds >= deadline:
        raise SystemExit("等待超时。保留 operation.json，稍后重新执行本段继续查询。")
    time.sleep(seconds)

while time.monotonic() < deadline:
    try:
        with urlopen(Request(status_url, headers=headers), timeout=60) as response:
            status = json.load(response)
            retry_after = response.headers.get("Retry-After", "3")
    except HTTPError as error:
        if error.code in (429, 503):
            pause(error.headers.get("Retry-After", "3"))
            continue
        raise SystemExit(f"查询失败：HTTP {error.code}。保留原任务，检查错误后继续查询。")

    state = status["state"]
    print("任务状态：", state, flush=True)
    if state == "succeeded":
        artifacts = status.get("artifacts", [])
        if not artifacts:
            raise SystemExit("任务成功但缺少结果文件，请保留任务 ID 并联系支持。")
        artifact = artifacts[0]
        if artifact["media_type"] != expected_media_type:
            raise SystemExit("结果格式与预期不同，请检查任务和模型。")
        download_url = base + "/v1/artifacts/" + quote(artifact["id"], safe="") + "/content"
        temporary = output.with_name(output.name + ".part")
        digest = hashlib.sha256()
        size = 0
        with urlopen(Request(download_url, headers=headers), timeout=60) as response:
            with temporary.open("wb") as file:
                while chunk := response.read(1024 * 1024):
                    file.write(chunk)
                    digest.update(chunk)
                    size += len(chunk)
        if size != artifact["size_bytes"] or digest.hexdigest() != artifact["sha256"]:
            raise SystemExit("文件校验失败，保留原任务并重新执行本段下载。")
        temporary.replace(output)
        print("已保存：", output)
        break
    if state not in ("queued", "running", "submitting"):
        raise SystemExit(f"任务已停止：{state}。请保留任务 ID；不要换幂等键重复提交。")
    pause(retry_after)
else:
    raise SystemExit("等待超时。保留 operation.json，稍后继续查询原任务。")
PY