Mubert Music Library：按节奏与时长搜索曲库

曲库检索：GET /api/v3/public/music-library/tracks，可按 bpm、duration、genres 等过滤。不耗生成配额，应作为默认首选。

先看清成品

• mock 返回带 url 的曲目与 meta.total。
• 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-B-src/
├── library_search.py
├── mock_mubert_lib.py
└── test_111b.py

第 1 步：客户端

#!/usr/bin/env python3
"""Mubert Music Library: GET /api/v3/public/music-library/tracks (free vs generating).

Search curated catalogue before spending a generation quota.
"""
from __future__ import annotations
import argparse, json, os, sys, urllib.error, urllib.parse, urllib.request


def search(base, customer_id, token, **filters):
    qs = urllib.parse.urlencode({k: v for k, v in filters.items() if v is not None and v != ""})
    url = f"{base.rstrip('/')}/api/v3/public/music-library/tracks"
    if qs:
        url += "?" + qs
    req = urllib.request.Request(url, headers={"customer-id": customer_id, "access-token": token}, method="GET")
    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 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("--bpm", default=None)
    ap.add_argument("--duration", default=None)
    ap.add_argument("--genres", default=None)
    args = ap.parse_args(argv)
    if not args.customer_id or not args.token:
        print("设置 MUBERT_CUSTOMER_ID 与 MUBERT_ACCESS_TOKEN", file=sys.stderr); return 2
    r = search(args.base, args.customer_id, args.token, bpm=args.bpm, duration=args.duration, genres=args.genres)
    print(json.dumps(r, ensure_ascii=False, indent=2))
    return 0


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

第 2 步：Mock

#!/usr/bin/env python3
"""Mock Mubert music-library tracks."""
from __future__ import annotations
import json
from urllib.parse import parse_qs, urlparse
from http.server import BaseHTTPRequestHandler, HTTPServer

CID, TOK = "cust-test", "tok-test"

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_GET(self):
        if self.headers.get("customer-id") != CID or self.headers.get("access-token") != TOK:
            self._send(401, {"error": "Unauthorized"}); return
        u = urlparse(self.path)
        if u.path != "/api/v3/public/music-library/tracks":
            self.send_error(404); return
        qs = parse_qs(u.query)
        bpm = (qs.get("bpm") or ["120"])[0]
        dur = int((qs.get("duration") or ["30"])[0])
        self._send(200, {
            "data": [{
                "id": "lib-1", "duration": dur, "bpm": int(bpm), "mode": "track",
                "intensity": "medium",
                "generations": [{"status": "done", "format": "mp3", "bitrate": 320,
                                 "url": "https://cdn.example/lib/calm-30.mp3"}],
            }],
            "meta": {"offset": 0, "limit": 100, "total": 1},
        })

def main():
    import argparse
    ap = argparse.ArgumentParser(); ap.add_argument("--port", type=int, default=9112)
    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_library_search():
    port = 19112
    proc = subprocess.Popen([sys.executable, str(HERE/"mock_mubert_lib.py"), "--port", str(port)],
                            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        time.sleep(0.3)
        r = subprocess.run(
            [sys.executable, str(HERE/"library_search.py"), "--base", f"http://127.0.0.1:{port}",
             "--customer-id", "cust-test", "--token", "tok-test", "--bpm", "120", "--duration", "30"],
            capture_output=True, text=True, check=True)
        data = json.loads(r.stdout)
        assert data["meta"]["total"] >= 1
        assert data["data"][0]["generations"][0]["url"]
    finally:
        proc.terminate(); proc.wait(timeout=5)

第 4 步：运行

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

怎么检查结果？

• data 非空且含 generations.url。
• meta.total ≥ 1。

调试示例

空结果。 放宽 bpm/genres，或改用 params 接口看该过滤组合是否有库存。

常见问题

模板规则

• 默认先搜曲库。
• 记录选用的 track id。
• 仍要核对授权。

项目资源

曲库命中，比盲目生成更省钱。