#!/usr/bin/env python3 """ sjnmusic - tiny CLI for sjnmusicd HTTP daemon. Talks to the webserver at http://127.0.0.1:14567 by default. Override with --host or --port, or set SJNMUSIC_HOST / SJNMUSIC_PORT environment variables. """ import argparse import json import os import sys import time import urllib.error import urllib.parse import urllib.request DEFAULT_HOST = "127.0.0.1" DEFAULT_PORT = 14567 def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="sjnmusic", description="Talk to the sjnmusicd HTTP daemon.", ) p.add_argument("--host", default=os.environ.get("SJNMUSIC_HOST", DEFAULT_HOST)) p.add_argument( "--port", type=int, default=int(os.environ.get("SJNMUSIC_PORT", str(DEFAULT_PORT))), ) p.add_argument( "--token", default=os.environ.get("SJNMUSIC_TOKEN"), help=( "bearer token for the daemon (Authorization: Bearer ). " "Required when talking to a non-loopback daemon. Can also be set " "via the SJNMUSIC_TOKEN environment variable. On loopback the " "CLI auto-discovers the token from GET /auth/token." ), ) p.add_argument("--json", action="store_true", help="print raw JSON responses") sub = p.add_subparsers(dest="cmd", required=True) def add_search_fields(sp): sp.add_argument("query", nargs="*", help="search text") sp.add_argument("-q", "--query", dest="query_flag", help="search text") return sp add_search_fields(sub.add_parser("play", help="search & start playing immediately")) add_search_fields(sub.add_parser("add", help="search & append to queue")) add_search_fields(sub.add_parser("del", help="search & delete from disk+db")) init_p = add_search_fields( sub.add_parser( "init", help="download via yt-dlp and register (async); picks a yt-dlp candidate first if interactive", ) ) init_p.add_argument( "--pick", action="store_true", help="force interactive candidate picker even when stdin is not a TTY", ) init_p.add_argument( "--auto", action="store_true", help="skip the candidate picker and download yt-dlp's top result", ) init_p.add_argument( "--limit", type=int, default=None, help="number of yt-dlp candidates to fetch from /search/yt (default: daemon config)", ) add_search_fields(sub.add_parser("search", help="search the song database (best match)")) add_search_fields(sub.add_parser("search-all", help="search every match above threshold")) sub.add_parser("queue", help="show the current queue") sub.add_parser("now", help="alias for /now-playing") sub.add_parser("status", help="one-line summary of current playback") sub.add_parser("skip", help="skip the next song") sub.add_parser("pause", help="pause playback") sub.add_parser("resume", help="resume playback") sub.add_parser("shuffle", help="shuffle the queue in place") sub.add_parser("clear", help="empty the queue") sub.add_parser( "play-all", help="shuffle the entire library into the queue and start playing", ) sub.add_parser("songs", help="list all known songs") sub.add_parser("playlists", help="list all playlists") sub.add_parser("downloads", help="list recent yt-dlp download jobs") sub.add_parser("history", help="show recently played songs") sub.add_parser("stats", help="aggregate stats from history") sub.add_parser("help", help="list HTTP endpoints") vol = sub.add_parser("volume", help="get/set volume (0-100 or floating 0.0-2.0)") vol.add_argument("value", nargs="?", type=float, help="0-100 (interpreted as percent)") seek = sub.add_parser("seek", help="jump to position within the current track") seek.add_argument("secs", type=float, help="position in seconds") repeat = sub.add_parser("repeat", help="get/set repeat mode (off/one/all)") repeat.add_argument("mode", nargs="?", help="off, one, all") pl_new = sub.add_parser("pl-new", help="create a playlist") pl_new.add_argument("name") pl_show = sub.add_parser("pl-show", help="show playlist contents") pl_show.add_argument("name") pl_add = sub.add_parser("pl-add", help="add a song to a playlist") pl_add.add_argument("name") pl_add.add_argument("query", nargs="*") pl_add.add_argument("-q", "--query", dest="query_flag") pl_play = sub.add_parser("pl-play", help="load a playlist into the queue") pl_play.add_argument("name") pl_rm = sub.add_parser("pl-rm", help="delete a playlist") pl_rm.add_argument("name") pl_rm_song = sub.add_parser("pl-rm-song", help="remove a song from a playlist") pl_rm_song.add_argument("name") pl_rm_song.add_argument("song_id", type=int) pl_rename = sub.add_parser("pl-rename", help="rename a playlist") pl_rename.add_argument("old") pl_rename.add_argument("new") pl_dup = sub.add_parser("pl-dup", help="duplicate a playlist under a new name") pl_dup.add_argument("src") pl_dup.add_argument("dest") pl_reorder = sub.add_parser("pl-reorder", help="reorder songs in a playlist") pl_reorder.add_argument("name") pl_reorder.add_argument("from_pos", type=int) pl_reorder.add_argument("to_pos", type=int) init_batch = sub.add_parser("init-batch", help="batch-download several songs at once") init_batch.add_argument("queries", nargs="+") return p def join_query(args) -> str: parts = [] if getattr(args, "query_flag", None): parts.append(args.query_flag) if getattr(args, "query", None): parts.extend(args.query) return " ".join(parts).strip() def call(host, port, method, route, body=None, query=None, timeout=15, headers=None): qs = "" if query: qs = "?" + urllib.parse.urlencode(query) url = f"http://{host}:{port}{route}{qs}" data = None hdrs = dict(headers or {}) if body is not None: data = json.dumps(body).encode("utf-8") hdrs.setdefault("Content-Type", "application/json") req = urllib.request.Request(url=url, data=data, headers=hdrs, method=method) try: with urllib.request.urlopen(req, timeout=timeout) as resp: payload = resp.read().decode("utf-8") return resp.status, payload except urllib.error.HTTPError as e: return e.code, e.read().decode("utf-8", errors="replace") except urllib.error.URLError as e: print(f"error: cannot reach daemon at {url}: {e.reason}", file=sys.stderr) sys.exit(2) # Loopback addresses that skip the bearer-token check on the daemon # side. Kept in sync with the daemon's `auth::is_local_request` so the # CLI auto-fetch fallback only ever fires when the daemon would # actually let us through unauthenticated. LOCAL_HOSTS = frozenset({"127.0.0.1", "::1", "localhost"}) def is_local(host): return host in LOCAL_HOSTS def fetch_token(host, port, timeout=5): """ Best-effort localhost-only token probe. GET /auth/token is itself loopback-only on the daemon side, so reaching it implies we're a process running on the same host. We do NOT retry on failure — a 403 here means a remote daemon is pretending to be local and refusing (we should propagate the error), a 404 means the daemon has auth disabled (don't auto-fetch any token), and any other non-200 is opaque. We surface the decoded JSON `token` field on success and `None` on every other outcome. """ try: status, payload = call( host, port, "GET", "/auth/token", timeout=timeout ) except SystemExit: # `call` exits the process on URLError; treat that as # "couldn't reach daemon, no token". We re-raise because the # caller is already going to error out on the original # request anyway, but only after a transparent re-raise so # surfacing happens at the right stack frame. return None if status != 200: return None obj = try_parse(payload) if not isinstance(obj, dict): return None tok = obj.get("token") if isinstance(tok, str) and tok: return tok return None def request(args, host, port, method, route, body=None, query=None, timeout=15): """ Auth-aware wrapper around `call`. Order of operations: 1. If the operator supplied `--token` (or `SJNMUSIC_TOKEN`), attach `Authorization: Bearer ` and send straight to the daemon. We never auto-fetch on top of an explicit token — that would silently swap a user-supplied secret with the daemon's current one (which is rarely what an operator wants). 2. Otherwise: if the peer is loopback, send without a token and then, if the daemon rejected with 401, do a one-shot `GET /auth/token` and retry. This keeps today's CLI workflows unchanged: a user with auto-fetch works even though the daemon now requires tokens for non-loopback callers. We decline to silently fetch on a non-loopback host because that would defeat the whole point of forcing operators to give an explicit token to remote daemons. 3. If the peer is non-loopback and no token was supplied, the call proceeds without a header so the operator gets a daemon-issued 401 instead of a CLI-level error — easier to debug in the daemon logs. """ explicit_token = getattr(args, "token", None) or "" hdrs = {} if explicit_token: hdrs["Authorization"] = f"Bearer {explicit_token}" status, payload = call( host, port, method, route, body=body, query=query, timeout=timeout, headers=hdrs, ) # Auto-fetch only on FIRST observation of 401 on a loopback peer # when no explicit token was supplied AND this is not already the # token endpoint itself (which would obviously never require a # token). The 401 must have come from a request method that the # server would normally respond 2xx to without auth; if /auth/token # is what we just hit, we'd loop forever. if ( status == 401 and not explicit_token and is_local(host) and route != "/auth/token" ): fetched = fetch_token(host, port) if fetched: hdrs["Authorization"] = f"Bearer {fetched}" status, payload = call( host, port, method, route, body=body, query=query, timeout=timeout, headers=hdrs, ) return status, payload def fmt_secs(s): if s is None: return "--:--" try: s = float(s) except (TypeError, ValueError): return "--:--" if s < 0: return "--:--" m, sec = divmod(int(s), 60) return f"{m:d}:{sec:02d}" def try_parse(payload): try: return json.loads(payload) except json.JSONDecodeError: return None def print_candidate(idx, c): """Pretty-print one ranked candidate row for the picker.""" score = c.get("score") score_str = f"score={score:+d}" if score is not None else "" flags = " ".join(c.get("flags") or []) flag_str = f" [{flags}]" if flags else "" print( f"[{idx}] {c.get('title', '?')} -- " f"{c.get('uploader') or '(unknown)':24} " f"({fmt_secs(c.get('duration_secs')):>5}) " f"{score_str}{flag_str}" ) def interactive_pick(candidates): """Show the picker; return the user's choice (or candidates[0]).""" for i, c in enumerate(candidates, 1): print_candidate(i, c) try: raw = input("Pick a song [1]: ").strip() except EOFError: print( "error: no input available; re-run with --auto or supply a piped number", file=sys.stderr, ) sys.exit(2) if raw == "": return candidates[0] try: idx = int(raw) - 1 except ValueError: print(f"error: not a number: {raw!r}", file=sys.stderr) sys.exit(2) if idx < 0 or idx >= len(candidates): print( f"error: choice out of range (1-{len(candidates)})", file=sys.stderr, ) sys.exit(2) return candidates[idx] def legacy_pick(results, force_auto): """Legacy /search/yt picker (no scoring fields).""" if force_auto: return results[0] return interactive_pick(results) def do_init(args, host, port, q, force_pick, force_auto, limit): """ Drive the `init` subcommand through the new ranked pipeline. Default behaviour: `GET /pick?q=…` and let the daemon either auto-select (when the top score beats the runner-up by margin) or surface ranked candidates for the picker. Falls back to the legacy `/search/yt` picker if the daemon doesn't speak `/pick` (older builds), so the CLI stays forward/backward compatible. Returns the chosen candidate dict (with at least a 'url' key) so `main()` can dispatch the actual `/init` POST uniformly. """ pick_query = {"q": q} if limit is not None and limit > 0: pick_query["limit"] = str(limit) # 3 parallel yt-dlp fetches can easily take 30+ seconds each # on a slow network; the daemon drops its state mutex before # invoking them, but the network round-trips themselves are # unchanged. Generous timeout to avoid ghost-failing. pick_status, pick_payload = request( args, host, port, "GET", "/pick", query=pick_query, timeout=180 ) pick_obj = try_parse(pick_payload) if pick_status != 200 or pick_obj is None or "kind" not in pick_obj: # Fall back to legacy `/search/yt` if the daemon is too old # to know about `/pick`. Keeps the CLI working across daemon # upgrades in either direction. legacy_status, legacy_payload = request( args, host, port, "GET", "/search/yt", query=pick_query, timeout=120, ) legacy_obj = try_parse(legacy_payload) or {} legacy_results = legacy_obj.get("results") or [] if legacy_status != 200: return (legacy_status, legacy_payload) if not legacy_results: print( f"error: no yt-dlp results for {q!r}", file=sys.stderr, ) sys.exit(1) return ( 200, json.dumps({"chosen": legacy_pick(legacy_results, force_auto)}), ) kind = pick_obj["kind"] if kind == "empty": print( f"error: no yt-dlp results for {q!r}: " f"{pick_obj.get('message', '')}", file=sys.stderr, ) sys.exit(1) if kind == "auto": # Daemon is confident — surface the choice so the user sees # what was picked, then download. score = pick_obj.get("score", 0) print( f"[auto-pick score={score}] " f"{pick_obj.get('title', '?')} -- " f"{pick_obj.get('uploader', '(unknown)')} " f"({fmt_secs(pick_obj.get('duration_secs'))})" ) if force_pick: # `--pick` overrides the auto decision: ask the user. ranked_status, ranked_payload = request( args, host, port, "GET", "/search/yt/ranked", query=pick_query, timeout=180, ) ranked_obj = try_parse(ranked_payload) or {} ranked_results = ranked_obj.get("results") or [] if ranked_status != 200 or not ranked_results: print_response_args(args, ranked_status, ranked_payload) sys.exit(1) return ( 200, json.dumps({"chosen": interactive_pick(ranked_results)}), ) return ( 200, json.dumps({ "chosen": { "url": pick_obj.get("url"), "title": pick_obj.get("title"), "uploader": pick_obj.get("uploader"), "score": score, "duration_secs": pick_obj.get("duration_secs"), "flags": pick_obj.get("flags", []), }, }), ) if kind == "needs_choice": candidates = pick_obj.get("candidates") or [] if not candidates: print( f"error: no yt-dlp results for {q!r}", file=sys.stderr, ) sys.exit(1) if force_auto: chosen = candidates[0] print( f"[auto-pick score={chosen.get('score', 0)}] " f"{chosen.get('title', '?')} -- " f"{chosen.get('uploader', '(unknown)')} " f"({fmt_secs(chosen.get('duration_secs'))})" ) else: chosen = interactive_pick(candidates) return (200, json.dumps({"chosen": chosen})) print(f"error: unknown /pick kind {kind!r}", file=sys.stderr) sys.exit(1) def print_response_args(args, status, payload): """Helper for subroutes that already pre-formatted a payload.""" print_response(args, status, payload) def print_response(args, status, payload): if args.json: print(f"HTTP {status}\n{payload}") return obj = try_parse(payload) if obj is None: print(f"HTTP {status}: {payload}") return if isinstance(obj, dict) and "error" in obj: print(f"error: {obj['error']}", file=sys.stderr) sys.exit(1) # Specialised formatters used by status / history / stats / downloads. if ( isinstance(obj, dict) and "current" in obj and "queue_len" in obj and "elapsed_secs" in obj ): cur = obj.get("current") or {} name = cur.get("name") or "(idle)" elapsed = fmt_secs(obj.get("elapsed_secs")) total = fmt_secs(obj.get("duration_secs")) print( f"[{cur.get('id', '-')}] {name} {elapsed}/{total} " f"q={obj.get('queue_len', 0)} " f"vol={obj.get('volume', 0):.2f} " f"paused={obj.get('paused', False)} " f"repeat={obj.get('repeat', 'off')}" ) return if isinstance(obj, dict) and "history" in obj: for h in obj["history"]: ts = h.get("played_at", 0) when = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(ts)) print( f"{when} [{h.get('song_id', '-')}] {h.get('song_name') or '(missing)'} " f"{fmt_secs(h.get('duration_secs_played', 0))}" ) return if isinstance(obj, dict) and "top_songs" in obj: print(f"total plays: {obj.get('total_plays', 0)}") print(f"total time: {fmt_secs(obj.get('total_secs', 0))}") for t in obj.get("top_songs", []): print( f" - [{t.get('song_id', '-')}] {t.get('name') or '(missing)'} " f"plays={t.get('plays', 0)} time={fmt_secs(t.get('total_secs', 0))}" ) return if isinstance(obj, dict) and "downloads" in obj: for d in obj["downloads"]: extras = [] if d.get("song_id") is not None: extras.append(f"song={d['song_id']}") if d.get("error"): extras.append(f"err={d['error']}") extra_str = (" " + " ".join(extras)) if extras else "" print(f"[{d.get('id', '-')}] {d.get('status', '?')} {d.get('name', '?')}{extra_str}") return if isinstance(obj, dict) and "matches" in obj: for s in obj["matches"]: print(f"[{s['id']}] {s['name']}") return if isinstance(obj, dict) and "queue" in obj and "current" in obj: repeat = obj.get("repeat", "off") print(f"queue (repeat={repeat}):") for s in obj.get("queue", []): print(f" - [{s['id']}] {s['name']}") cur = obj.get("current") if cur: print(f"now playing: [{cur['id']}] {cur['name']}") else: print("now playing: (idle)") return if isinstance(obj, dict) and "queued" in obj: print(f"queued {obj['queued']} song(s) in random order") return if isinstance(obj, dict) and "songs" in obj: for s in obj["songs"]: print(f"[{s['id']}] {s['name']}") return if isinstance(obj, list): if obj and isinstance(obj[0], dict) and obj[0].get("songs") is not None: for pl in obj: print(f"{pl['name']} ({len(pl.get('songs', []))} songs)") for s in pl.get("songs", []): print(f" - [{s['id']}] {s['name']}") else: for s in obj: print(s) return if isinstance(obj, dict) and "name" in obj and "songs" in obj: pl = obj print(f"playlist: {pl['name']}") for s in pl.get("songs", []): print(f" - [{s['id']}] {s['name']}") return if isinstance(obj, dict) and "endpoints" in obj: for ep in obj["endpoints"]: print(ep) return if isinstance(obj, dict) and "added" in obj: print(f"added [{obj['added']['id']}] {obj['added']['name']} -> {obj['playlist']['name']}") return print(json.dumps(obj, indent=2)) def main(argv=None): args = build_parser().parse_args(argv) host, port = args.host, args.port q = join_query(args) def with_q(body): body = dict(body or {}) if q: body.setdefault("query", q) return body def pl_route(name): return f"/playlists/{urllib.parse.quote(name)}" if args.cmd == "play": status, payload = request(args, host, port, "POST", "/play", body=with_q(None)) elif args.cmd == "add": status, payload = request(args, host, port, "POST", "/add", body=with_q(None)) elif args.cmd == "del": if not q: print("error: 'del' requires a search query", file=sys.stderr) sys.exit(2) status, payload = request(args, host, port, "POST", "/del", body=with_q(None)) elif args.cmd == "init": if not q: print("error: 'init' requires a song name", file=sys.stderr) sys.exit(2) force_pick = getattr(args, "pick", False) force_auto = getattr(args, "auto", False) limit = getattr(args, "limit", None) # Drive the ranked pipeline (do_init returns the chosen # candidate as a JSON envelope so we don't need separate # branches below for `/init`). Routing through `request` # means auto-fetch-from-/auth/token also works for the # search (/pick) and the eventual /init POST. _, envelope_payload = do_init( args, host, port, q, force_pick, force_auto, limit, ) chosen_obj = try_parse(envelope_payload) or {} chosen = chosen_obj.get("chosen") or {} status, payload = request( args, host, port, "POST", "/init", body={"name": q, "url": chosen.get("url") or ""}, ) elif args.cmd == "init-batch": status, payload = request( args, host, port, "POST", "/init/batch", body={"songs": args.queries} ) elif args.cmd == "search": if not q: print("error: 'search' requires a query", file=sys.stderr) sys.exit(2) status, payload = request(args, host, port, "GET", "/search", query={"q": q}) elif args.cmd == "search-all": if not q: print("error: 'search-all' requires a query", file=sys.stderr) sys.exit(2) status, payload = request(args, host, port, "GET", "/search/all", query={"q": q}) elif args.cmd == "queue": status, payload = request(args, host, port, "GET", "/queue") elif args.cmd in ("now", "status"): status, payload = request(args, host, port, "GET", "/now-playing") elif args.cmd == "skip": status, payload = request(args, host, port, "POST", "/skip") elif args.cmd == "pause": status, payload = request(args, host, port, "POST", "/pause") elif args.cmd == "resume": status, payload = request(args, host, port, "POST", "/resume") elif args.cmd == "seek": status, payload = request(args, host, port, "POST", "/seek", body={"secs": args.secs}) elif args.cmd == "shuffle": status, payload = request(args, host, port, "POST", "/queue/shuffle") elif args.cmd == "clear": status, payload = request(args, host, port, "POST", "/queue/clear") elif args.cmd == "play-all": status, payload = request(args, host, port, "POST", "/play/all", body={}) elif args.cmd == "songs": status, payload = request(args, host, port, "GET", "/songs") elif args.cmd == "downloads": status, payload = request(args, host, port, "GET", "/downloads") elif args.cmd == "history": status, payload = request(args, host, port, "GET", "/history") elif args.cmd == "stats": status, payload = request(args, host, port, "GET", "/stats") elif args.cmd == "volume": if args.value is None: status, payload = request(args, host, port, "GET", "/volume") else: # Treat 0..100 as a percentage of rodio's default 1.0. v = args.value if v > 2.0: v = v / 100.0 status, payload = request(args, host, port, "POST", "/volume", body={"vol": v}) elif args.cmd == "repeat": if args.mode is None: status, payload = request(args, host, port, "GET", "/repeat") else: status, payload = request(args, host, port, "POST", "/repeat", body={"mode": args.mode}) elif args.cmd == "playlists": status, payload = request(args, host, port, "GET", "/playlists") elif args.cmd == "help": status, payload = request(args, host, port, "GET", "/help") elif args.cmd == "pl-new": status, payload = request(args, host, port, "POST", "/playlists", body={"name": args.name}) elif args.cmd == "pl-show": status, payload = request(args, host, port, "GET", pl_route(args.name)) elif args.cmd == "pl-add": if not q: print("error: 'pl-add' requires a song query", file=sys.stderr) sys.exit(2) status, payload = request( args, host, port, "POST", f"{pl_route(args.name)}/add", body={"query": q} ) elif args.cmd == "pl-play": status, payload = request(args, host, port, "POST", f"{pl_route(args.name)}/play") elif args.cmd == "pl-rm": status, payload = request(args, host, port, "DELETE", pl_route(args.name)) elif args.cmd == "pl-rm-song": status, payload = request( args, host, port, "DELETE", f"{pl_route(args.name)}/songs/{args.song_id}" ) elif args.cmd == "pl-rename": status, payload = request( args, host, port, "PATCH", pl_route(args.old), body={"name": args.new} ) elif args.cmd == "pl-dup": status, payload = request( args, host, port, "POST", f"{pl_route(args.src)}/duplicate", body={"name": args.dest} ) elif args.cmd == "pl-reorder": status, payload = request( args, host, port, "POST", f"{pl_route(args.name)}/reorder", body={"from": args.from_pos, "to": args.to_pos}, ) else: status, payload = request(args, host, port, "GET", "/help") print_response(args, status, payload) if __name__ == "__main__": main()