#!/usr/bin/env python3 """Master Blaster from the command line (Python 3, standard library only). python3 master.py track.wav --out ./mastered master a whole track python3 master.py track.wav --clip the 20 s demo excerpt python3 master.py usage limits, usage, prepaid tracks, plan python3 master.py buy track 5 a Stripe link to pay for 5 tracks (a person opens it) python3 master.py buy plan 2 a Stripe link for a monthly plan of 2 x 12 tracks The key is read from MASTERBLASTER_API_KEY (starts with mb_). It is never printed. Exit codes: 0 done, 1 error, 2 payment needed (402), 3 wait and retry (429/503). """ import argparse import json import os import sys import time import urllib.error import urllib.parse import urllib.request BASE = os.environ.get("MASTERBLASTER_URL", "https://masterblaster.mediaatlas.si").rstrip("/") PRICING = "https://masterblaster.mediaatlas.si/#pricing" def key() -> str: k = os.environ.get("MASTERBLASTER_API_KEY", "").strip() if not k: sys.exit(f"No key: set MASTERBLASTER_API_KEY (an mb_... key). To get one, a person pays once: {PRICING}") return k def call(method: str, path: str, data: bytes | None = None, headers: dict | None = None, auth: bool = True) -> tuple[int, dict, dict]: h = {"User-Agent": "masterblaster-skill/1", **(headers or {})} if auth: h["Authorization"] = f"Bearer {key()}" req = urllib.request.Request(BASE + path, data, h, method=method) try: with urllib.request.urlopen(req, timeout=300) as r: return r.status, json.loads(r.read() or b"{}"), dict(r.headers) except urllib.error.HTTPError as e: try: body = json.loads(e.read()) except ValueError: body = {} return e.code, body, dict(e.headers) def fail(status: int, body: dict, headers: dict) -> None: err = body.get("error") or {} msg = err.get("message") if isinstance(err, dict) else err code = err.get("code") if isinstance(err, dict) else "" print(f"Error {status} {code}: {msg}", file=sys.stderr) if status == 402: print(f"Payment is needed. A person can pay here: {PRICING}\nOr make a link: python3 {sys.argv[0]} buy track 5", file=sys.stderr) sys.exit(2) if status in (429, 503): print(f"Wait {headers.get('Retry-After', '60')} s and try again.", file=sys.stderr) sys.exit(3) sys.exit(1) def download(path: str, to: str) -> None: req = urllib.request.Request(BASE + path, headers={"Authorization": f"Bearer {key()}", "User-Agent": "masterblaster-skill/1"}) with urllib.request.urlopen(req, timeout=600) as r, open(to, "wb") as f: while chunk := r.read(1 << 20): f.write(chunk) def master(a: argparse.Namespace) -> None: q = {"mode": "clip" if a.clip else "full", "name": os.path.basename(a.file)} if a.chips: q["chips"] = a.chips if a.target: q["target"] = a.target path = "/v1/masters?" + urllib.parse.urlencode(q) with open(a.file, "rb") as f: status, body, headers = call("POST", path, f.read()) if status != 202: fail(status, body, headers) jid = body["id"] print(f"Submitted {jid}", file=sys.stderr) last = "" while True: status, job, headers = call("GET", f"/v1/masters/{jid}") if status == 429: time.sleep(int(headers.get("Retry-After", "10"))) continue if status != 200: fail(status, job, headers) line = f"{job['status']}" + (f" (place {job['position']} in the queue)" if job.get("position") else "") + (f": {job['stage']}" if job.get("stage") else "") if line != last: print(line, file=sys.stderr) last = line if job["status"] == "done": break if job["status"] == "error": sys.exit(f"The job failed: {job.get('error')}") time.sleep(a.every) os.makedirs(a.out, exist_ok=True) for name in ("original", "mastered"): download(job["files"][name], os.path.join(a.out, f"{name}.wav")) r = job["result"] proof = job.get("proof") or {} print(json.dumps({"files": [os.path.join(a.out, "original.wav"), os.path.join(a.out, "mastered.wav")], "mode": r["mode"], "duration_s": r["duration_s"], "original": r["original"], "mastered": r["mastered"], "gain_db": r["gain_db"], "findings": r["findings"], "proof_id": proof.get("id") if isinstance(proof, dict) else proof, "expires_in_s": job.get("expires_s")}, indent=2)) def usage(_: argparse.Namespace) -> None: status, body, headers = call("GET", "/v1/usage") if status != 200: fail(status, body, headers) print(json.dumps(body, indent=2)) def buy(a: argparse.Namespace) -> None: has_key = bool(os.environ.get("MASTERBLASTER_API_KEY", "").strip()) status, body, headers = call("POST", "/api/checkout", json.dumps({"kind": a.kind, "n": a.n}).encode(), {"Content-Type": "application/json"}, auth=has_key and a.kind == "track") if status != 200: fail(status, body, headers) what = f"{a.n} track(s)" if a.kind == "track" else f"a monthly plan of {a.n * 12} tracks" print(f"Ask the person to open this link and pay for {what}:\n{body['url']}") print("After paying they see an API key once; they send it to you (MASTERBLASTER_API_KEY)." if not has_key or a.kind == "plan" else "The tracks are added to your key as soon as the payment is done.") def main() -> None: p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) sub = p.add_subparsers(dest="cmd") u = sub.add_parser("usage") u.set_defaults(fn=usage) b = sub.add_parser("buy") b.add_argument("kind", choices=["track", "plan"]) b.add_argument("n", type=int, help="tracks to buy, or packs of 12 tracks a month (1 to 7)") b.set_defaults(fn=buy) # the default command: master FILE if len(sys.argv) > 1 and sys.argv[1] not in ("usage", "buy", "-h", "--help"): m = argparse.ArgumentParser(prog="master.py") m.add_argument("file") m.add_argument("--out", default="./mastered") m.add_argument("--chips", help="comma-separated wishes, e.g. balance,warmer (GET /api/prompts)") m.add_argument("--target", help="source, -14, -16 or -18 (LUFS)") m.add_argument("--clip", action="store_true", help="the 20 s demo excerpt instead of the whole track") m.add_argument("--every", type=int, default=6, help="seconds between polls (keep it above 2)") a = m.parse_args() if not os.path.isfile(a.file): sys.exit(f"No such file: {a.file}") return master(a) a = p.parse_args() if not getattr(a, "fn", None): p.print_help() sys.exit(1) a.fn(a) if __name__ == "__main__": main()