Auphonic API：创建 Production、上传音频并启动处理

官方 JSON API 典型回路：POST /api/productions.json 创建 →（可选）multipart 上传 input_file → POST .../start.json。也可在创建时给 input_file URL 并带 action: start。 本篇客户端 + mock 跑通本地回路；Authorization 使用 bearer <token>。

先看清成品

• mock 完成创建、上传与 start。
• 官方无效 Token HTTP 403。
• 1 项 pytest 通过。

合规提醒：只处理你有权处理的录音；采访/播客需取得嘉宾与版权授权。Auphonic 按处理时长计费，免费额度与套餐以官网为准，本文不编造分钟数或价格。API Token 等同账号权限，勿提交公开仓库。

运行说明：无正式 Auphonic Token。A/C 为本地元数据与章节校验；B 用 mock 演示 create→upload→start。对官方 auphonic.com/api 无效 Token 实测 HTTP 403（Token doesn't exist）。2026-10-23（Asia/Shanghai）pytest 通过。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

101-auphonic-B-src/
├── auphonic_client.py
├── mock_auphonic.py
└── test_101b.py

第 1 步：客户端

#!/usr/bin/env python3
"""Auphonic JSON API: create production, optional upload path, start."""
from __future__ import annotations

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


def req(base, key, method, path, payload=None, multipart_file=None):
    headers = {"Authorization": f"bearer {key}"}
    data = None
    if multipart_file:
        # minimal multipart for mock/tests; real uploads should use requests or similar
        boundary = "----binbhoBoundary"
        filename = multipart_file.split("/")[-1]
        with open(multipart_file, "rb") as f:
            content = f.read()
        body = (
            f"--{boundary}\r\nContent-Disposition: form-data; name=\"input_file\"; filename=\"{filename}\"\r\n"
            f"Content-Type: application/octet-stream\r\n\r\n"
        ).encode() + content + f"\r\n--{boundary}--\r\n".encode()
        headers["Content-Type"] = f"multipart/form-data; boundary={boundary}"
        data = body
    elif payload is not None:
        data = json.dumps(payload).encode()
        headers["Content-Type"] = "application/json"
    request = urllib.request.Request(f"{base.rstrip('/')}{path}", data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(request, 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 main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--base", default=os.environ.get("AUPHONIC_BASE_URL", "https://auphonic.com"))
    ap.add_argument("--key", default=os.environ.get("AUPHONIC_API_KEY", ""))
    ap.add_argument("--preset", default="My Podcast Preset")
    ap.add_argument("--title", default="My first Production")
    ap.add_argument("--input-url", default="", help="HTTP(S) input_file URL; skips local upload")
    ap.add_argument("--upload", default="", help="local audio path for upload step")
    ap.add_argument("--start", action="store_true")
    args = ap.parse_args(argv)
    if not args.key:
        print("设置 AUPHONIC_API_KEY", file=sys.stderr)
        return 2
    payload = {"preset": args.preset, "metadata": {"title": args.title}}
    if args.input_url:
        payload["input_file"] = args.input_url
        if args.start:
            payload["action"] = "start"
    created = req(args.base, args.key, "POST", "/api/productions.json", payload)
    data = created.get("data") or {}
    uid = data.get("uuid")
    print("uuid", uid, "status", data.get("status_string"))
    if args.upload:
        if not uid:
            print("no uuid", file=sys.stderr)
            return 1
        up = req(args.base, args.key, "POST", f"/api/production/{uid}/upload.json", multipart_file=args.upload)
        print("upload", (up.get("data") or {}).get("status_string"))
    if args.start and not args.input_url:
        st = req(args.base, args.key, "POST", f"/api/production/{uid}/start.json")
        print("start", (st.get("data") or {}).get("status_string"))
    return 0


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

第 2 步：Mock

#!/usr/bin/env python3
"""Mock Auphonic productions create / upload / start."""
from __future__ import annotations

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

API_KEY = "auphonic-test-key"
PRODS: dict[str, dict] = {}


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

    def _auth(self) -> bool:
        a = self.headers.get("Authorization", "")
        return a.lower() == f"bearer {API_KEY}".lower()

    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 not self._auth():
            self._send(401, {"error_message": "Unauthorized", "status_code": 401})
            return
        n = int(self.headers.get("Content-Length", 0))
        raw = self.rfile.read(n) if n else b""
        path = self.path.split("?", 1)[0]
        if path == "/api/productions.json":
            body = json.loads(raw or b"{}")
            uid = uuid.uuid4().hex[:22]
            PRODS[uid] = {
                "uuid": uid,
                "status_string": "Waiting for input file" if not body.get("input_file") else "Waiting",
                "metadata": body.get("metadata") or {},
                "preset": body.get("preset"),
                "has_file": bool(body.get("input_file")),
                "started": body.get("action") == "start" and bool(body.get("input_file")),
            }
            if PRODS[uid]["started"]:
                PRODS[uid]["status_string"] = "Done"
            self._send(200, {"status_code": 200, "data": PRODS[uid], "error_message": ""})
            return
        if path.startswith("/api/production/") and path.endswith("/upload.json"):
            uid = path.split("/")[3]
            if uid not in PRODS:
                self._send(404, {"error_message": "not found"})
                return
            # multipart ignored; mark file present
            PRODS[uid]["has_file"] = True
            PRODS[uid]["status_string"] = "Ready to start"
            self._send(200, {"status_code": 200, "data": PRODS[uid]})
            return
        if path.startswith("/api/production/") and path.endswith("/start.json"):
            uid = path.split("/")[3]
            if uid not in PRODS:
                self._send(404, {"error_message": "not found"})
                return
            if not PRODS[uid].get("has_file"):
                self._send(400, {"error_message": "no input file", "status_code": 400})
                return
            PRODS[uid]["started"] = True
            PRODS[uid]["status_string"] = "Done"
            self._send(200, {"status_code": 200, "data": PRODS[uid]})
            return
        self.send_error(404)


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


if __name__ == "__main__":
    main()

第 3 步：测试

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

def test_flow(tmp_path):
    port = 19101
    proc = subprocess.Popen([sys.executable, str(HERE/"mock_auphonic.py"), "--port", str(port)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        time.sleep(0.3)
        audio = tmp_path / "a.wav"
        audio.write_bytes(b"RIFF....WAVEfmt ")
        r = subprocess.run([
            sys.executable, str(HERE/"auphonic_client.py"),
            "--base", f"http://127.0.0.1:{port}", "--key", "auphonic-test-key",
            "--upload", str(audio), "--start", "--title", "demo"
        ], capture_output=True, text=True, check=True)
        assert "uuid" in r.stdout and "Done" in r.stdout
    finally:
        proc.terminate(); proc.wait(timeout=5)

第 4 步：运行

$ pytest -q (mock create/upload/start)
.                                                                        [100%]
1 passed in 0.37s

$ invalid key against api.auphonic.com
HTTP 403
{
    "status_code": 403,
    "error_code": null,
    "error_message": "Token doesn't exist",
    "form_errors": {},
    "data": null
}

怎么检查结果？

• pytest 绿。
• 正式调用前先用小文件验证 Token 与 preset。

调试示例

403 Token doesn't exist。 到 Auphonic 账户重新生成 API Token，确认头是 bearer 而不是 Basic 拼错。不要把网页 Cookie 当 API Token。

常见问题

模板规则

• 先小文件试跑。
• Token 勿入库。
• 尊重嘉宾授权。

项目资源

create → upload → start，再谈批量播客流水线。