火山即梦 AI 4.0：轮询图片生成结果

查询结果：Action=CVSync2AsyncGetResult，body 带 req_key 与 task_id；状态常见 in_queue / generating / done。

先看清成品

• mock 轮询至 done 并给出 image_urls。
• 1 项 pytest 通过。

合规提醒：即梦网页与火山 API 均须遵守平台规则与国内 AI 生成合成内容标识要求。不上传未授权肖像/作品，不做未经同意的换脸仿声，不生成违法或伪造新闻画面。商用以会员/合同为准。教学客户端用 X-Access-Key 对接 mock；生产请用火山 SDK 做 Signature V4。本文不编造价格。

运行说明：无正式 VOLC AK/SK。A/B 用 mock 提交与轮询；未对 visual.volcengineapi.com 做真实签名调用。C 为发布前标识自检。2026-10-23（Asia/Shanghai）pytest 通过。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

113-jimeng-ai-B-src/
├── jimeng_poll.py
├── mock_jimeng.py
└── test_113b.py

第 1 步：客户端

#!/usr/bin/env python3
"""即梦异步结果查询：Action=CVSync2AsyncGetResult，body 含 req_key + task_id。"""
from __future__ import annotations
import argparse, json, os, sys, time, urllib.error, urllib.parse, urllib.request


def poll_once(base, access_key, task_id, req_key="jimeng_t2i_v40"):
    qs = urllib.parse.urlencode({"Action": "CVSync2AsyncGetResult", "Version": "2022-08-31"})
    body = {"req_key": req_key, "task_id": task_id}
    req = urllib.request.Request(
        f"{base.rstrip('/')}/?{qs}",
        data=json.dumps(body).encode(),
        headers={"Content-Type": "application/json", "X-Access-Key": access_key},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            return json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        raise SystemExit(f"HTTP {e.code}: {e.read().decode(errors='replace')}") from e


def wait(base, ak, task_id, timeout=20, interval=0.2):
    deadline = time.time() + timeout
    while time.time() < deadline:
        r = poll_once(base, ak, task_id)
        st = (r.get("data") or {}).get("status")
        if st in ("done", "success"):
            return r
        if st in ("failed", "error"):
            raise SystemExit(json.dumps(r, ensure_ascii=False))
        time.sleep(interval)
    raise SystemExit("轮询超时")


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("task_id")
    ap.add_argument("--base", default=os.environ.get("JIMENG_BASE_URL", "https://visual.volcengineapi.com"))
    ap.add_argument("--ak", default=os.environ.get("VOLC_ACCESS_KEY", ""))
    ap.add_argument("--wait", action="store_true")
    args = ap.parse_args(argv)
    if not args.ak:
        print("设置 VOLC_ACCESS_KEY", file=sys.stderr); return 2
    r = wait(args.base, args.ak, args.task_id) if args.wait else poll_once(args.base, args.ak, args.task_id)
    print(json.dumps(r, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

第 2 步：Mock

#!/usr/bin/env python3
"""Mock Jimeng CVSync2AsyncSubmitTask / GetResult (simplified AK header)."""
from __future__ import annotations
import json, uuid
from urllib.parse import parse_qs, urlparse
from http.server import BaseHTTPRequestHandler, HTTPServer

AK = "volc-test-ak"
TASKS = {}

class H(BaseHTTPRequestHandler):
    def log_message(self, *a): pass
    def _send(self, code, obj):
        b = json.dumps(obj).encode()
        self.send_response(code); self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(b))); self.end_headers(); self.wfile.write(b)
    def do_POST(self):
        if self.headers.get("X-Access-Key") != AK:
            self._send(401, {"code": 401, "message": "Unauthorized"}); return
        qs = parse_qs(urlparse(self.path).query)
        action = (qs.get("Action") or [""])[0]
        n = int(self.headers.get("Content-Length", 0))
        body = json.loads(self.rfile.read(n) or b"{}")
        if action == "CVSync2AsyncSubmitTask":
            if body.get("req_key") != "jimeng_t2i_v40" or not body.get("prompt"):
                self._send(400, {"code": 400, "message": "bad body"}); return
            tid = str(uuid.uuid4())
            TASKS[tid] = {"prompt": body["prompt"], "status": "in_queue"}
            self._send(200, {"code": 10000, "data": {"task_id": tid}}); return
        if action == "CVSync2AsyncGetResult":
            tid = body.get("task_id")
            t = TASKS.get(tid)
            if not t: self._send(404, {"code": 404, "message": "unknown task"}); return
            self._send(200, {"code": 10000, "data": {
                "status": "done",
                "image_urls": [f"https://cdn.example/jimeng/{tid}.png"],
            }}); return
        self.send_error(404)

def main():
    import argparse
    ap = argparse.ArgumentParser(); ap.add_argument("--port", type=int, default=9114)
    args = ap.parse_args()
    HTTPServer(("127.0.0.1", args.port), H).serve_forever()

if __name__ == "__main__":
    main()

第 3 步：测试

import json, subprocess, sys, time, urllib.request
from pathlib import Path
HERE = Path(__file__).resolve().parent

def test_poll():
    port = 19115
    proc = subprocess.Popen([sys.executable, str(HERE/"mock_jimeng.py"), "--port", str(port)],
                            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        time.sleep(0.3)
        # submit via urllib to seed task
        import urllib.parse
        qs = urllib.parse.urlencode({"Action": "CVSync2AsyncSubmitTask", "Version": "2022-08-31"})
        body = json.dumps({"req_key": "jimeng_t2i_v40", "prompt": "test", "force_single": True}).encode()
        req = urllib.request.Request(f"http://127.0.0.1:{port}/?{qs}", data=body,
                                     headers={"Content-Type": "application/json", "X-Access-Key": "volc-test-ak"}, method="POST")
        with urllib.request.urlopen(req) as resp:
            tid = json.loads(resp.read().decode())["data"]["task_id"]
        r = subprocess.run(
            [sys.executable, str(HERE/"jimeng_poll.py"), tid, "--base", f"http://127.0.0.1:{port}",
             "--ak", "volc-test-ak", "--wait"],
            capture_output=True, text=True, check=True)
        data = json.loads(r.stdout)
        assert data["data"]["status"] == "done"
        assert data["data"]["image_urls"]
    finally:
        proc.terminate(); proc.wait(timeout=5)

第 4 步：运行

$ pytest 113b
.                                                                        [100%]
1 passed in 0.40s

怎么检查结果？

• status=done 且 image_urls 非空。
• 正式环境加大间隔，避免打爆限流。

调试示例

一直 in_queue。 查控制台配额与服务是否开通；高峰可拉长 sleep，不要并行狂刷同一 task。

常见问题

模板规则

• 保存 task_id。
• 成功再下载。
• 发布走 C 篇清单。

项目资源

图出来了，标识和授权还没完。