用 PixVerse API 提交视频生成任务：轮询状态、下载结果与错误处理

PixVerse 提供官方 API，可以在自己的服务端批量生成视频。 这篇做 pixverse_job.py：提交文生视频任务、按官方建议轮询状态、下载结果，并把每次请求的 trace id 和错误码记进日志。重点是官方文档强调的一点：HTTP 200 不等于成功。

先看清成品

• 用无效 key 真实调用：返回 HTTP 200 + ErrCode 10005。只检查 HTTP 状态码的客户端会把它当成成功。
• 本地替身上完整跑通：限流 712008 → 退避后提交成功 → 轮询两次 → 下载；每个请求的 trace id 都不同。
• 发现并修复两个问题：错误信息丢掉了处理建议；sleep=time.sleep 默认参数导致测试无法替换等待。pytest 6 项通过。

合规提醒：只上传你有权使用的图片和视频；不生成真实人物的误导性内容，不做未经同意的换脸；发布时按平台规则标注 AI 生成；商用前确认所用套餐的授权范围（以 PixVerse 官方条款为准）。本文只使用 PixVerse 官方网页/App 和官方 API。API key 只放在服务端，不放进前端代码或日志。本文不列价格，积分消耗以控制台为准。

运行说明：box 上没有 PixVerse 账号和 API key，没有生成过任何视频，也没有花费积分。我用无效 key 真实调用了文生视频接口和查询接口，两次都得到 HTTP 200 和 {"ErrCode":10005,"ErrMsg":"apiKey is not registered"}（2026-10-21 20:06，UTC+8）。其余接口响应来自我写的本地替身（错误码和状态值按官方文档）。C 篇的视频片段是我用 ffmpeg 合成的测试画面，不是 PixVerse 的输出。所有脚本和测试都在 box 上真实运行过（Python 3.13、ffmpeg 7.1.5）。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

72-pixverse-A-src/
├── pixverse_job.py
├── fake_pixverse.py
├── demo_local.py
└── test_job.py

第 1 步：创建 API key（手动）

登录 PixVerse API 平台，在账户设置中创建 key，放进服务端环境变量 PIXVERSE_API_KEY。（本文没有账号，没有实际操作这一步。）

第 2 步：主脚本

"""PixVerse 官方 API：提交文生视频任务 → 轮询状态 → 下载结果，并把每次请求记到 jobs.jsonl。
官方要点（docs.platform.pixverse.ai）：
- 请求头 API-KEY，以及每个请求都要新的 Ai-trace-id（UUID v4）
- **HTTP 200 不等于成功**：只有 ErrCode=0 才算成功（key 无效、余额不足、限流都返回 HTTP 200）
- 轮询 GET /openapi/v2/video/result/{video_id}，每 3～5 秒一次；status 1 完成、5 生成中、7 审核未通过、8 失败
- 任务一直是 5 时不要重复提交，继续查原来的 video_id
key 只放在服务端环境变量 PIXVERSE_API_KEY 里，不要放进前端或日志。"""
import argparse, json, os, sys, time, uuid, requests

API = os.environ.get("PIXVERSE_API", "https://app-api.pixverse.ai")
RETRY = {712008: "请求频率超限", 500044: "并发数已满", 500069: "服务繁忙"}
STOP = {10004: "缺少 API key", 10005: "API key 未注册", 400012: "账号无效", 500020: "没有权限", 500090: "积分余额不足",
        500054: "内容审核未通过", 500063: "预审核未通过", 400018: "提示词太长", 400013: "参数类型或取值错误"}

class PixVerseError(Exception):
    def __init__(self, code, msg, trace):
        label = STOP.get(code) or RETRY.get(code) or ""
        super().__init__(f"ErrCode {code}（{label + '：' if label else ''}{msg}）trace={trace}"); self.code = code

class Client:
    def __init__(self, key=None, sleep=None, log="jobs.jsonl"):
        # sleep 默认写成 time.sleep 会在定义函数时就绑定，测试里替换 time.sleep 不起作用，所以运行时再取
        self.key = key or os.environ.get("PIXVERSE_API_KEY"); self.sleep = sleep or (lambda s: time.sleep(s)); self.log = log
        if not self.key: raise SystemExit("没有设置 PIXVERSE_API_KEY")

    def call(self, method, path, body=None, tries=4):
        for i in range(tries):
            trace = str(uuid.uuid4())                                    # 每个请求一个新的 trace id
            r = requests.request(method, API + path, json=body, timeout=60,
                                 headers={"API-KEY": self.key, "Ai-trace-id": trace, "User-Agent": "binbho-tutorial/1.0"})
            j = r.json() if r.headers.get("Content-Type", "").startswith("application/json") else {}
            code = j.get("ErrCode", -1 if r.status_code != 200 else 0)
            self._log({"t": time.strftime("%F %T"), "path": path, "http": r.status_code, "ErrCode": code, "ErrMsg": j.get("ErrMsg"), "trace": trace})
            if r.status_code == 200 and code == 0: return j["Resp"]
            if code in RETRY and i < tries - 1: self.sleep(2 ** i * 2); continue    # 2、4、8 秒退避
            raise PixVerseError(code, j.get("ErrMsg") or f"HTTP {r.status_code}", trace)

    def _log(self, rec):
        with open(self.log, "a", encoding="utf-8") as f: f.write(json.dumps(rec, ensure_ascii=False) + "\n")

    def text_to_video(self, prompt, model="v6", duration=5, quality="540p", aspect_ratio="16:9", seed=None):
        body = {"model": model, "prompt": prompt, "duration": duration, "quality": quality, "aspect_ratio": aspect_ratio}
        if seed is not None: body["seed"] = seed
        return self.call("POST", "/openapi/v2/video/text/generate", body)["video_id"]

    def wait(self, video_id, every=4, timeout=900):
        t0 = time.monotonic()
        while True:
            r = self.call("GET", f"/openapi/v2/video/result/{video_id}")
            if r["status"] == 1 and r.get("url"): return r
            if r["status"] == 7: raise PixVerseError(500054, "生成结果审核未通过：修改内容后再新建任务，不要原样重试", "-")
            if r["status"] == 8: raise PixVerseError(8, "生成失败：保留 trace id，在账户流水中核对积分是否退回", "-")
            if time.monotonic() - t0 > timeout: raise TimeoutError(f"{video_id} 超过 {timeout} 秒仍未完成；不要重复提交，稍后继续查询")
            self.sleep(every)

def download(url, path):
    with requests.get(url, stream=True, timeout=120) as r:
        r.raise_for_status()
        with open(path, "wb") as f:
            for chunk in r.iter_content(1 << 16): f.write(chunk)
    return path

def main(argv=None):
    ap = argparse.ArgumentParser(); ap.add_argument("prompt"); ap.add_argument("--model", default="v6"); ap.add_argument("--duration", type=int, default=5)
    ap.add_argument("--quality", default="540p"); ap.add_argument("--aspect", default="16:9"); ap.add_argument("--seed", type=int); ap.add_argument("--out", default="clip.mp4")
    ap.add_argument("--resume", type=int, help="继续查询已有的 video_id，而不是新建任务"); a = ap.parse_args(argv)
    c = Client()
    try:
        vid = a.resume or c.text_to_video(a.prompt, a.model, a.duration, a.quality, a.aspect, a.seed); print("video_id", vid)
        r = c.wait(vid); print("完成", r["outputWidth"], "x", r["outputHeight"], "seed", r.get("seed")); print("保存", download(r["url"], a.out))
    except PixVerseError as e:
        print("PixVerse 错误：", e); return 1
    return 0

if __name__ == "__main__":
    sys.exit(main())

第 3 步：本地替身

"""本地替身（我写的，不是 PixVerse 官方的）。错误码和 HTTP 状态按官方 Error Codes 页面；10005 的响应体是对真实接口的实测结果；
status 取值按官方 Query Task Status 页面。生成的“视频”只是几个字节的占位数据。"""
import json, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

class State:
    def __init__(self): self.calls = []; self.traces = []; self.script = []; self.polls_until_done = 2; self.final = 1; self.next_err = []

def make(st):
    class H(BaseHTTPRequestHandler):
        def log_message(self, *a): pass
        def send(self, code, obj, ctype="application/json"):
            b = obj if isinstance(obj, bytes) else json.dumps(obj).encode(); self.send_response(code); self.send_header("Content-Type", ctype)
            self.send_header("Content-Length", str(len(b))); self.end_headers(); self.wfile.write(b)
        def auth(self):
            st.calls.append((self.command, self.path)); st.traces.append(self.headers.get("Ai-trace-id"))
            if self.path.startswith("/files/"): return True
            k = self.headers.get("API-KEY")
            if not k: self.send(200, {"ErrCode": 10004, "ErrMsg": "apiKey is missing"}); return False
            if k != "good": self.send(200, {"ErrCode": 10005, "ErrMsg": "apiKey is not registered"}); return False
            if st.next_err: c = st.next_err.pop(0); self.send(200, {"ErrCode": c, "ErrMsg": "simulated"}); return False
            return True
        def do_POST(self):
            n = int(self.headers.get("Content-Length", 0)); body = json.loads(self.rfile.read(n) or b"{}")
            if not self.auth(): return
            if self.path == "/openapi/v2/video/text/generate":
                if not all(k in body for k in ("model", "prompt", "duration", "quality", "aspect_ratio")): return self.send(200, {"ErrCode": 400011, "ErrMsg": "missing param"})
                if not isinstance(body["duration"], int): return self.send(400, {"ErrCode": 400013, "ErrMsg": "invalid type"})
                return self.send(200, {"ErrCode": 0, "ErrMsg": "success", "Resp": {"video_id": 123456789}})
            self.send(404, {"ErrCode": 99999, "ErrMsg": "not found"})
        def do_GET(self):
            if not self.auth(): return
            if self.path.startswith("/files/"): return self.send(200, b"\x00\x00\x00\x18ftypmp42demo", "video/mp4")
            if self.path.startswith("/openapi/v2/video/result/"):
                st.polls_until_done -= 1; done = st.polls_until_done <= 0
                status = st.final if done else 5
                url = f"http://{self.headers['Host']}/files/123456789.mp4"
                return self.send(200, {"ErrCode": 0, "ErrMsg": "success", "Resp": {"id": 123456789, "status": status, "url": url,   # 生成中也可能已有 url
                                  "outputWidth": 960 if status == 1 else 0, "outputHeight": 540 if status == 1 else 0, "seed": 12345}})
            self.send(404, {"ErrCode": 99999, "ErrMsg": "not found"})
    srv = ThreadingHTTPServer(("127.0.0.1", 0), H); threading.Thread(target=srv.serve_forever, daemon=True).start()
    return srv, f"http://127.0.0.1:{srv.server_address[1]}"

第 4 步：演示脚本

"""用本地替身演示一次完整流程（不连接 PixVerse，不花积分）：python demo_local.py [key]"""
import os, sys, fake_pixverse as F
st = F.State(); st.next_err = [712008]                 # 第一次提交模拟“请求频率超限”
srv, url = F.make(st); os.environ["PIXVERSE_API_KEY"] = sys.argv[1] if len(sys.argv) > 1 else "good"
import pixverse_job as P
P.API = url
try: code = P.main(["A blue glass marble rolls across a clean white table", "--duration", "1", "--quality", "360p", "--out", "clip.mp4"])
finally: srv.shutdown()
sys.exit(code)

第 5 步：测试

import json, pytest, fake_pixverse as F, pixverse_job as P

@pytest.fixture
def env(monkeypatch, tmp_path):
    st = F.State(); srv, url = F.make(st); monkeypatch.setattr(P, "API", url); monkeypatch.chdir(tmp_path)
    slept = []; yield st, slept; srv.shutdown()

def client(env, key="good"): return P.Client(key, sleep=env[1].append)

def test_http200_but_bad_key(env):
    # 真实接口：无效 key 返回 HTTP 200 + {"ErrCode":10005,"ErrMsg":"apiKey is not registered"}
    with pytest.raises(P.PixVerseError, match="10005"):
        client(env, "bad").text_to_video("a marble")

def test_full_flow_and_trace_ids(env):
    st, slept = env; c = client(env)
    vid = c.text_to_video("A blue glass marble rolls across a clean white table", duration=1, quality="360p")
    r = c.wait(vid)
    assert vid == 123456789 and r["status"] == 1 and r["outputWidth"] == 960
    assert slept == [4] and len(set(st.traces)) == len(st.traces) == 3          # 每个请求的 trace id 都不同
    assert P.download(r["url"], "clip.mp4") and open("clip.mp4", "rb").read(8)[4:8] == b"ftyp"
    log = [json.loads(l) for l in open("jobs.jsonl")]
    assert [l["ErrCode"] for l in log] == [0, 0, 0] and "API-KEY" not in open("jobs.jsonl").read()

def test_url_present_while_generating_is_not_done(env):
    st, _ = env; st.polls_until_done = 3; c = client(env)
    assert c.wait(1)["status"] == 1 and len([x for x in st.calls if "result" in x[1]]) == 3

def test_retry_on_rate_limit_then_stop_on_balance(env):
    st, slept = env; st.next_err = [712008, 712008]
    assert client(env).text_to_video("x") == 123456789 and slept == [2, 4]
    st.next_err = [500090]
    with pytest.raises(P.PixVerseError, match="积分余额不足"): client(env).text_to_video("x")

def test_moderation_and_failure_states(env):
    st, _ = env; st.final = 7
    with pytest.raises(P.PixVerseError, match="不要原样重试"): client(env).wait(1)
    st.final = 8; st.polls_until_done = 1
    with pytest.raises(P.PixVerseError, match="积分是否退回"): client(env).wait(1)

def test_resume_does_not_resubmit(env, monkeypatch):
    st, _ = env; monkeypatch.setenv("PIXVERSE_API_KEY", "good"); monkeypatch.setattr(P.time, "sleep", lambda s: None)
    assert P.main(["unused", "--resume", "123456789"]) == 0
    assert not any(m == "POST" for m, _ in st.calls)

第 6 步：运行

$ python demo_local.py
video_id 123456789
完成 960 x 540 seed 12345
保存 clip.mp4
$ cat jobs.jsonl
{"t": "2026-10-21 20:25:01", "path": "/openapi/v2/video/text/generate", "http": 200, "ErrCode": 712008, "ErrMsg": "simulated", "trace": "78f68d75-a393-4be2-986a-bf2273c35ce8"}
{"t": "2026-10-21 20:25:03", "path": "/openapi/v2/video/text/generate", "http": 200, "ErrCode": 0, "ErrMsg": "success", "trace": "622164c6-6555-4de8-a035-187c06b5a4c1"}
{"t": "2026-10-21 20:25:03", "path": "/openapi/v2/video/result/123456789", "http": 200, "ErrCode": 0, "ErrMsg": "success", "trace": "3ba9a29b-afd1-4c91-9361-12e57e0baaaf"}
{"t": "2026-10-21 20:25:07", "path": "/openapi/v2/video/result/123456789", "http": 200, "ErrCode": 0, "ErrMsg": "success", "trace": "c07c3627-dc37-49e5-bd55-f9c084105a8c"}
$ python demo_local.py bad
PixVerse 错误： ErrCode 10005（API key 未注册：apiKey is not registered）trace=d0f6a771-e76f-4efb-8b14-71b0b2030986
$ curl -s -w ' HTTP %{http_code}' -X POST https://app-api.pixverse.ai/openapi/v2/video/text/generate \
    -H 'API-KEY: invalid' -H "Ai-trace-id: $(uuidgen)" -H 'Content-Type: application/json' \
    -d '{"model":"v6","prompt":"A blue glass marble rolls across a white table","duration":1,"quality":"360p","aspect_ratio":"16:9"}'     # 真实接口
{"ErrCode":10005,"ErrMsg":"apiKey is not registered"} HTTP 200
$ python -m pytest -q
6 passed

怎么检查结果？

• 日志中每一行的 trace 都不同，且不包含 API key。
• 限流（712008）、并发满（500044）、服务繁忙（500069）会退避重试；余额不足、审核未通过、key 错误不会重试。
• --resume video_id 只查询，不会新建任务（测试检查没有 POST）。
• 联系客服时提供 trace id 和 video_id，不要提供完整 key。

调试示例

1. HTTP 200 的失败。 用无效 key 调用真实接口，返回的是 HTTP 200 和 {"ErrCode":10005,"ErrMsg":"apiKey is not registered"}。如果按一般习惯写 r.raise_for_status()，这一步不会报错，后面读取 Resp.video_id 时才会出现 KeyError，很难看出原因。官方错误码页面也明确写了：只有 ErrCode=0 才算成功。脚本对每个响应都同时检查 HTTP 状态和 ErrCode。

2. 错误信息丢了处理建议。 第一版的错误信息只显示错误码对应的名称。生成结果审核未通过时，我传入的提示“修改内容后再新建任务，不要原样重试”被名称“内容审核未通过”替换掉了，测试因此失败。修复：名称和具体说明都显示。

3. 测试用了 7 秒。 Client.__init__(sleep=time.sleep) 中的默认值在函数定义时就已绑定，测试里替换 time.sleep 不起作用，轮询时真的等了 4 秒。修复：默认值改为 None，运行时再取 time.sleep。测试时间降到约 3 秒。

常见问题

模板规则

• 同时检查 HTTP 状态和 ErrCode。
• 每个请求都用新的 trace id，并记入日志。
• 只在 status=1 时下载。
• 不重复提交仍在生成的任务。
• API key 只放在服务端。

项目资源

把错误码处理好，积分才不会浪费在重复提交上。