Archive dead/manual scripts to pipeline/lib/manual (R8)

Move six scripts that are not wired into the app, scheduler, or web UI out of
the active pipeline/lib into pipeline/lib/manual, so the scheduled scripts are
easy to see and these stay available for hands-on use: convert-m4a.sh,
fix-empty-album.sh, backfill-spotify-tags.sh, backfill-buy-url.py,
recover-azuracast-playlists.py, import-dj-collection.py. Adds a README
explaining each, and extends the Dockerfile chmod to cover the new folder.
Nothing referenced these from active code (verified), so no wiring changes.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

pipeline/lib/manual/README.md

@@ -0,0 +1,31 @@
+# Manual / archived scripts
+
+These scripts are kept for occasional hands-on use but are **not** wired into
+the app, the scheduler, or the web UI. Nothing here runs automatically. They
+live outside `pipeline/lib/` so the active, scheduled scripts there stay easy
+to see at a glance.
+
+Run one by hand inside the container when you actually need it, e.g.:
+
+```bash
+docker exec -it alembic bash /app/pipeline/lib/manual/convert-m4a.sh
+docker exec -it alembic python /app/pipeline/lib/manual/import-dj-collection.py --help
+```
+
+What's here and why it's manual-only:
+
+- **convert-m4a.sh** — one-off bulk M4A to FLAC/MP3 conversion. Not part of the
+  normal download flow.
+- **fix-empty-album.sh** — repairs tracks with an empty album tag. A cleanup
+  tool for a specific past problem, not a recurring job.
+- **backfill-spotify-tags.sh** — re-pulls Spotify tags for already-imported
+  tracks. A migration/backfill aid.
+- **backfill-buy-url.py** — one-time backfill of buy-link tags across the
+  library (the scheduled `enrich-buy-url.py` handles new tracks going forward).
+- **recover-azuracast-playlists.py** — rebuilds AzuraCast playlist assignments
+  after a GROUPING-tag loss. Recovery tool, only relevant if you run AzuraCast.
+- **import-dj-collection.py** — bulk-imports a personal DJ set collection with
+  fingerprint dedup against the existing library. A large one-off importer.
+
+These have not all been hardened with `set -euo pipefail`; treat them as
+run-and-watch tools rather than unattended automation.

pipeline/lib/manual/backfill-buy-url.py

@@ -0,0 +1,215 @@
+#!/usr/bin/env python3
+"""
+backfill-buy-url.py — stamp the Bandcamp purchase URL onto library files for
+albums/tracks already imported from a Bandcamp purchase.
+
+The forward path (sync-bandcamp.py) tags new purchases at download time. This
+one-shot covers everything bought *before* that change landed. It:
+
+  1. Pulls the current Bandcamp collection (cookies auth) and builds an index
+     keyed by (album_artist, album) -> release URL. (Bandcamp's embedded ALBUM
+     tag matches the collection item_title exactly, and ALBUMARTIST matches the
+     band name — verified against the library.)
+  2. Walks the FLAC library, reads ALBUMARTIST/ALBUM, and where a match exists,
+     writes the COMMERCIAL_INFORMATION Vorbis comment (AzuraCast maps this to
+     the buy_url custom field — see sync-bandcamp.py BUY_URL_TAG).
+  3. Optionally tells AzuraCast to reprocess each touched file so the custom
+     field repopulates without waiting for the periodic media scan.
+
+Dry-run by default. Stdlib only (mutagen NOT required — uses metaflac).
+
+Usage:
+  backfill-buy-url.py --cookies F --state F [--library DIR] [--apply]
+                      [--azuracast-key ID:SECRET] [--azuracast-base URL]
+                      [--station N] [--force]
+"""
+import sys, os, re, json, time, html, argparse, subprocess
+import urllib.request
+from http.cookiejar import MozillaCookieJar
+from pathlib import Path
+
+COLLECTION_ITEMS_URL = "https://bandcamp.com/api/fancollection/1/collection_items"
+UA = "Mozilla/5.0 (X11; Linux x86_64) backfill-buy-url/1.0"
+BUY_URL_TAG = "COMMERCIAL_INFORMATION"
+
+
+# ---- Bandcamp collection ---------------------------------------------------
+def opener(jar):
+    return urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
+
+
+def get_fan_id(username, jar):
+    req = urllib.request.Request(f"https://bandcamp.com/{username}",
+                                 headers={"User-Agent": UA})
+    page = html.unescape(opener(jar).open(req, timeout=60).read().decode("utf-8", "replace"))
+    m = re.search(r'"fan_id"\s*:\s*"?(\d+)', page)
+    if not m:
+        sys.exit("[backfill] could not find fan_id (cookies expired / wrong user?)")
+    return int(m.group(1))
+
+
+def fetch_collection(fan_id, jar):
+    items = []
+    tok = "9999999999::a::"
+    while True:
+        body = json.dumps({"fan_id": fan_id, "older_than_token": tok, "count": 100}).encode()
+        req = urllib.request.Request(
+            COLLECTION_ITEMS_URL, data=body,
+            headers={"User-Agent": UA, "Content-Type": "application/json",
+                     "Accept": "application/json", "Referer": "https://bandcamp.com/"})
+        data = json.loads(opener(jar).open(req, timeout=60).read())
+        page = data.get("items", [])
+        items.extend(page)
+        if not data.get("more_available") or not page:
+            break
+        nt = data.get("last_token")
+        if not nt or nt == tok:
+            break
+        tok = nt
+    return items
+
+
+# ---- matching --------------------------------------------------------------
+def norm(s):
+    return re.sub(r"\s+", " ", (s or "").strip()).casefold()
+
+
+def build_index(items):
+    """(album_artist, album) -> url. Index both the bare title and the
+    "<title> - Single" form (single-track purchases get that ALBUM tag)."""
+    idx = {}
+    for it in items:
+        url = it.get("item_url")
+        band = it.get("band_name")
+        title = it.get("item_title")
+        if not (url and band and title):
+            continue
+        for album in (title, f"{title} - Single"):
+            idx.setdefault((norm(band), norm(album)), url)
+    return idx
+
+
+def flac_tag(path, name):
+    r = subprocess.run(["metaflac", f"--show-tag={name}", path],
+                       capture_output=True, text=True)
+    for line in r.stdout.splitlines():
+        if "=" in line:
+            return line.split("=", 1)[1]
+    return ""
+
+
+def set_flac_tag(path, name, value):
+    subprocess.run(["metaflac", f"--remove-tag={name}", path], capture_output=True)
+    return subprocess.run(["metaflac", f"--set-tag={name}={value}", path],
+                          capture_output=True).returncode == 0
+
+
+# ---- AzuraCast reprocess ---------------------------------------------------
+def az_reprocess(base, key, station, host_paths, log):
+    """Find each host path in AzuraCast and batch-reprocess so the custom
+    field repopulates. host_path -> relative AzuraCast path is library-root
+    stripped."""
+    root = f"{os.environ.get('MUSIC_DATA_DIR', '/data/music')}/Library/"
+    want = {p[len(root):] for p in host_paths if p.startswith(root)}
+    headers = {"X-API-Key": key, "Accept": "application/json"}
+
+    def get(path):
+        req = urllib.request.Request(base + path, headers=headers)
+        return json.loads(urllib.request.urlopen(req, timeout=60).read())
+
+    unique_ids, page = [], 1
+    while True:
+        d = get(f"/api/station/{station}/files?per_page=500&page={page}")
+        for row in d["rows"]:
+            if row["path"] in want:
+                unique_ids.append(row["unique_id"])
+        if page >= d["total_pages"]:
+            break
+        page += 1
+    if not unique_ids:
+        log("  no matching AzuraCast files to reprocess")
+        return
+    body = json.dumps({"do": "reprocess", "files": unique_ids}).encode()
+    req = urllib.request.Request(base + f"/api/station/{station}/files/batch",
+                                 data=body, method="PUT",
+                                 headers={**headers, "Content-Type": "application/json"})
+    urllib.request.urlopen(req, timeout=120).read()
+    log(f"  queued reprocess for {len(unique_ids)} AzuraCast file(s)")
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--cookies", required=True)
+    ap.add_argument("--state", required=True, help="state.json (for reference/logging)")
+    ap.add_argument("--library", default=f"{os.environ.get('MUSIC_DATA_DIR', '/data/music')}/Library")
+    ap.add_argument("--apply", action="store_true")
+    ap.add_argument("--force", action="store_true",
+                    help="overwrite an existing buy-link tag")
+    ap.add_argument("--azuracast-key", default=None)
+    ap.add_argument("--azuracast-base", default=os.environ.get("AZURACAST_BASE", ""))
+    ap.add_argument("--station", type=int, default=1)
+    ap.add_argument("--username", default=None,
+                    help="Bandcamp username; default read from config.env if present")
+    args = ap.parse_args()
+
+    user = args.username
+    if not user:
+        cfg = f"{os.environ.get('ALEMBIC_CONFIG_DIR', '/config')}/pipeline/bandcamp/config.env"
+        if os.path.exists(cfg):
+            for line in Path(cfg).read_text().splitlines():
+                if line.startswith("BANDCAMP_USERNAME="):
+                    user = line.split("=", 1)[1].strip().strip('"')
+    if not user:
+        sys.exit("[backfill] --username required (no config.env BANDCAMP_USERNAME)")
+
+    print(f"[backfill] mode={'APPLY' if args.apply else 'DRY RUN'}  user={user!r}")
+    jar = MozillaCookieJar(args.cookies)
+    jar.load(ignore_discard=True, ignore_expires=True)
+    fan_id = get_fan_id(user, jar)
+    items = fetch_collection(fan_id, jar)
+    idx = build_index(items)
+    print(f"[backfill] collection: {len(items)} items, {len(idx)} match keys")
+
+    flacs = list(Path(args.library).rglob("*.flac"))
+    print(f"[backfill] scanning {len(flacs)} FLAC files...\n")
+
+    matched = tagged = already = unmatched = 0
+    touched = []
+    for p in flacs:
+        sp = str(p)
+        albumartist = flac_tag(sp, "ALBUMARTIST") or flac_tag(sp, "ARTIST")
+        album = flac_tag(sp, "ALBUM")
+        url = idx.get((norm(albumartist), norm(album)))
+        if not url:
+            unmatched += 1
+            continue
+        matched += 1
+        existing = flac_tag(sp, BUY_URL_TAG)
+        if existing and not args.force:
+            already += 1
+            continue
+        rel = sp[len(args.library):].lstrip("/")
+        print(f"  + {rel}\n      -> {url}")
+        if args.apply:
+            if set_flac_tag(sp, BUY_URL_TAG, url):
+                tagged += 1
+                touched.append(sp)
+            else:
+                print("      ! metaflac write failed")
+        else:
+            tagged += 1
+            touched.append(sp)
+
+    print(f"\n[backfill] matched={matched}  {'tagged' if args.apply else 'would-tag'}={tagged}"
+          f"  already-had={already}  unmatched-files={unmatched}")
+
+    if args.apply and touched and args.azuracast_key:
+        print("[backfill] telling AzuraCast to reprocess touched files...")
+        az_reprocess(args.azuracast_base, args.azuracast_key, args.station,
+                     touched, lambda m: print(m))
+    elif not args.apply:
+        print("[backfill] re-run with --apply to write (add --azuracast-key to reprocess)")
+
+
+if __name__ == "__main__":
+    main()

pipeline/lib/manual/backfill-spotify-tags.sh

@@ -0,0 +1,77 @@
+#!/bin/bash
+# One-shot backfill: rewrite tags from Spotify for every track currently in the
+# beets library that has a grouping:<playlist> tag matching one of our configs.
+#
+# After the rewrite, sync beets DB from file tags, then move files to clean
+# paths based on the new tags, then trigger a Navidrome rescan.
+
+set -u
+
+CONFIGS="${ALEMBIC_CONFIG_DIR:-/config}/pipeline"
+LOG="${ALEMBIC_CONFIG_DIR:-/config}/logs/backfill-$(date +%Y%m%d-%H%M%S).log"
+RETAG="${PIPELINE_DIR:-/app/pipeline}/lib/spotify-retag.py"
+
+mkdir -p "$(dirname "$LOG")"
+echo "[$(date -Iseconds)] === Backfill start ===" | tee -a "$LOG"
+
+TOTAL_TRACKS=0
+TOTAL_PLAYLISTS=0
+
+for conf in "$CONFIGS"/*.conf; do
+  [[ "$conf" == "$CONFIGS/_template.conf" ]] && continue
+  playlist=$(basename "$conf" .conf)
+
+  url=$(sed -n 's/^input *= *//p' "$conf" | tr -d ' ')
+  cid=$(sed -n 's/^spotify-id *= *//p' "$conf" | tr -d ' ')
+  csec=$(sed -n 's/^spotify-secret *= *//p' "$conf" | tr -d ' ')
+
+  if [[ -z "$url" || -z "$cid" || -z "$csec" ]]; then
+    echo "[$playlist] SKIP: missing url/creds in conf" | tee -a "$LOG"
+    continue
+  fi
+
+  # beets runs in-process in this same container, so its reported paths
+  # need no container-to-host translation.
+  paths=$(beet ls -f '$path' "grouping:$playlist" 2>/dev/null)
+  count=$(echo -n "$paths" | grep -c '^' || true)
+
+  if [[ "$count" -eq 0 ]]; then
+    echo "[$playlist] no tracks tagged in beets — skip" | tee -a "$LOG"
+    continue
+  fi
+
+  echo "[$playlist] retagging $count tracks via Spotify" | tee -a "$LOG"
+  if echo "$paths" | SPOTIFY_CLIENT_ID="$cid" SPOTIFY_CLIENT_SECRET="$csec" \
+        python3 "$RETAG" "$url" - >> "$LOG" 2>&1; then
+    echo "[$playlist] retag OK" | tee -a "$LOG"
+    TOTAL_PLAYLISTS=$((TOTAL_PLAYLISTS + 1))
+    TOTAL_TRACKS=$((TOTAL_TRACKS + count))
+  else
+    echo "[$playlist] retag FAILED — see $LOG" | tee -a "$LOG"
+  fi
+done
+
+echo "[$(date -Iseconds)] === Retagged $TOTAL_TRACKS tracks across $TOTAL_PLAYLISTS playlists ===" | tee -a "$LOG"
+
+echo "[$(date -Iseconds)] Syncing beets DB from file tags (beet update)..." | tee -a "$LOG"
+beet update -F path 2>&1 | tail -20 >> "$LOG"
+
+echo "[$(date -Iseconds)] Moving files into clean paths (beet move)..." | tee -a "$LOG"
+beet move 2>&1 | tail -5 >> "$LOG"
+
+NAVIDROME_ENV="${ALEMBIC_CONFIG_DIR:-/config}/pipeline/navidrome/admin.env"
+[ -f "$NAVIDROME_ENV" ] && source "$NAVIDROME_ENV"
+ND_BASE="${ND_BASE:-http://navidrome:4533}"
+ND_USER="${ND_USER:-}"
+ND_PASS="${ND_PASS:-}"
+
+echo "[$(date -Iseconds)] Triggering Navidrome full scan" | tee -a "$LOG"
+curl -s -G "$ND_BASE/rest/startScan.view" \
+  --data-urlencode "u=$ND_USER" \
+  --data-urlencode "p=$ND_PASS" \
+  --data-urlencode 'v=1.16.0' \
+  --data-urlencode 'c=backfill' \
+  --data-urlencode 'f=json' \
+  --data-urlencode 'fullScan=true' >> "$LOG"
+
+echo "[$(date -Iseconds)] === Backfill complete. Log: $LOG ===" | tee -a "$LOG"

pipeline/lib/manual/convert-m4a.sh

@@ -0,0 +1,57 @@
+#!/bin/bash
+# Convert m4a files in-place: AAC -> MP3 (V0 VBR), ALAC -> FLAC (lossless).
+# Preserves metadata + embedded cover art. Deletes original on success.
+set -u
+
+LOG=${ALEMBIC_CONFIG_DIR:-/config}/logs/convert-m4a-$(date +%Y%m%d-%H%M%S).log
+mkdir -p "$(dirname "$LOG")"
+
+log() { echo "[$(date -Iseconds)] $*" | tee -a "$LOG"; }
+
+OK_AAC=0
+OK_ALAC=0
+FAIL=0
+
+while IFS= read -r -d '' f; do
+  codec=$(ffprobe -v error -select_streams a:0 -show_entries stream=codec_name -of csv=p=0 "$f" 2>/dev/null | tr -d ',')
+  base="${f%.m4a}"
+  case "$codec" in
+    aac)
+      out="${base}.mp3"
+      if [[ -e "$out" ]]; then log "SKIP-EXISTS: $out"; FAIL=$((FAIL+1)); continue; fi
+      if ffmpeg -nostdin -hide_banner -loglevel error -i "$f" \
+           -map 0 -c:v copy -c:a libmp3lame -q:a 0 -id3v2_version 3 \
+           "$out" 2>>"$LOG"; then
+        rm -f "$f"
+        log "AAC→MP3:    $out"
+        OK_AAC=$((OK_AAC+1))
+      else
+        log "FAIL-AAC:   $f"
+        rm -f "$out" 2>/dev/null
+        FAIL=$((FAIL+1))
+      fi
+      ;;
+    alac)
+      out="${base}.flac"
+      if [[ -e "$out" ]]; then log "SKIP-EXISTS: $out"; FAIL=$((FAIL+1)); continue; fi
+      if ffmpeg -nostdin -hide_banner -loglevel error -i "$f" \
+           -map 0 -c:v copy -c:a flac -compression_level 8 \
+           "$out" 2>>"$LOG"; then
+        rm -f "$f"
+        log "ALAC→FLAC:  $out"
+        OK_ALAC=$((OK_ALAC+1))
+      else
+        log "FAIL-ALAC:  $f"
+        rm -f "$out" 2>/dev/null
+        FAIL=$((FAIL+1))
+      fi
+      ;;
+    *)
+      log "SKIP-OTHER ($codec): $f"
+      ;;
+  esac
+done < <(find ${MUSIC_DATA_DIR:-/data/music}/Library -type f -iname "*.m4a" -print0 2>/dev/null)
+
+log ""
+log "=== AAC→MP3:  $OK_AAC | ALAC→FLAC: $OK_ALAC | Failed: $FAIL ==="
+log "Log: $LOG"

pipeline/lib/manual/fix-empty-album.sh

@@ -0,0 +1,109 @@
+#!/bin/bash
+# Find audio files in the library that have no ALBUM tag, set ALBUM to
+# "{TITLE} - Single" — matches Spotify / Apple Music's convention for
+# single-track releases. Otherwise these tracks land as [Unknown Album]
+# in Navidrome.
+#
+# Usage:
+#   fix-empty-album.sh                # dry run (default)
+#   fix-empty-album.sh --apply        # actually rewrite tags
+#
+# Bandcamp single-track downloads are the most common source of this
+# issue (Bandcamp serves single tracks with no ALBUM tag). New downloads
+# via sync-bandcamp.py are auto-tagged; this script handles backfill.
+
+set -u
+APPLY=0
+[[ "${1:-}" == "--apply" ]] && APPLY=1
+
+LOG=${ALEMBIC_CONFIG_DIR:-/config}/logs/fix-empty-album-$(date +%Y%m%d-%H%M%S).log
+mkdir -p "$(dirname "$LOG")"
+
+NAVIDROME_ENV="${ALEMBIC_CONFIG_DIR:-/config}/pipeline/navidrome/admin.env"
+[ -f "$NAVIDROME_ENV" ] && source "$NAVIDROME_ENV"
+ND_BASE="${ND_BASE:-http://navidrome:4533}"
+ND_USER="${ND_USER:-}"
+ND_PASS="${ND_PASS:-}"
+log() { echo "[$(date -Iseconds)] $*" | tee -a "$LOG"; }
+
+if [[ $APPLY -eq 1 ]]; then
+  log "=== FIX EMPTY ALBUM (APPLY) ==="
+else
+  log "=== FIX EMPTY ALBUM (DRY RUN — pass --apply to write) ==="
+fi
+
+FIXED=0
+SKIPPED=0
+
+while IFS= read -r -d '' f; do
+  ext="${f,,}"; ext="${ext##*.}"
+  album="" ; title=""
+  case "$ext" in
+    flac)
+      album=$(metaflac --show-tag=ALBUM "$f" 2>/dev/null | sed -n 's/^[^=]*=//p' | head -1)
+      title=$(metaflac --show-tag=TITLE "$f" 2>/dev/null | sed -n 's/^[^=]*=//p' | head -1)
+      ;;
+    mp3)
+      album=$(id3v2 -l "$f" 2>/dev/null | sed -n 's/^TALB[^:]*: //p' | head -1)
+      title=$(id3v2 -l "$f" 2>/dev/null | sed -n 's/^TIT2[^:]*: //p' | head -1)
+      ;;
+    *) continue ;;
+  esac
+
+  if [[ -n "$album" ]]; then
+    continue   # already has an album, leave it alone
+  fi
+  if [[ -z "$title" ]]; then
+    log "[skip-no-title]  $f"
+    SKIPPED=$((SKIPPED + 1))
+    continue
+  fi
+
+  # Some titles have a leading "NN " track-number prefix baked in from a
+  # bad filename parse (e.g. title="01 Domestic Violence"). If we use the
+  # title as-is, the album becomes "01 Domestic Violence - Single" which
+  # is broken. Strip leading 1-3 digits + space, and also fix the title
+  # tag in place so it isn't shown with the prefix in clients.
+  clean_title=$(echo "$title" | sed -E 's/^[[:space:]]*[0-9]{1,3}[[:space:]]+//')
+  if [[ "$clean_title" != "$title" && -n "$clean_title" ]]; then
+    log "  trimmed leading track-number from title: \"$title\" -> \"$clean_title\""
+    if [[ $APPLY -eq 1 ]]; then
+      case "$ext" in
+        flac) metaflac --remove-tag=TITLE --set-tag="TITLE=${clean_title}" "$f" 2>>"$LOG" ;;
+        mp3)  id3v2 --TIT2 "${clean_title}" "$f" 2>>"$LOG" ;;
+      esac
+    fi
+    title="$clean_title"
+  fi
+
+  new_album="${title} - Single"
+  log "[$( [[ $APPLY -eq 1 ]] && echo SET || echo WOULD-SET )] album=\"${new_album}\"  -  $f"
+
+  if [[ $APPLY -eq 1 ]]; then
+    case "$ext" in
+      flac) metaflac --set-tag="ALBUM=${new_album}" "$f" 2>>"$LOG" ;;
+      mp3)  id3v2 --TALB "${new_album}" "$f" 2>>"$LOG" ;;
+    esac
+  fi
+  FIXED=$((FIXED + 1))
+done < <(find ${MUSIC_DATA_DIR:-/data/music}/Library -type f \( -iname "*.flac" -o -iname "*.mp3" \) -print0 2>/dev/null)
+
+log ""
+log "=== Summary: $FIXED fixed, $SKIPPED skipped (no title) ==="
+
+if [[ $APPLY -eq 1 && $FIXED -gt 0 ]]; then
+  log "Syncing beets DB from new tags..."
+  beet update 2>&1 | tail -3 >> "$LOG"
+  log "Moving files to reflect new album in path templates..."
+  beet move 2>&1 | tail -3 >> "$LOG"
+  log "Triggering Navidrome rescan..."
+  curl -s -G "$ND_BASE/rest/startScan.view" \
+    --data-urlencode "u=$ND_USER" \
+    --data-urlencode "p=$ND_PASS" \
+    --data-urlencode 'v=1.16.0' \
+    --data-urlencode 'c=fix-empty-album' \
+    --data-urlencode 'f=json' \
+    --data-urlencode 'fullScan=true' >> "$LOG" 2>&1
+fi
+
+log "=== Done. Log: $LOG ==="

pipeline/lib/manual/recover-azuracast-playlists.py

@@ -0,0 +1,315 @@
+#!/usr/bin/env python3
+"""
+recover-azuracast-playlists.py — recover playlist assignment for files that
+lost their GROUPING tag (e.g. from a pre-tag-copy upgrade-mp3-to-flac.sh run).
+
+For each AzuraCast file currently unassigned to any playlist:
+  1. Resolve the source playlist by matching (artist, title) against the
+     Spotify playlists named in $ALEMBIC_CONFIG_DIR/pipeline/*.conf.
+  2. Write GROUPING=<playlist> onto the FLAC/MP3 file.
+  3. `beet update` so beets sees the new tag.
+  4. Assign the file to the matching AzuraCast playlist via API.
+
+Default is dry-run. Pass --apply to actually write tags + assign.
+
+Env / args:
+  --azuracast-key KEY    AzuraCast API key (required)
+  --azuracast-base URL   default from AZURACAST_BASE env (empty if unset)
+  --station N            default 1
+  --apply                actually act (default is dry-run)
+  --only PLAYLIST        limit recovery to one playlist (testing)
+"""
+import argparse, base64, json, os, re, subprocess, sys, urllib.parse, urllib.request
+from difflib import SequenceMatcher
+from pathlib import Path
+
+CONF_DIR = Path(f"{os.environ.get('ALEMBIC_CONFIG_DIR', '/config')}/pipeline")
+LIB_ROOT_HOST = f"{os.environ.get('MUSIC_DATA_DIR', '/data/music')}/Library"
+# Still "/music" through the migration's Stage 0-3 transitional beets mount;
+# revisit at Stage 4 once beets' directory: becomes MUSIC_DATA_DIR/Library.
+LIB_ROOT_CONTAINER = "/music"
+
+
+def normalize(s: str) -> str:
+    s = (s or "").lower()
+    s = re.sub(r"\s*\([^)]*\)", "", s)
+    s = re.sub(r"\s*\[[^\]]*\]", "", s)
+    s = re.sub(r"\s*-?\s*(extended|original|radio|club|vip|vocal)\s*(mix|edit|version)\s*$", "", s)
+    s = re.sub(r"\s*feat\.?\s.*", "", s)
+    s = re.sub(r"\s*ft\.?\s.*", "", s)
+    s = re.sub(r"[^a-z0-9]+", "", s)
+    return s
+
+
+def read_conf(path: Path) -> dict:
+    out = {}
+    for line in path.read_text().splitlines():
+        m = re.match(r"\s*([a-z0-9_-]+)\s*=\s*(.*?)\s*$", line)
+        if m: out[m.group(1)] = m.group(2)
+    return out
+
+
+def spotify_token(cid: str, csec: str) -> str:
+    creds = base64.b64encode(f"{cid}:{csec}".encode()).decode()
+    req = urllib.request.Request(
+        "https://accounts.spotify.com/api/token",
+        data=urllib.parse.urlencode({"grant_type": "client_credentials"}).encode(),
+        headers={"Authorization": f"Basic {creds}",
+                 "Content-Type": "application/x-www-form-urlencoded"})
+    with urllib.request.urlopen(req, timeout=15) as r:
+        return json.loads(r.read())["access_token"]
+
+
+def fetch_spotify_playlist(url: str, token: str) -> list[tuple[str, str]]:
+    """Return list of (artists_normalized_joined, title_normalized)."""
+    m = re.search(r"playlist[/:]([A-Za-z0-9]+)", url)
+    if not m: return []
+    pid = m.group(1)
+    out = []
+    next_url = f"https://api.spotify.com/v1/playlists/{pid}/tracks?limit=100"
+    while next_url:
+        req = urllib.request.Request(next_url, headers={"Authorization": f"Bearer {token}"})
+        with urllib.request.urlopen(req, timeout=20) as r:
+            data = json.loads(r.read())
+        for item in data["items"]:
+            t = item.get("track")
+            if not t or t.get("is_local"): continue
+            artists = [a["name"] for a in t["artists"]]
+            out.append((artists, t["name"]))
+        next_url = data.get("next")
+    return out
+
+
+def beets_dump() -> dict:
+    """Return path → {grouping, artist, albumartist, album, title}"""
+    proc = subprocess.run(
+        ["beet", "ls", "-f",
+         "$path\t$grouping\t$artist\t$albumartist\t$album\t$title"],
+        capture_output=True, text=True, check=True)
+    out = {}
+    for line in proc.stdout.splitlines():
+        parts = line.split("\t")
+        if len(parts) < 6: continue
+        out[parts[0]] = {
+            "grouping": parts[1],
+            "artist": parts[2],
+            "albumartist": parts[3],
+            "album": parts[4],
+            "title": parts[5],
+        }
+    return out
+
+
+def az_request(base: str, key: str, path: str, method="GET", body=None) -> dict:
+    url = f"{base}{path}"
+    headers = {"X-API-Key": key, "Accept": "application/json"}
+    data = None
+    if body is not None:
+        headers["Content-Type"] = "application/json"
+        data = json.dumps(body).encode()
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=30) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def az_all_files(base: str, key: str, station: int) -> list[dict]:
+    out = []
+    page = 1
+    while True:
+        d = az_request(base, key, f"/api/station/{station}/files?per_page=500&page={page}")
+        out.extend(d["rows"])
+        if page >= d["total_pages"]: break
+        page += 1
+    return out
+
+
+def az_playlists_by_name(base: str, key: str, station: int) -> dict:
+    return {p["name"]: p["id"]
+            for p in az_request(base, key, f"/api/station/{station}/playlists")}
+
+
+def write_grouping(host_path: str, grouping: str) -> bool:
+    """Write GROUPING tag in-place. Return True on success."""
+    ext = host_path.rsplit(".", 1)[-1].lower()
+    if ext == "flac":
+        subprocess.run(["metaflac", "--remove-tag=GROUPING", host_path],
+                       capture_output=True, check=False)
+        return subprocess.run(["metaflac", f"--set-tag=GROUPING={grouping}", host_path],
+                              capture_output=True).returncode == 0
+    if ext == "mp3":
+        return subprocess.run(["id3v2", "--TIT1", grouping, host_path],
+                              capture_output=True).returncode == 0
+    return False
+
+
+def beet_update(container_paths: list[str], log) -> None:
+    """Re-read tags from disk into beets DB for the given container paths.
+    Uses `path:<exact>` query (not `path::<regex>`) and pipes "y" since
+    beet update prompts interactively for each change."""
+    if not container_paths: return
+    for cp in container_paths:
+        subprocess.run(
+            ["beet", "update", "--nomove", f"path:{cp}"],
+            input="y\n", capture_output=True, text=True, check=False)
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--azuracast-key", required=True)
+    ap.add_argument("--azuracast-base", default=os.environ.get("AZURACAST_BASE", ""))
+    ap.add_argument("--station", type=int, default=1)
+    ap.add_argument("--apply", action="store_true")
+    ap.add_argument("--only", default=None, help="limit recovery to one playlist name")
+    args = ap.parse_args()
+
+    print("[1/5] Loading beets metadata...", flush=True)
+    beets = beets_dump()
+    print(f"      {len(beets)} tracks in beets")
+
+    print("[2/5] Fetching Spotify playlist contents...", flush=True)
+    # Build (title_norm) → list of (playlist_name, artists_list)
+    title_idx: dict[str, list[tuple[str, list[str]]]] = {}
+    confs = sorted(CONF_DIR.glob("*.conf"))
+    confs = [c for c in confs if not c.name.startswith("_")]
+    if not confs:
+        print(f"ERROR: no *.conf files in {CONF_DIR}/", file=sys.stderr); return 1
+    first = read_conf(confs[0])
+    token = spotify_token(first["spotify-id"], first["spotify-secret"])
+    pl_count = 0
+    for c in confs:
+        cfg = read_conf(c)
+        url = cfg.get("input"); pname = c.stem
+        if not url or not url.startswith("https://open.spotify.com/"): continue
+        try:
+            tracks = fetch_spotify_playlist(url, token)
+        except Exception as e:
+            print(f"      WARN: {pname}: {e}"); continue
+        for artists, title in tracks:
+            title_idx.setdefault(normalize(title), []).append((pname, artists))
+        pl_count += 1
+        print(f"      {pname}: {len(tracks)} tracks")
+    print(f"      indexed {pl_count} playlists, {sum(len(v) for v in title_idx.values())} title→playlist entries")
+
+    print("[3/5] Fetching AzuraCast playlists + files...", flush=True)
+    playlists_by_name = az_playlists_by_name(args.azuracast_base, args.azuracast_key, args.station)
+    print(f"      {len(playlists_by_name)} AzuraCast playlists")
+    all_files = az_all_files(args.azuracast_base, args.azuracast_key, args.station)
+    unassigned = [r for r in all_files if not r.get("playlists")]
+    print(f"      {len(all_files)} total files, {len(unassigned)} unassigned")
+
+    print("[4/5] Matching unassigned files to playlists...", flush=True)
+    matches: list[dict] = []  # rows with .target_playlist
+    unmatched: list[dict] = []
+    multi_match: list[dict] = []  # ambiguous
+    no_az_playlist: list[dict] = []
+
+    for r in unassigned:
+        cp = f"{LIB_ROOT_CONTAINER}/{r['path']}"
+        bmeta = beets.get(cp, {})
+        # 1) trust beets GROUPING if present
+        target = bmeta.get("grouping", "").strip() if bmeta else ""
+        source = "beets_grouping" if target else None
+
+        if not target:
+            # 2) resolve via Spotify (artist, title) match
+            title_n = normalize(bmeta.get("title") or r.get("title") or "")
+            file_artists = bmeta.get("artist") or r.get("artist") or ""
+            candidates = title_idx.get(title_n, [])
+            if not candidates:
+                unmatched.append({"id": r["id"], "path": r["path"], "reason": "no spotify title hit"})
+                continue
+            # Score candidates by artist overlap
+            file_artist_n = normalize(file_artists)
+            scored = []
+            for pname, artists in candidates:
+                joined_n = normalize("; ".join(artists))
+                score = SequenceMatcher(None, file_artist_n, joined_n).ratio()
+                scored.append((score, pname))
+            scored.sort(reverse=True)
+            top_score, top_pname = scored[0]
+            if top_score < 0.4:
+                unmatched.append({"id": r["id"], "path": r["path"], "reason": f"low artist score {top_score:.2f}"})
+                continue
+            # If second-best is within 0.05, it's ambiguous between playlists
+            distinct_playlists = {p for _, p in scored}
+            if len(distinct_playlists) > 1 and len(scored) > 1 and scored[1][0] > top_score - 0.05 and scored[1][1] != top_pname:
+                multi_match.append({"id": r["id"], "path": r["path"], "candidates": [p for _, p in scored[:3]]})
+                continue
+            target = top_pname
+            source = "spotify_match"
+
+        if args.only and target != args.only:
+            continue
+        if target not in playlists_by_name:
+            no_az_playlist.append({"id": r["id"], "path": r["path"], "target": target})
+            continue
+
+        matches.append({
+            "id": r["id"], "path": r["path"], "target": target, "source": source,
+            "host_path": f"{LIB_ROOT_HOST}/{r['path']}",
+            "container_path": cp,
+            "playlist_id": playlists_by_name[target],
+        })
+
+    # Report
+    by_target: dict[str, int] = {}
+    for m in matches:
+        by_target[m["target"]] = by_target.get(m["target"], 0) + 1
+    print()
+    print(f"=== Match summary ===")
+    print(f"  matched: {len(matches)}")
+    for pname, n in sorted(by_target.items(), key=lambda x: -x[1]):
+        print(f"    {pname}: {n}")
+    print(f"  unmatched (no spotify hit): {len(unmatched)}")
+    print(f"  ambiguous (multiple playlists tied): {len(multi_match)}")
+    print(f"  no matching AzuraCast playlist: {len(no_az_playlist)}")
+    if unmatched[:5]:
+        print("  unmatched sample:")
+        for u in unmatched[:5]:
+            print(f"    {u['path']}  ({u['reason']})")
+    if multi_match[:5]:
+        print("  ambiguous sample:")
+        for u in multi_match[:5]:
+            print(f"    {u['path']}  candidates={u['candidates']}")
+
+    if not args.apply:
+        print()
+        print("DRY RUN — pass --apply to (a) write GROUPING tags, (b) beet update, (c) assign in AzuraCast")
+        return 0
+
+    print()
+    print("[5/5] Applying...", flush=True)
+    tag_ok = 0; tag_fail = 0
+    az_ok = 0; az_fail = 0
+    for m in matches:
+        # Write tag
+        if write_grouping(m["host_path"], m["target"]):
+            tag_ok += 1
+        else:
+            tag_fail += 1
+            print(f"  TAG-FAIL {m['host_path']}")
+        # Assign in AzuraCast: PUT /api/station/N/file/{id} with playlists field.
+        # Send ONLY the playlists key — sending the full record back causes
+        # AzuraCast to reject with HTTP 500 (validation fails on derived fields).
+        try:
+            az_request(args.azuracast_base, args.azuracast_key,
+                       f"/api/station/{args.station}/file/{m['id']}",
+                       method="PUT", body={"playlists": [{"id": m["playlist_id"]}]})
+            az_ok += 1
+        except Exception as e:
+            az_fail += 1
+            print(f"  AZ-FAIL {m['path']}: {e}")
+
+    print(f"  tags written: {tag_ok}  (failed: {tag_fail})")
+    print(f"  azuracast assigned: {az_ok}  (failed: {az_fail})")
+
+    # beet update so beets DB sees the new GROUPING
+    print("  running beet update on changed paths...")
+    beet_update([m["container_path"] for m in matches], None)
+    print("  done")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())