#!/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("--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("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): qs = "" if query: qs = "?" + urllib.parse.urlencode(query) url = f"http://{host}:{port}{route}{qs}" data = None headers = {} if body is not None: data = json.dumps(body).encode("utf-8") headers["Content-Type"] = "application/json" req = urllib.request.Request(url=url, data=data, headers=headers, 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) 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_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 "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 = call(host, port, "POST", "/play", body=with_q(None)) elif args.cmd == "add": status, payload = call(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 = call(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) # `stdin_isatty` controls whether we *auto*-select (non-interactive) # versus *prompt* the user. `--pick` overrides a non-TTY stdin so # scripts can still drive the picker via a piped index if they want. interactive = force_pick or ( sys.stdin.isatty() and not force_auto ) if not interactive: # Legacy / scripted path: top result only, no round-trip for # the candidate list. The daemon falls back to ytsearch1:NAME # when no URL is supplied. status, payload = call( host, port, "POST", "/init", body={"song": q} ) else: limit = getattr(args, "limit", None) query = {"q": q} if limit is not None and limit > 0: query["limit"] = str(limit) # yt-dlp searches can easily take 30+ seconds on a slow network # (the daemon calls `ytsearchN:QUERY -j` which hits YouTube). # The default 15 s `urllib` timeout short-circuits the picker # before the daemon even returns, so give this one endpoint a # generous budget. All other endpoints keep the 15 s default. status, payload = call( host, port, "GET", "/search/yt", query=query, timeout=120 ) obj = try_parse(payload) results = (obj or {}).get("results") or [] if status != 200: print_response(args, status, payload) sys.exit(1) if not results: print( f"error: no yt-dlp results for {q!r}", file=sys.stderr, ) sys.exit(1) for i, r in enumerate(results, 1): uploader = r.get("uploader") or "(unknown)" print( f"[{i}] {r.get('title', '?')} -- {uploader} " f"({fmt_secs(r.get('duration_secs'))})" ) 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 == "": chosen_idx = 0 else: try: chosen_idx = int(raw) - 1 except ValueError: print( f"error: not a number: {raw!r}", file=sys.stderr, ) sys.exit(2) if chosen_idx < 0 or chosen_idx >= len(results): print( f"error: choice out of range (1-{len(results)})", file=sys.stderr, ) sys.exit(2) chosen = results[chosen_idx] status, payload = call( host, port, "POST", "/init", body={"name": q, "url": chosen.get("url") or ""}, ) elif args.cmd == "init-batch": status, payload = call( 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 = call(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 = call(host, port, "GET", "/search/all", query={"q": q}) elif args.cmd == "queue": status, payload = call(host, port, "GET", "/queue") elif args.cmd in ("now", "status"): status, payload = call(host, port, "GET", "/now-playing") elif args.cmd == "skip": status, payload = call(host, port, "POST", "/skip") elif args.cmd == "pause": status, payload = call(host, port, "POST", "/pause") elif args.cmd == "resume": status, payload = call(host, port, "POST", "/resume") elif args.cmd == "seek": status, payload = call(host, port, "POST", "/seek", body={"secs": args.secs}) elif args.cmd == "shuffle": status, payload = call(host, port, "POST", "/queue/shuffle") elif args.cmd == "clear": status, payload = call(host, port, "POST", "/queue/clear") elif args.cmd == "songs": status, payload = call(host, port, "GET", "/songs") elif args.cmd == "downloads": status, payload = call(host, port, "GET", "/downloads") elif args.cmd == "history": status, payload = call(host, port, "GET", "/history") elif args.cmd == "stats": status, payload = call(host, port, "GET", "/stats") elif args.cmd == "volume": if args.value is None: status, payload = call(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 = call(host, port, "POST", "/volume", body={"vol": v}) elif args.cmd == "repeat": if args.mode is None: status, payload = call(host, port, "GET", "/repeat") else: status, payload = call(host, port, "POST", "/repeat", body={"mode": args.mode}) elif args.cmd == "playlists": status, payload = call(host, port, "GET", "/playlists") elif args.cmd == "help": status, payload = call(host, port, "GET", "/help") elif args.cmd == "pl-new": status, payload = call(host, port, "POST", "/playlists", body={"name": args.name}) elif args.cmd == "pl-show": status, payload = call(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 = call( host, port, "POST", f"{pl_route(args.name)}/add", body={"query": q} ) elif args.cmd == "pl-play": status, payload = call(host, port, "POST", f"{pl_route(args.name)}/play") elif args.cmd == "pl-rm": status, payload = call(host, port, "DELETE", pl_route(args.name)) elif args.cmd == "pl-rm-song": status, payload = call( host, port, "DELETE", f"{pl_route(args.name)}/songs/{args.song_id}" ) elif args.cmd == "pl-rename": status, payload = call( host, port, "PATCH", pl_route(args.old), body={"name": args.new} ) elif args.cmd == "pl-dup": status, payload = call( host, port, "POST", f"{pl_route(args.src)}/duplicate", body={"name": args.dest} ) elif args.cmd == "pl-reorder": status, payload = call( host, port, "POST", f"{pl_route(args.name)}/reorder", body={"from": args.from_pos, "to": args.to_pos}, ) else: status, payload = call(host, port, "GET", "/help") print_response(args, status, payload) if __name__ == "__main__": main()