diff --git a/subwave/recreate_broadcast.py b/subwave/recreate_broadcast.py new file mode 100755 index 0000000..d367654 --- /dev/null +++ b/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()