subwave: add recreate_broadcast.py to build historical-chart shows

Given a hand-transcribed {station, date, tracks} JSON file (chart data is
sourced manually in conversation, never scraped in code -- ARSA's robots.txt
disallows automated fetching of its survey pages), matches tracks against the
Navidrome library, builds/updates a Navidrome playlist from the matches, and
wires it into a SUB/WAVE show with playlistStrict so it plays only that chart.

Claude-Session: https://claude.ai/code/session_01L7Rwa6guD5wK8F8tWQwcJX

subwave/recreate_broadcast.py

@@ -0,0 +1,366 @@
+#!/usr/bin/env python3
+"""
+Recreate a historical radio broadcast (or a national chart for a given week)
+as a SUB/WAVE show, from a hand-transcribed track list.
+
+Input is a small JSON file:
+
+    {
+      "station": "WABC",
+      "date": "1968-10-19",
+      "tracks": [
+        {"rank": 1, "artist": "The Beatles", "title": "Hey Jude"},
+        {"rank": 2, "artist": "O.C. Smith", "title": "Little Green Apples"}
+      ]
+    }
+
+"station" is really just a chart label -- a call sign, or something like
+"Billboard Modern Rock Tracks". This script never fetches anything from any
+chart/survey web site itself; it only ever consumes this structured input,
+which you transcribe by hand.
+
+Each track is matched against the Navidrome library, matched tracks become a
+Navidrome playlist, and that playlist is wired into a SUB/WAVE show
+(playlistStrict=true, so the show plays only these tracks).
+
+Usage:
+    subwave/recreate_broadcast.py subwave/broadcasts/wabc-1968-10-19.json
+    subwave/recreate_broadcast.py subwave/broadcasts/wabc-1968-10-19.json --commit -y
+    subwave/recreate_broadcast.py subwave/broadcasts/wabc-1968-10-19.json --commit -y --play-now 60
+"""
+
+import argparse
+import difflib
+import hashlib
+import json
+import re
+import secrets
+import subprocess
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+from pathlib import Path
+
+SCRIPT_DIR = Path(__file__).resolve().parent
+DEFAULT_SETUP_CONFIG = "/srv/subwave/state/setup-config.json"
+DEFAULT_ENV_FILE = SCRIPT_DIR / ".env"
+DEFAULT_CONTROLLER_CONTAINER = "sub-wave-controller"
+SUBSONIC_CLIENT = "subwave-recreate-broadcast"
+SUBSONIC_VERSION = "1.16.1"
+
+STOPWORDS = {"the", "a", "an", "and", "of", "feat", "featuring", "ft"}
+
+
+# --------------------------------------------------------------------------
+# Connection resolution
+# --------------------------------------------------------------------------
+
+def parse_env_file(path):
+    values = {}
+    if not path.exists():
+        return values
+    for line in path.read_text().splitlines():
+        line = line.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+        key, _, val = line.partition("=")
+        values[key.strip()] = val.strip()
+    return values
+
+
+def load_navidrome_config(args):
+    if args.navidrome_url and args.navidrome_user and args.navidrome_pass:
+        return {"url": args.navidrome_url.rstrip("/"), "user": args.navidrome_user, "pass": args.navidrome_pass}
+    cfg = json.loads(Path(args.setup_config).read_text())["navidrome"]
+    return {"url": cfg["url"].rstrip("/"), "user": cfg["user"], "pass": cfg["pass"]}
+
+
+def load_admin_config(args):
+    user, pw = args.admin_user, args.admin_pass
+    if not (user and pw):
+        env = parse_env_file(DEFAULT_ENV_FILE)
+        user = user or env.get("ADMIN_USER", "")
+        pw = pw or env.get("ADMIN_PASS", "")
+    return user, pw
+
+
+# --------------------------------------------------------------------------
+# Subsonic (Navidrome) client
+# --------------------------------------------------------------------------
+
+def subsonic_call(nd_cfg, endpoint, params=None, retry_local=True):
+    salt = secrets.token_hex(8)
+    token = hashlib.md5((nd_cfg["pass"] + salt).encode("utf-8")).hexdigest()
+    q = {"u": nd_cfg["user"], "t": token, "s": salt, "v": SUBSONIC_VERSION, "c": SUBSONIC_CLIENT, "f": "json"}
+    q.update(params or {})
+    url = f"{nd_cfg['url']}/rest/{endpoint}?{urllib.parse.urlencode(q, doseq=True)}"
+    try:
+        with urllib.request.urlopen(url, timeout=10) as r:
+            data = json.load(r)
+    except (urllib.error.URLError, ConnectionError, TimeoutError) as e:
+        if retry_local and "navidrome:" in nd_cfg["url"]:
+            fallback = dict(nd_cfg, url=nd_cfg["url"].replace("navidrome:", "localhost:"))
+            return subsonic_call(fallback, endpoint, params, retry_local=False)
+        raise RuntimeError(f"could not reach Navidrome at {nd_cfg['url']}: {e}") from e
+    sub = data["subsonic-response"]
+    if sub.get("status") != "ok":
+        raise RuntimeError(f"Subsonic {endpoint} failed: {sub.get('error')}")
+    return sub
+
+
+def search3(nd_cfg, query, song_count=10):
+    if not query.strip():
+        return []
+    sub = subsonic_call(nd_cfg, "search3", {"query": query, "songCount": song_count, "artistCount": 0, "albumCount": 0})
+    return sub.get("searchResult3", {}).get("song", [])
+
+
+def get_playlists(nd_cfg):
+    return subsonic_call(nd_cfg, "getPlaylists").get("playlists", {}).get("playlist", []) or []
+
+
+def create_playlist(nd_cfg, name, song_ids, playlist_id=None):
+    """Create a playlist, or (if playlist_id given) fully replace its contents."""
+    params = {"name": name}
+    if playlist_id:
+        params["playlistId"] = playlist_id
+    first_chunk, remaining = song_ids[:100], song_ids[100:]
+    params["songId"] = first_chunk
+    sub = subsonic_call(nd_cfg, "createPlaylist", params)
+    new_id = sub["playlist"]["id"]
+    while remaining:
+        chunk, remaining = remaining[:100], remaining[100:]
+        subsonic_call(nd_cfg, "updatePlaylist", {"playlistId": new_id, "songIdToAdd": chunk})
+    subsonic_call(nd_cfg, "updatePlaylist", {"playlistId": new_id, "public": "true"})
+    return new_id
+
+
+# --------------------------------------------------------------------------
+# Matching
+# --------------------------------------------------------------------------
+
+def normalize(s):
+    s = re.sub(r"[^\w\s]", " ", s.lower())
+    return re.sub(r"\s+", " ", s).strip()
+
+
+def strip_parens(s):
+    return re.sub(r"\s*[\(\[][^\)\]]*[\)\]]", "", s).strip()
+
+
+def strip_stopwords(s):
+    words = [w for w in normalize(s).split() if w not in STOPWORDS]
+    return " ".join(words) if words else normalize(s)
+
+
+def similarity(a, b):
+    return difflib.SequenceMatcher(None, normalize(a), normalize(b)).ratio()
+
+
+def score_candidate(song, artist, title):
+    return similarity(song.get("title", ""), title) * 0.6 + similarity(song.get("artist", ""), artist) * 0.4
+
+
+def match_track(nd_cfg, artist, title, min_score):
+    attempts = [
+        ("combined", f"{artist} {title}", 10),
+        ("title-only", title, 20),
+        ("keyword", f"{strip_stopwords(artist)} {strip_stopwords(strip_parens(title))}", 20),
+    ]
+    for stage, query, count in attempts:
+        candidates = search3(nd_cfg, query, song_count=count)
+        if not candidates:
+            continue
+        best_score, best_song = max(((score_candidate(c, artist, title), c) for c in candidates), key=lambda x: x[0])
+        if best_score >= min_score:
+            status = "matched" if stage == "combined" and best_score >= 0.92 else "low-confidence"
+            return {"status": status, "score": best_score, "stage": stage, "song": best_song}
+    return {"status": "unmatched", "score": 0.0, "stage": None, "song": None}
+
+
+# --------------------------------------------------------------------------
+# Report
+# --------------------------------------------------------------------------
+
+STATUS_LABEL = {"matched": "MATCH", "low-confidence": "LOW  ", "unmatched": "MISS "}
+
+
+def print_report(tracks, results):
+    for track, result in zip(tracks, results):
+        rank = track.get("rank", "-")
+        src = f"{track['artist']} — {track['title']}"
+        if result["song"]:
+            s = result["song"]
+            dst = f"nd:{s['id'][:8]} \"{s.get('title', '?')}\" by {s.get('artist', '?')} ({s.get('album', '?')})"
+        else:
+            dst = "(no match found)"
+        print(f"#{str(rank):>3} [{STATUS_LABEL[result['status']]}] {src:<50} -> {dst}")
+
+    matched = sum(1 for r in results if r["status"] == "matched")
+    low = sum(1 for r in results if r["status"] == "low-confidence")
+    missed = sum(1 for r in results if r["status"] == "unmatched")
+    print(f"\n{matched + low}/{len(results)} matched ({low} low-confidence, {missed} unmatched)")
+    return matched, low, missed
+
+
+# --------------------------------------------------------------------------
+# SUB/WAVE admin API (via docker exec -- the controller's port isn't
+# published to the host)
+# --------------------------------------------------------------------------
+
+def admin_request(container, admin_user, admin_pass, method, path, body=None):
+    cmd = ["docker", "exec", container, "curl", "-s", "-X", method]
+    if admin_user and admin_pass:
+        cmd += ["-u", f"{admin_user}:{admin_pass}"]
+    if body is not None:
+        cmd += ["-H", "Content-Type: application/json", "--data", json.dumps(body)]
+    cmd += [f"http://localhost:7701{path}"]
+    result = subprocess.run(cmd, capture_output=True, text=True, timeout=20)
+    if result.returncode != 0:
+        raise RuntimeError(f"docker exec into {container} failed: {result.stderr.strip()}")
+    try:
+        parsed = json.loads(result.stdout)
+    except json.JSONDecodeError:
+        raise RuntimeError(f"unexpected response from SUB/WAVE controller: {result.stdout[:300]!r}")
+    if isinstance(parsed, dict) and "error" in parsed:
+        raise RuntimeError(f"SUB/WAVE controller rejected {method} {path}: {parsed['error']}")
+    return parsed
+
+
+def find_existing_show(settings, name):
+    for show in settings.get("shows", []) or []:
+        if show.get("name", "").strip().lower() == name.strip().lower():
+            return show
+    return None
+
+
+def resolve_persona_id(settings, persona_id, persona_name):
+    personas = settings.get("personas", []) or []
+    if persona_id:
+        if any(p["id"] == persona_id for p in personas):
+            return persona_id
+        raise RuntimeError(f"no persona with id {persona_id!r} in the roster")
+    if persona_name:
+        for p in personas:
+            if p.get("name", "").strip().lower() == persona_name.strip().lower():
+                return p["id"]
+        raise RuntimeError(f"no persona named {persona_name!r} in the roster")
+    if settings.get("activePersonaId"):
+        return settings["activePersonaId"]
+    if personas:
+        return personas[0]["id"]
+    raise RuntimeError("no personas found in the SUB/WAVE roster")
+
+
+# --------------------------------------------------------------------------
+# Orchestration
+# --------------------------------------------------------------------------
+
+def load_input(path):
+    data = json.loads(Path(path).read_text())
+    tracks = data.get("tracks") or []
+    if not tracks:
+        raise RuntimeError("input file has no tracks")
+    return data["station"], data["date"], tracks
+
+
+def run(args):
+    station, date, tracks = load_input(args.input)
+    show_name = f"{station} — {date}"
+
+    if args.limit:
+        tracks = tracks[: args.limit]
+
+    nd_cfg = load_navidrome_config(args)
+
+    print(f'Matching {len(tracks)} tracks for "{show_name}" against the Navidrome library...\n')
+    results = [match_track(nd_cfg, t["artist"], t["title"], args.min_match_score) for t in tracks]
+    matched, low, missed = print_report(tracks, results)
+    total_usable = matched + low
+
+    if total_usable < args.min_tracks and not args.force:
+        print(f"\nOnly {total_usable} usable match(es) (minimum {args.min_tracks}) — refusing to continue. Pass --force to override.")
+        sys.exit(1)
+
+    if not args.commit:
+        print("\nDry run only — pass --commit to create/update the playlist and show.")
+        return
+
+    if not args.yes:
+        reply = input(f'\nCreate/update "{show_name}" from {total_usable} matched track(s)? [y/N] ')
+        if reply.strip().lower() not in ("y", "yes"):
+            print("Aborted.")
+            return
+
+    song_ids = [r["song"]["id"] for r in results if r["song"]]
+
+    existing_playlist = next((p for p in get_playlists(nd_cfg) if p.get("name") == show_name), None)
+    playlist_id = create_playlist(nd_cfg, show_name, song_ids, playlist_id=existing_playlist["id"] if existing_playlist else None)
+    print(f'{"Updated" if existing_playlist else "Created"} Navidrome playlist "{show_name}" (id {playlist_id}) with {len(song_ids)} track(s).')
+
+    admin_user, admin_pass = load_admin_config(args)
+    settings = admin_request(args.controller_container, admin_user, admin_pass, "GET", "/settings").get("values", {})
+    persona_id = resolve_persona_id(settings, args.persona_id, args.persona_name)
+    existing_show = find_existing_show(settings, show_name)
+
+    show_body = {
+        "name": show_name,
+        "topic": f"A recreation of the {station} chart from {date}.",
+        "personaId": persona_id,
+        "guestPersonaIds": [],
+        "playlistIds": [playlist_id],
+        "playlistStrict": True,
+        "filtersStrict": True,
+        "moods": [], "genres": [], "energies": [], "eras": [],
+        "excludedPlaylistIds": [],
+        "maxTrackSeconds": 0,
+    }
+    if existing_show:
+        show_body["id"] = existing_show["id"]
+
+    resp = admin_request(args.controller_container, admin_user, admin_pass, "POST", "/shows", {"show": show_body})
+    show = resp["show"]
+    print(f'{"Updated" if existing_show else "Created"} SUB/WAVE show "{show["name"]}" (id {show["id"]}), hosted by persona {persona_id}.')
+
+    if args.play_now:
+        admin_request(
+            args.controller_container, admin_user, admin_pass, "POST", "/schedule/override",
+            {"showId": show["id"], "minutes": args.play_now, "until": "fixed"},
+        )
+        print(f'Takeover started: "{show_name}" airing now for {args.play_now} minute(s).')
+
+
+def parse_args():
+    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    p.add_argument("input", help="path to a JSON file: {station, date, tracks: [{rank, artist, title}]}")
+    p.add_argument("--commit", action="store_true", help="create/update the playlist and show (default is dry-run)")
+    p.add_argument("-y", "--yes", action="store_true", help="skip the confirmation prompt when --commit is passed")
+    p.add_argument("--play-now", type=int, metavar="MINUTES", help="also start a live takeover for N minutes after creating the show (SUB/WAVE requires 15-720)")
+    p.add_argument("--persona-id", help="SUB/WAVE persona id to host the show")
+    p.add_argument("--persona-name", help="SUB/WAVE persona name to host the show (alternative to --persona-id)")
+    p.add_argument("--navidrome-url")
+    p.add_argument("--navidrome-user")
+    p.add_argument("--navidrome-pass")
+    p.add_argument("--setup-config", default=DEFAULT_SETUP_CONFIG, help=f"default: {DEFAULT_SETUP_CONFIG}")
+    p.add_argument("--admin-user")
+    p.add_argument("--admin-pass")
+    p.add_argument("--controller-container", default=DEFAULT_CONTROLLER_CONTAINER)
+    p.add_argument("--min-match-score", type=float, default=0.80)
+    p.add_argument("--min-tracks", type=int, default=3, help="minimum usable matches required to proceed (default: 3)")
+    p.add_argument("--force", action="store_true", help="proceed even if fewer than --min-tracks matched")
+    p.add_argument("--limit", type=int, help="only process the first N tracks (quick smoke test)")
+    return p.parse_args()
+
+
+def main():
+    args = parse_args()
+    try:
+        run(args)
+    except Exception as e:
+        print(f"error: {e}", file=sys.stderr)
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()