alembic/pipeline/lib/spotify-retag.py

#!/usr/bin/env python3
"""
spotify-retag.py — rewrite tags on downloaded files using Spotify as source of truth.

Reads a Spotify playlist's tracks via Client Credentials, matches each audio file
in <scan_dir> to a Spotify track, and overwrites:

  ARTIST       — semicolon-joined list of all credited artists
  ALBUMARTIST  — primary (first) artist only — kills ghost combined artists
  ALBUM
  TITLE
  TRACKNUMBER
  DISCNUMBER
  DATE         — release year

Tags we leave alone:
  GROUPING     — set by run-playlist.sh for playlist-membership tracking
  Anything else not in the list above

Usage:
  spotify-retag.py <playlist_url> <scan_dir>     # walk a directory tree
  spotify-retag.py <playlist_url> -              # read newline-separated paths from stdin

Credentials are read from env vars SPOTIFY_CLIENT_ID and SPOTIFY_CLIENT_SECRET.
SPOTIFY_REFRESH_TOKEN, if set, mints a user token instead of client-credentials
(required for newly created Spotify apps -- see _spotify_auth.get_token).

Matching strategy (per file):
  1. Parse "Artist - Title.ext" from filename. If multiple " - " separators,
     try every possible split position.
  2. Normalize title (lowercase, strip parens/brackets/feat./punct).
  3. Look up by normalized title in the playlist index. If exactly one hit,
     use it. If multiple, score each by artist-string overlap and pick the
     best. If nothing close, leave the file alone.
"""
import sys, os, re, json, urllib.request, urllib.parse, subprocess
from pathlib import Path
from difflib import SequenceMatcher

from _spotify_auth import get_token

API = "https://api.spotify.com/v1"


def fetch_playlist(url: str, token: str) -> list[dict]:
    m = re.search(r"playlist[/:]([A-Za-z0-9]+)", url)
    if not m:
        sys.exit(f"Could not parse playlist ID from {url!r}")
    pid = m.group(1)
    tracks = []
    # /items (not the removed /tracks) -- see February 2026 Spotify API changes.
    next_url = f"{API}/playlists/{pid}/items?limit=100"
    while next_url:
        req = urllib.request.Request(
            next_url, headers={"Authorization": f"Bearer {token}"}
        )
        with urllib.request.urlopen(req, timeout=15) as r:
            data = json.loads(r.read())
        # Entries carry "track" (grandfathered apps) or "item" (apps on the
        # February 2026 renamed shape); items is absent entirely when the
        # connected account can't read the playlist's contents.
        for item in data.get("items", []):
            t = item.get("track") or item.get("item")
            if not t or t.get("is_local"):
                continue
            tracks.append(
                {
                    "id": t["id"],
                    "title": t["name"],
                    "artists": [a["name"] for a in t["artists"]],
                    "album": t["album"]["name"],
                    "track_number": t.get("track_number") or 0,
                    "disc_number": t.get("disc_number") or 1,
                    "year": (t["album"].get("release_date") or "").split("-")[0],
                }
            )
        next_url = data.get("next")
    return tracks


def normalize(s: str) -> str:
    s = s.lower()
    s = re.sub(r"\s*\([^)]*\)", "", s)        # strip ( ... )
    s = re.sub(r"\s*\[[^\]]*\]", "", s)       # strip [ ... ]
    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)      # strip "feat. X" trailing
    s = re.sub(r"\s*ft\.?\s.*", "", s)
    s = re.sub(r"[^a-z0-9]+", "", s)
    return s


def index_by_title(tracks: list[dict]) -> dict[str, list[dict]]:
    idx: dict[str, list[dict]] = {}
    for t in tracks:
        key = normalize(t["title"])
        # Fully non-Latin titles (Japanese/Korean/Cyrillic) normalize to "" —
        # indexing them would dump every such track into one shared bucket and
        # cross-match unrelated songs (their artists also normalize to "", so
        # SequenceMatcher scores them 1.0). Leave them unmatched instead; the
        # file keeps its sldl-supplied tags rather than gaining wrong ones.
        if not key:
            continue
        idx.setdefault(key, []).append(t)
    return idx


def candidate_splits(stem: str) -> list[tuple[str, str]]:
    """For 'A - B - C', return both forward and reversed splits, e.g.
       [('A','B - C'), ('A - B','C'),    # forward: artist - title
        ('B - C','A'), ('C','A - B')]    # reversed: title - artist
    Some Soulseek uploaders write 'Title - Artist' instead of 'Artist - Title'.
    """
    parts = stem.split(" - ")
    if len(parts) < 2:
        return []
    forward = [
        (" - ".join(parts[:i]), " - ".join(parts[i:])) for i in range(1, len(parts))
    ]
    reversed_ = [(b, a) for a, b in forward]
    return forward + reversed_


def best_match(file_artist_raw: str, candidates: list[dict]) -> dict | None:
    """Pick the candidate whose artist-string overlaps best with the filename's."""
    if len(candidates) == 1:
        return candidates[0]
    fnorm = normalize(file_artist_raw)
    best, best_score = None, 0.0
    for c in candidates:
        # Compare against the joined artist string
        joined = "; ".join(c["artists"])
        score = SequenceMatcher(None, fnorm, normalize(joined)).ratio()
        if score > best_score:
            best, best_score = c, score
    return best if best_score >= 0.4 else None


def match_file(path: Path, idx: dict[str, list[dict]]) -> dict | None:
    stem = path.stem
    for artist_guess, title_guess in candidate_splits(stem):
        hits = idx.get(normalize(title_guess))
        if hits:
            picked = best_match(artist_guess, hits)
            if picked:
                return picked
    # Last resort: try the whole stem as title (e.g. "Title.flac" with no " - ")
    hits = idx.get(normalize(stem))
    if hits and len(hits) == 1:
        return hits[0]
    return None


def write_flac_tags(path: str, t: dict, log) -> None:
    artists_joined = "; ".join(t["artists"])
    primary = t["artists"][0]
    pairs = {
        "ARTIST": artists_joined,
        "ALBUMARTIST": primary,
        "ALBUM": t["album"],
        "TITLE": t["title"],
        "TRACKNUMBER": str(t["track_number"]),
        "DISCNUMBER": str(t["disc_number"]),
    }
    if t["year"]:
        pairs["DATE"] = t["year"]
    # Strip "release identification" tags from MB-tagged Soulseek uploads.
    # Navidrome's BFR scanner uses these (alongside MB IDs) to differentiate
    # releases of the same album, so leftover RELEASE*/BARCODE/CATALOGNUMBER/
    # LABEL/MEDIA tags on one track will split it off into its own album card.
    # Saw this with Danny Brown's Atrocity Exhibition where Pneumonia had MB
    # IDs AND a CATALOGNUMBER and split off twice in succession before we
    # broadened the strip list. Keep this list in sync with strip-mb-tags.sh.
    stale_mb_tags = [
        # MusicBrainz IDs
        "MUSICBRAINZ_ALBUMID", "MUSICBRAINZ_RELEASEGROUPID",
        "MUSICBRAINZ_RELEASETRACKID", "MUSICBRAINZ_ALBUMARTISTID",
        "MUSICBRAINZ_ARTISTID", "MUSICBRAINZ_ALBUMSTATUS",
        "MUSICBRAINZ_ALBUMTYPE", "MUSICBRAINZ_TRACKID", "MUSICBRAINZ_WORKID",
        # Release identification
        "RELEASESTATUS", "RELEASETYPE", "RELEASECOUNTRY",
        "BARCODE", "CATALOGNUMBER", "LABEL", "PUBLISHER",
        "MEDIA", "ORIGINALDATE", "ASIN", "ISRC", "SCRIPT", "LANGUAGE",
        # Duplicate / sort variants of artist fields (we use canonical ARTIST/ALBUMARTIST)
        "ALBUM ARTIST", "ALBUM_ARTIST",
        "ALBUMARTIST_CREDIT", "ALBUMARTISTSORT", "ALBUMARTISTS",
        "ALBUM_ARTISTS", "ALBUMARTISTS_CREDIT", "ALBUMARTISTS_SORT",
        "ARTIST_CREDIT", "ARTISTSORT", "ARTISTS",
        "ARTISTS_CREDIT", "ARTISTS_SORT", "COMPOSERSORT",
        # Misc cruft
        "ACOUSTID_ID", "ACOUSTID_FINGERPRINT", "COMPILATION",
        "YEAR",  # duplicate of DATE
        "DISCSUBTITLE", "DISCC", "TOTALDISCS", "TRACKC", "TOTALTRACKS",
        # Kept in sync with mb-tags.sh (was missing these four)
        "MUSICBRAINZ_ALBUMCOMMENT", "ALBUMCOMMENT",
        "MUSICBRAINZ_DISCID", "MUSICBRAINZ_TRMID",
    ]
    # One metaflac invocation: options are applied in order and the file is
    # rewritten once, so a failure leaves the original tags intact — the old
    # per-tag loop (~65 subprocesses) could die between remove and set, leaving
    # a file stripped of ARTIST/TITLE.
    args = ["metaflac"]
    args += [f"--remove-tag={tag}" for tag in list(pairs) + stale_mb_tags]
    args += [f"--set-tag={tag}={val}" for tag, val in pairs.items()]
    args.append(path)
    subprocess.run(args, check=True, stderr=log)


def write_mp3_tags(path: str, t: dict, log) -> None:
    artists_joined = "; ".join(t["artists"])
    args = [
        "id3v2",
        "--TPE1", artists_joined,
        "--TPE2", t["artists"][0],
        "--TALB", t["album"],
        "--TIT2", t["title"],
        "--TRCK", str(t["track_number"]),
        "--TPOS", str(t["disc_number"]),
    ]
    if t["year"]:
        args += ["--TYER", t["year"]]
    args.append(path)
    subprocess.run(args, check=True, stderr=log)


def main() -> int:
    if len(sys.argv) != 3:
        sys.exit(f"Usage: {sys.argv[0]} <playlist_url> <scan_dir|->")
    url, target = sys.argv[1], sys.argv[2]
    cid = os.environ.get("SPOTIFY_CLIENT_ID")
    csec = os.environ.get("SPOTIFY_CLIENT_SECRET")
    refresh = os.environ.get("SPOTIFY_REFRESH_TOKEN") or None
    if not cid or not csec:
        sys.exit("SPOTIFY_CLIENT_ID and SPOTIFY_CLIENT_SECRET must be set")

    if target == "-":
        files_iter = (Path(line.strip()) for line in sys.stdin if line.strip())
    else:
        files_iter = Path(target).rglob("*")

    print(f"[spotify-retag] Fetching playlist {url}")
    token = get_token(cid, csec, refresh)
    tracks = fetch_playlist(url, token)
    print(f"[spotify-retag] Got {len(tracks)} tracks from Spotify")

    idx = index_by_title(tracks)
    matched = unmatched = failed = 0
    unmatched_names: list[str] = []

    for f in sorted(files_iter):
        if not f.is_file() or f.suffix.lower() not in (".flac", ".mp3"):
            continue
        match = match_file(f, idx)
        if not match:
            unmatched += 1
            unmatched_names.append(f.name)
            continue
        try:
            if f.suffix.lower() == ".flac":
                write_flac_tags(str(f), match, sys.stderr)
            else:
                write_mp3_tags(str(f), match, sys.stderr)
            matched += 1
            print(f"  OK   {match['artists'][0]} — {match['title']}  ({f.name})")
        except subprocess.CalledProcessError as e:
            failed += 1
            print(f"  FAIL {f.name}: {e}", file=sys.stderr)

    print(f"\n[spotify-retag] {matched} matched, {unmatched} unmatched, {failed} failed")
    for u in unmatched_names[:25]:
        print(f"  unmatched: {u}")
    if len(unmatched_names) > 25:
        print(f"  ...and {len(unmatched_names) - 25} more")
    return 0


if __name__ == "__main__":
    sys.exit(main())