Synthesia Video API：创建视频并轮询完成状态

Video API 是异步的：POST /v2/videos 后轮询 GET /v2/videos/{id} 直到 complete，再用限时 download 链接取回 mp4。 这篇用 mock 把该回路跑通；请求头使用 Legacy (v2) API key。

先看清成品

• mock 创建后约 0.5s 变为 complete 并给出假 download URL。
• 官方无效密钥 HTTP 403。
• 1 项 pytest 通过。

合规提醒：定制以真人为原型的数字人必须取得本人同意，并按平台完成身份验证；不得用数字人冒充他人、从事欺诈或未披露的政治宣传。Video API 通常需 Pro 及以上（以官网为准）；test: true 有水印且有每日测试限额。本文不编造价格与配额数字（除文档已写明的测试限额提示外，以你控制台为准）。

运行说明：无 Synthesia 正式密钥。A/C 本地合规检查；B 用 mock 创建+轮询到 complete。对官方主机无效 Authorization 实测 HTTP 403。2026-10-23（Asia/Shanghai）pytest 通过。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

100-synthesia-B-src/
├── synthesia_client.py
├── mock_synthesia.py
└── test_100b.py

第 1 步：客户端

#!/usr/bin/env python3
"""Create a Synthesia test video and poll until complete (Legacy v2 key)."""
from __future__ import annotations

import argparse
import json
import os
import sys
import time
import urllib.error
import urllib.request


def request(base, key, method, path, payload=None):
    data = None if payload is None else json.dumps(payload).encode()
    req = urllib.request.Request(
        f"{base.rstrip('/')}{path}",
        data=data,
        headers={"Authorization": key, "Content-Type": "application/json"},
        method=method,
    )
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return resp.status, 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 main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--base", default=os.environ.get("SYNTHESIA_BASE_URL", "https://api.synthesia.io"))
    ap.add_argument("--key", default=os.environ.get("SYNTHESIA_API_KEY", ""))
    ap.add_argument("--title", default="My first Synthetic video")
    ap.add_argument(
        "--script",
        default="Hello, World! This is my first synthetic video, made with the Synthesia API!",
    )
    ap.add_argument("--avatar", default="anna_costume1_cameraA")
    ap.add_argument("--background", default="green_screen")
    ap.add_argument("--no-test", action="store_true")
    ap.add_argument("--poll", type=float, default=0.2)
    ap.add_argument("--timeout", type=float, default=10)
    args = ap.parse_args(argv)
    if not args.key:
        print("设置 SYNTHESIA_API_KEY（Legacy v2 scope）", file=sys.stderr)
        return 2
    payload = {
        "test": not args.no_test,
        "title": args.title,
        "scriptText": args.script,
        "avatar": args.avatar,
        "background": args.background,
    }
    code, created = request(args.base, args.key, "POST", "/v2/videos", payload)
    print("create", code, created.get("id"), created.get("status"))
    vid = created["id"]
    deadline = time.time() + args.timeout
    while time.time() < deadline:
        _, info = request(args.base, args.key, "GET", f"/v2/videos/{vid}")
        st = info.get("status")
        print("status", st)
        if st == "complete":
            print("download", info.get("download"))
            return 0
        if st in {"error", "rejected"}:
            print(info)
            return 1
        time.sleep(args.poll)
    print("timeout", file=sys.stderr)
    return 1


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

第 2 步：Mock

#!/usr/bin/env python3
"""Mock Synthesia Video API create + retrieve."""
from __future__ import annotations

import json
import time
import uuid
from http.server import BaseHTTPRequestHandler, HTTPServer

API_KEY = "syn-test-key"
VIDEOS: dict[str, dict] = {}


class H(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def _auth(self) -> bool:
        return self.headers.get("Authorization") == API_KEY

    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.path != "/v2/videos":
            self.send_error(404)
            return
        if not self._auth():
            self._send(403, {"error": "Forbidden - Wrong API key"})
            return
        n = int(self.headers.get("Content-Length", 0))
        body = json.loads(self.rfile.read(n) or b"{}")
        vid = str(uuid.uuid4())
        VIDEOS[vid] = {
            "id": vid,
            "title": body.get("title") or "Untitled",
            "status": "in_progress",
            "createdAt": int(time.time()),
            "test": bool(body.get("test", False)),
            "scriptText": body.get("scriptText") or "",
            "_ready_at": time.time() + 0.5,
        }
        self._send(201, {"id": vid, "status": "in_progress"})

    def do_GET(self):
        if not self.path.startswith("/v2/videos/"):
            self.send_error(404)
            return
        if not self._auth():
            self._send(403, {"error": "Forbidden - Wrong API key"})
            return
        vid = self.path.split("/v2/videos/", 1)[1].split("?", 1)[0]
        v = VIDEOS.get(vid)
        if not v:
            self._send(404, {"error": "not found"})
            return
        if v["status"] == "in_progress" and time.time() >= v["_ready_at"]:
            v["status"] = "complete"
            v["download"] = f"http://127.0.0.1/fake/{vid}.mp4"
            v["duration"] = "0:05"
        out = {k: val for k, val in v.items() if not k.startswith("_")}
        self._send(200, out)


def main():
    import argparse

    ap = argparse.ArgumentParser()
    ap.add_argument("--port", type=int, default=9000)
    args = ap.parse_args()
    HTTPServer(("127.0.0.1", args.port), H).serve_forever()


if __name__ == "__main__":
    main()

第 3 步：测试

import subprocess
import sys
import time
from pathlib import Path

HERE = Path(__file__).resolve().parent


def test_create_poll():
    port = 19000
    proc = subprocess.Popen(
        [sys.executable, str(HERE / "mock_synthesia.py"), "--port", str(port)],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    try:
        time.sleep(0.3)
        r = subprocess.run(
            [
                sys.executable,
                str(HERE / "synthesia_client.py"),
                "--base",
                f"http://127.0.0.1:{port}",
                "--key",
                "syn-test-key",
                "--title",
                "demo",
                "--timeout",
                "5",
            ],
            capture_output=True,
            text=True,
            check=True,
        )
        assert "create" in r.stdout
        assert "complete" in r.stdout
        assert "download" in r.stdout
    finally:
        proc.terminate()
        proc.wait(timeout=5)

第 4 步：运行

.                                                                        [100%]
1 passed in 0.97s
create 201 b419a017-f034-4b15-9e59-ce0fc049f1a5 in_progress
status in_progress
status in_progress
status in_progress
status complete
download http://127.0.0.1/fake/b419a017-f034-4b15-9e59-ce0fc049f1a5.mp4
HTTP 403
{"context":"User is not authenticated","error":"Forbidden"}

怎么检查结果？

• 默认 test=true。
• 正式环境轮询间隔应更长，或改用 webhook。

调试示例

Interactive Avatars key 不能当 Video API 用。 官方明确 Video 请求要 Legacy v2 scope。403 时先检查 key 类型而不是 endpoint 拼写。

常见问题

模板规则

• 先 test。
• 密钥按 scope 创建。
• 下载链勿长期公开。

项目资源

异步回路跑通，再接入业务系统。