Mubert API：生成音乐并轮询曲目状态

Mubert API v3：POST /api/v3/public/tracks（customer-id + access-token），可用 playlist_index 或 prompt（Text-2-Music），再 GET /tracks/{id} 轮询至 generations[].status=done。

先看清成品

• mock 创建后轮询得到 mp3 URL。
• 官方无效凭证 HTTP 401。
• 1 项 pytest 通过。

合规提醒：Mubert 曲目授权取决于套餐与合同；免费/低档常需署名且用途受限。发布前核对 license 与是否允许商用/变现平台。曲目下载 URL 会过期，及时落盘。本文不编造套餐价格。

运行说明：无正式 MUBERT_CUSTOMER_ID / ACCESS_TOKEN。A/B 用 mock；对 music-api.mubert.com 无效凭证实测 HTTP 401。C 为本地 license 门禁。2026-10-23（Asia/Shanghai）pytest 通过。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

111-mubert-A-src/
├── generate_track.py
├── mock_mubert.py
└── test_111a.py

第 1 步：客户端

#!/usr/bin/env python3
"""Mubert API v3: create a track then poll until generations[].status == done.

Auth (public): customer-id + access-token headers.
Prefer playlist_index OR prompt (text-2-music). Save the download URL before it expires.
"""
from __future__ import annotations
import argparse, json, os, sys, time, urllib.error, urllib.request


def _req(method, url, headers, body=None):
    data = None if body is None else json.dumps(body).encode()
    h = {**headers}
    if data is not None:
        h["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=h, method=method)
    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 create_track(base, customer_id, token, *, playlist_index=None, prompt=None, duration=30, fmt="mp3", mode="track", intensity="medium"):
    body = {"duration": duration, "format": fmt, "mode": mode, "intensity": intensity}
    if prompt:
        body["prompt"] = prompt
    elif playlist_index:
        body["playlist_index"] = playlist_index
    else:
        raise SystemExit("需要 --playlist-index 或 --prompt")
    return _req("POST", f"{base.rstrip('/')}/api/v3/public/tracks",
                {"customer-id": customer_id, "access-token": token}, body)


def get_track(base, customer_id, token, track_id):
    return _req("GET", f"{base.rstrip('/')}/api/v3/public/tracks/{track_id}",
                {"customer-id": customer_id, "access-token": token})


def wait_done(base, customer_id, token, track_id, *, timeout=30, interval=0.2):
    deadline = time.time() + timeout
    while time.time() < deadline:
        data = get_track(base, customer_id, token, track_id)
        gens = (data.get("data") or data).get("generations") or []
        if gens and gens[0].get("status") == "done" and gens[0].get("url"):
            return data
        time.sleep(interval)
    raise SystemExit("轮询超时：请检查配额或改用 webhook")


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--base", default=os.environ.get("MUBERT_BASE_URL", "https://music-api.mubert.com"))
    ap.add_argument("--customer-id", default=os.environ.get("MUBERT_CUSTOMER_ID", ""))
    ap.add_argument("--token", default=os.environ.get("MUBERT_ACCESS_TOKEN", ""))
    ap.add_argument("--playlist-index", default=None)
    ap.add_argument("--prompt", default=None)
    ap.add_argument("--duration", type=int, default=30)
    ap.add_argument("--format", default="mp3", choices=["mp3", "wav"])
    ap.add_argument("--mode", default="track", choices=["track", "loop", "jingle", "mix"])
    ap.add_argument("--intensity", default="medium", choices=["low", "medium", "high"])
    ap.add_argument("--poll", action="store_true", help="创建后轮询至 done")
    args = ap.parse_args(argv)
    if not args.customer_id or not args.token:
        print("设置 MUBERT_CUSTOMER_ID 与 MUBERT_ACCESS_TOKEN（由 service/customers 签发）", file=sys.stderr)
        return 2
    created = create_track(
        args.base, args.customer_id, args.token,
        playlist_index=args.playlist_index, prompt=args.prompt,
        duration=args.duration, fmt=args.format, mode=args.mode, intensity=args.intensity,
    )
    tid = (created.get("data") or created).get("id")
    out = wait_done(args.base, args.customer_id, args.token, tid) if args.poll else created
    print(json.dumps(out, ensure_ascii=False, indent=2))
    return 0


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

第 2 步：Mock

#!/usr/bin/env python3
"""Mock Mubert public tracks API."""
from __future__ import annotations
import json, uuid
from http.server import BaseHTTPRequestHandler, HTTPServer

CID, TOK = "cust-test", "tok-test"
TRACKS = {}

class H(BaseHTTPRequestHandler):
    def log_message(self, *a): pass
    def _auth(self):
        return self.headers.get("customer-id") == CID and self.headers.get("access-token") == TOK
    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": "Unauthorized"}); return
        if self.path.rstrip("/") != "/api/v3/public/tracks":
            self.send_error(404); return
        n = int(self.headers.get("Content-Length", 0))
        body = json.loads(self.rfile.read(n) or b"{}")
        if not body.get("playlist_index") and not body.get("prompt"):
            self._send(400, {"error": "playlist_index or prompt required"}); return
        tid = str(uuid.uuid4())
        TRACKS[tid] = {
            "id": tid, "playlist_index": body.get("playlist_index"),
            "prompt": body.get("prompt"), "duration": body.get("duration", 30),
            "intensity": body.get("intensity", "high"), "mode": body.get("mode", "track"),
            "generations": [{"session_id": "g1", "format": body.get("format", "mp3"),
                             "bitrate": 128, "status": "processing", "url": None}],
        }
        self._send(200, {"data": TRACKS[tid]})
    def do_GET(self):
        if not self._auth(): self._send(401, {"error": "Unauthorized"}); return
        prefix = "/api/v3/public/tracks/"
        if not self.path.startswith(prefix):
            self.send_error(404); return
        tid = self.path[len(prefix):].split("?")[0]
        t = TRACKS.get(tid)
        if not t: self._send(404, {"error": "not found"}); return
        t = dict(t)
        t["generations"] = [{**t["generations"][0], "status": "done",
                             "url": f"https://cdn.example/mock/{tid}.mp3"}]
        TRACKS[tid] = t
        self._send(200, {"data": t})

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

if __name__ == "__main__":
    main()

第 3 步：测试

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

def test_generate_and_poll():
    port = 19111
    proc = subprocess.Popen([sys.executable, str(HERE/"mock_mubert.py"), "--port", str(port)],
                            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        time.sleep(0.3)
        r = subprocess.run(
            [sys.executable, str(HERE/"generate_track.py"), "--base", f"http://127.0.0.1:{port}",
             "--customer-id", "cust-test", "--token", "tok-test",
             "--prompt", "calm lo-fi study beat", "--duration", "30", "--poll"],
            capture_output=True, text=True, check=True)
        data = json.loads(r.stdout)
        gens = (data.get("data") or data)["generations"]
        assert gens[0]["status"] == "done"
        assert gens[0]["url"].endswith(".mp3")
    finally:
        proc.terminate(); proc.wait(timeout=5)

第 4 步：运行

$ pytest 111a
.                                                                        [100%]
1 passed in 0.36s

$ invalid key music-api.mubert.com
HTTP 401
{"message":"Unauthenticated","code":"Unauthenticated"}

怎么检查结果？

• generations[0].status 为 done 且有 url。
• 正式调用前先跑 C 篇 license 门禁。

调试示例

401 Unauthenticated。 确认 customer/access 来自 service/customers，不是把 company/license 拿来调 public。

常见问题

模板规则

• 凭证分环境保管。
• 先 Library 后生成。
• URL 过期前落盘。

项目资源

生成通了，授权条款才是上线前最后一关。