poprhythm
435cb931b6
Distilled from tonight's session building SUB/WAVE anchor-playlist shows by hand: a CLI (search/artist/album/playlist get/create/replace/ add/remove) plus a playlist-audit command that flags accidental full-album dumps, replacing the one-off curl/Python snippets used throughout. SKILL.md documents the gotchas hit along the way (case-sensitive artist matching, deluxe-reissue duplicate tracks, createPlaylist's full-replace semantics) and this session's curation conventions (15-20 tracks/artist, prefer playlistStrict). Claude-Session: https://claude.ai/code/session_01L7Rwa6guD5wK8F8tWQwcJX
253 lines
10 KiB
Python
Executable File
253 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Navidrome Subsonic-API CLI. Auth/config lifted from /srv/subwave/state/setup-config.json.
|
|
|
|
Usage:
|
|
navidrome_api.py search <query> [--artists N] [--albums N] [--songs N]
|
|
navidrome_api.py artist <name> # case-insensitive; lists albums + track counts
|
|
navidrome_api.py album <artist> <album> # lists tracks with ids for one album
|
|
navidrome_api.py playlists # list all playlists (name, id, song count)
|
|
navidrome_api.py playlist get <id> # dump entries (artist | album | title | id)
|
|
navidrome_api.py playlist create <name> --ids id1,id2,...
|
|
navidrome_api.py playlist replace <id> --ids id1,id2,... # createPlaylist w/ playlistId: FULL REPLACE
|
|
navidrome_api.py playlist add <id> --ids id1,id2,... # updatePlaylist songIdToAdd: in-place append
|
|
navidrome_api.py playlist remove <id> --ids id1,id2,... # updatePlaylist songIdToRemove
|
|
navidrome_api.py playlist audit <id> # per-artist/per-album breakdown + full-album flags
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import secrets
|
|
import sys
|
|
import urllib.parse
|
|
import urllib.request
|
|
from collections import defaultdict
|
|
|
|
SETUP_CONFIG = "/srv/subwave/state/setup-config.json"
|
|
|
|
|
|
def load_config():
|
|
cfg = json.load(open(SETUP_CONFIG))["navidrome"]
|
|
return {
|
|
"url": cfg["url"].replace("navidrome:", "localhost:"),
|
|
"user": cfg["user"],
|
|
"pass": cfg["pass"],
|
|
}
|
|
|
|
|
|
def call(endpoint, params=None, doseq=False):
|
|
cfg = load_config()
|
|
salt = secrets.token_hex(6)
|
|
token = hashlib.md5((cfg["pass"] + salt).encode()).hexdigest()
|
|
p = {"u": cfg["user"], "t": token, "s": salt, "v": "1.16.1", "c": "navidrome-skill", "f": "json"}
|
|
if params:
|
|
p.update(params)
|
|
q = urllib.parse.urlencode(p, doseq=doseq)
|
|
with urllib.request.urlopen(f"{cfg['url']}/rest/{endpoint}?{q}") as r:
|
|
d = json.load(r)["subsonic-response"]
|
|
if d.get("status") != "ok":
|
|
err = d.get("error", {})
|
|
raise RuntimeError(f"{endpoint} failed: {err.get('code')} {err.get('message')}")
|
|
return d
|
|
|
|
|
|
def find_artist_exact(name):
|
|
"""Case-insensitive exact-name match. Navidrome artist tags are inconsistently
|
|
cased ('eels' lowercase, 'CAKE' uppercase) - a naive `== name` filter silently
|
|
drops real hits. This is the #1 false-negative bug hit repeatedly this session."""
|
|
resp = call("search3", {"query": name, "artistCount": 10, "songCount": 0, "albumCount": 0})
|
|
for a in resp.get("searchResult3", {}).get("artist", []):
|
|
if a["name"].lower() == name.lower():
|
|
return a
|
|
return None
|
|
|
|
|
|
def cmd_search(args):
|
|
resp = call("search3", {"query": args.query, "artistCount": args.artists,
|
|
"albumCount": args.albums, "songCount": args.songs})
|
|
r = resp.get("searchResult3", {})
|
|
for a in r.get("artist", []):
|
|
print(f"artist | {a['name']} [{a['id']}]")
|
|
for al in r.get("album", []):
|
|
print(f"album | {al['artist']} - {al['name']} ({al.get('songCount')} tracks) [{al['id']}]")
|
|
for s in r.get("song", []):
|
|
print(f"song | {s['artist']} - {s['title']} ({s.get('album')}) [{s['id']}]")
|
|
|
|
|
|
def cmd_artist(args):
|
|
a = find_artist_exact(args.name)
|
|
if not a:
|
|
print(f"No exact match for '{args.name}'. Loose/compilation tracks may still exist -- "
|
|
f"try: navidrome_api.py search '{args.name}' --songs 20")
|
|
return
|
|
r = call("getArtist", {"id": a["id"]})
|
|
albums = r.get("artist", {}).get("album", [])
|
|
if not albums:
|
|
print(f"{a['name']}: artist entry exists but 0 grouped albums -- checking loose tracks...")
|
|
resp = call("search3", {"query": args.name, "artistCount": 0, "songCount": 30, "albumCount": 0})
|
|
for s in resp.get("searchResult3", {}).get("song", []):
|
|
if s["artist"].lower() == args.name.lower():
|
|
print(f" {s['title']} ({s.get('album')}) [{s['id']}]")
|
|
return
|
|
total = sum(al.get("songCount", 0) for al in albums)
|
|
print(f"{a['name']}: {len(albums)} albums, {total} tracks")
|
|
for al in albums:
|
|
print(f" {al['name']} ({al.get('year')}) - {al.get('songCount')} tracks [{al['id']}]")
|
|
|
|
|
|
def cmd_album(args):
|
|
a = find_artist_exact(args.artist)
|
|
if not a:
|
|
print(f"No exact artist match for '{args.artist}'")
|
|
return
|
|
r = call("getArtist", {"id": a["id"]})
|
|
for al in r.get("artist", {}).get("album", []):
|
|
if al["name"].lower() == args.album.lower():
|
|
aresp = call("getAlbum", {"id": al["id"]})
|
|
for s in aresp.get("album", {}).get("song", []):
|
|
print(f" {s.get('track', '?')}. {s['title']} [{s['id']}]")
|
|
return
|
|
print(f"No album '{args.album}' found for {a['name']}")
|
|
|
|
|
|
def cmd_playlists(args):
|
|
resp = call("getPlaylists")
|
|
for p in resp.get("playlists", {}).get("playlist", []):
|
|
print(f"{p['id']} | {p.get('songCount', '?'):>4} tracks | {p['name']}")
|
|
|
|
|
|
def cmd_playlist_get(args):
|
|
resp = call("getPlaylist", {"id": args.id})
|
|
for e in resp["playlist"]["entry"]:
|
|
print(f"{e['artist']} | {e['album']} | {e['title']} [{e['id']}]")
|
|
|
|
|
|
def cmd_playlist_create(args):
|
|
ids = args.ids.split(",")
|
|
resp = call("createPlaylist", {"name": args.name, "songId": ids}, doseq=True)
|
|
pl = resp.get("playlist", {})
|
|
print(f"created '{args.name}': {pl.get('id')} ({pl.get('songCount')} tracks)")
|
|
|
|
|
|
def cmd_playlist_replace(args):
|
|
ids = args.ids.split(",")
|
|
resp = call("createPlaylist", {"playlistId": args.id, "songId": ids}, doseq=True)
|
|
pl = resp.get("playlist", {})
|
|
print(f"replaced {args.id}: now {pl.get('songCount')} tracks")
|
|
|
|
|
|
def cmd_playlist_add(args):
|
|
ids = args.ids.split(",")
|
|
call("updatePlaylist", {"playlistId": args.id, "songIdToAdd": ids}, doseq=True)
|
|
resp = call("getPlaylist", {"id": args.id})
|
|
print(f"added {len(ids)}; {args.id} now has {len(resp['playlist']['entry'])} tracks")
|
|
|
|
|
|
def cmd_playlist_remove(args):
|
|
remove_ids = set(args.ids.split(","))
|
|
resp = call("getPlaylist", {"id": args.id})
|
|
entries = resp["playlist"]["entry"]
|
|
remaining = [e["id"] for e in entries if e["id"] not in remove_ids]
|
|
resp2 = call("createPlaylist", {"playlistId": args.id, "songId": remaining}, doseq=True)
|
|
print(f"removed {len(entries) - len(remaining)}; {args.id} now has "
|
|
f"{resp2.get('playlist', {}).get('songCount')} tracks")
|
|
|
|
|
|
def cmd_playlist_audit(args):
|
|
"""The recurring 'did we dump a full album in here' check from tonight's session:
|
|
group by (artist, album), then compare each group's count against that album's
|
|
real total songCount. An exact match is a strong full-catalog signal worth a
|
|
second look -- but treat it as a flag, not an automatic verdict: a genuinely
|
|
short album (e.g. a 10-track record) will always show 10/10 with no trimming
|
|
possible, and an artist you were explicitly told to go 'heavy' on is fine full."""
|
|
resp = call("getPlaylist", {"id": args.id})
|
|
entries = resp["playlist"]["entry"]
|
|
print(f"total tracks: {len(entries)}")
|
|
by_album = defaultdict(list)
|
|
for e in entries:
|
|
by_album[(e["artist"], e["album"])].append(e)
|
|
|
|
for (artist, album), tracks in sorted(by_album.items(), key=lambda x: -len(x[1])):
|
|
if len(tracks) < args.min_size:
|
|
continue
|
|
real_total = None
|
|
try:
|
|
aresp = call("search3", {"query": album, "artistCount": 0, "songCount": 0, "albumCount": 10})
|
|
for al in aresp.get("searchResult3", {}).get("album", []):
|
|
if al["artist"] == artist and al["name"] == album:
|
|
real_total = al.get("songCount")
|
|
break
|
|
except Exception:
|
|
pass
|
|
flag = ""
|
|
if real_total is not None and real_total == len(tracks):
|
|
flag = " <-- FULL ALBUM (playlist count == real album length)"
|
|
print(f" {artist} - {album}: {len(tracks)}"
|
|
f"{f'/{real_total}' if real_total is not None else ''} tracks{flag}")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
sub = parser.add_subparsers(dest="cmd", required=True)
|
|
|
|
p = sub.add_parser("search")
|
|
p.add_argument("query")
|
|
p.add_argument("--artists", type=int, default=5)
|
|
p.add_argument("--albums", type=int, default=5)
|
|
p.add_argument("--songs", type=int, default=5)
|
|
p.set_defaults(func=cmd_search)
|
|
|
|
p = sub.add_parser("artist")
|
|
p.add_argument("name")
|
|
p.set_defaults(func=cmd_artist)
|
|
|
|
p = sub.add_parser("album")
|
|
p.add_argument("artist")
|
|
p.add_argument("album")
|
|
p.set_defaults(func=cmd_album)
|
|
|
|
p = sub.add_parser("playlists")
|
|
p.set_defaults(func=cmd_playlists)
|
|
|
|
pl = sub.add_parser("playlist")
|
|
pl_sub = pl.add_subparsers(dest="subcmd", required=True)
|
|
|
|
p = pl_sub.add_parser("get")
|
|
p.add_argument("id")
|
|
p.set_defaults(func=cmd_playlist_get)
|
|
|
|
p = pl_sub.add_parser("create")
|
|
p.add_argument("name")
|
|
p.add_argument("--ids", required=True, help="comma-separated song ids")
|
|
p.set_defaults(func=cmd_playlist_create)
|
|
|
|
p = pl_sub.add_parser("replace")
|
|
p.add_argument("id")
|
|
p.add_argument("--ids", required=True, help="comma-separated song ids; FULL REPLACE of contents")
|
|
p.set_defaults(func=cmd_playlist_replace)
|
|
|
|
p = pl_sub.add_parser("add")
|
|
p.add_argument("id")
|
|
p.add_argument("--ids", required=True, help="comma-separated song ids to append")
|
|
p.set_defaults(func=cmd_playlist_add)
|
|
|
|
p = pl_sub.add_parser("remove")
|
|
p.add_argument("id")
|
|
p.add_argument("--ids", required=True, help="comma-separated song ids to remove")
|
|
p.set_defaults(func=cmd_playlist_remove)
|
|
|
|
p = pl_sub.add_parser("audit")
|
|
p.add_argument("id")
|
|
p.add_argument("--min-size", type=int, default=8, help="only show (artist,album) groups >= this size")
|
|
p.set_defaults(func=cmd_playlist_audit)
|
|
|
|
args = parser.parse_args()
|
|
try:
|
|
args.func(args)
|
|
except Exception as e:
|
|
print(f"error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|