docker-infrastructure/subwave/recreate_broadcast.py

#!/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 contains_normalized(a, b):
    """True if the normalized forms of a/b contain one another as whole words.

    Handles library tags that carry extra baggage a chart title/artist won't --
    reissue composite credits ("Green Day • Billie Joe Armstrong, Mike Dirnt,
    & Tre Cool") or a subtitle the library dropped ("Sour Times" vs "Sour Times
    (Nobody Loves Me)") -- without this a correct match can score just under
    the threshold on artist or title alone even though one string is clearly
    the other plus extra text.
    """
    na, nb = f" {normalize(a)} ", f" {normalize(b)} "
    return bool(na.strip()) and bool(nb.strip()) and (na in nb or nb in na)


def score_candidate(song, artist, title):
    title_score = similarity(song.get("title", ""), title)
    if contains_normalized(song.get("title", ""), title):
        title_score = max(title_score, 0.9)
    artist_score = similarity(song.get("artist", ""), artist)
    if contains_normalized(song.get("artist", ""), artist):
        artist_score = 1.0
    return title_score * 0.6 + artist_score * 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()