#!/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 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 # walk a directory tree spotify-retag.py - # 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]} ") 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())