图片生成

curl -X POST https://api.anycap.io/v1/generations \  -H "Authorization: Bearer $ANYCAP_API_KEY" \  -H "Content-Type: application/json" \  -d '{    "type": "image",    "model": "<来自 GET /v1/generations/models?type=image 的模型>",    "input": { "prompt": "白底商品图" },    "parameters": { "size": "1:1", "n": 1 }  }'

{  "type": "image",  "model": "gpt-image-2.6-flare",  "input": {    "prompt": "A paper lantern gently swaying, seamless loop"  },  "parameters": {    "output_format": "gif",    "size": "512x512",    "frame_count": 9,    "frame_duration_ms": 200,    "quality": "low",    "n": 1  }}

{  "state": "failed",  "error_code": "upstream_call_failed",  "error_message": "Your request did not pass content safety review.",  "images": [],  "billing_status": "released",  "charged_microcredits": "0"}

{  "native_request": {    "model": "<来源准确模型名>",    "prompt": "A red apple on a white background",    "size": "1:1",    "resolution": "1k",    "quality": "medium",    "n": 1  }}

# Python 3 standard library; save as anycap_image.py and run on your server.import json, os, time, uuidfrom pathlib import Pathfrom urllib.request import Request, urlopenfrom urllib.error import HTTPError, URLError base = os.environ["ANYCAP_API_BASE"].rstrip("/")key = os.environ["ANYCAP_API_KEY"]state_file = Path(os.environ.get("ANYCAP_IMAGE_STATE_FILE", "anycap-image-job.json")) def call(method, path, body=None):    data = None if body is None else json.dumps(body).encode()    req = Request(base + path, data=data, method=method, headers={        "Authorization": "Bearer " + key, "Content-Type": "application/json"    })    with urlopen(req, timeout=30) as response:        return json.load(response) def poll(job_id):    for _ in range(120):        try:            job = call("GET", "/api/v1/image-executions/" + job_id)        except HTTPError as error:            if error.code == 404:                raise RuntimeError("Task not visible; retain state file and contact support. Do not resubmit.") from None            if error.code not in (429, 500, 502, 503, 504):                raise RuntimeError("Task query rejected: HTTP " + str(error.code)) from None            time.sleep(15)            continue        except (URLError, TimeoutError):            time.sleep(15)            continue        if job["state"] in ("completed", "failed", "pending"):            if job["state"] == "completed":                job = call("GET", "/api/v1/image-executions/" + job_id + "?result=true")            print(json.dumps(job))            return        time.sleep(15)    print(json.dumps({"id": job_id, "message": "Polling deadline; rerun to resume GET only."})) # An existing state file always resumes reads, including after a lost POST response.if state_file.exists():    saved = json.loads(state_file.read_text())    if saved["api_base"] != base:        raise RuntimeError("State belongs to another API origin")    poll(saved["id"])else:    models = call("GET", "/v1/images/models")["items"]    selected = os.environ.get("ANYCAP_IMAGE_MODEL")    if not selected:        print(json.dumps({"models": models}))        raise SystemExit("Set ANYCAP_IMAGE_MODEL and an exact priced combination; see the catalog above.")    model = next((m for m in models if m["model"] == selected), None)    if model is None:        raise RuntimeError("Selected model is unavailable or not authorized")    params = {k: os.environ["ANYCAP_IMAGE_" + k.upper()] for k in ("size", "resolution", "quality")}    axis = "quality"    if not any(c["size"] == params["size"] and c["resolution"] == params["resolution"] and c.get(axis, "") == params["quality"] for c in model["priced_combinations"]):        raise RuntimeError("Select an exact priced combination from the model catalog")    n = int(os.environ.get("ANYCAP_IMAGE_N", "1"))    native = {"model": selected, "prompt": os.environ["ANYCAP_IMAGE_PROMPT"], "size": params["size"], "resolution": params["resolution"], "n": n}    native[axis] = params["quality"]    body = {"native_request": native}    quote = call("POST", "/api/v1/capabilities/image.generate/quotes", body)    print(json.dumps({"quote": quote}))    limit = int(os.environ["ANYCAP_MAX_MICROCREDITS"])    if limit <= 0 or int(quote["amount_microcredits"]) > limit:        raise RuntimeError("Quote exceeds authorized budget")    if os.environ.get("ANYCAP_ALLOW_PAID_GENERATION") != "yes":        raise SystemExit("No generation submitted. Set ANYCAP_ALLOW_PAID_GENERATION=yes only after authorization.")    job_id = str(uuid.uuid4())    submission = dict(body, quote_id=quote["id"], idempotency_key=job_id)    # Exclusive private file creation; flush to disk BEFORE the billable POST.    fd = os.open(state_file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)    with os.fdopen(fd, "w") as saved:        json.dump({"api_base": base, "id": job_id, "submission": submission}, saved)        saved.flush()        os.fsync(saved.fileno())    print(json.dumps({"id": job_id, "state_file": str(state_file)}))    try:        call("POST", "/api/v1/capabilities/image.generate/executions", submission)    except (HTTPError, URLError, TimeoutError):        raise SystemExit("Submission not confirmed. Retain state file; rerun for GET only. Do not resubmit.") from None    poll(job_id)