andrew
bcd07750bf
- The client-credentials token fetch was copy-pasted across spotify-retag.py, spotify-genre.py, and fix-track-metadata.py (and the app's spotify_client). Add pipeline/lib/_spotify_auth.get_token (cached per id/secret); the three scripts now source their own credentials but delegate the request to it. The scripts run with pipeline/lib on sys.path, so the plain `from _spotify_auth import get_token` resolves. - The identical _parse_json_lines helper in dedup_review_service and genre_review_service is now a single app/services/_ndjson.parse_json_lines. Verified: unit test of the token helper (cache + request), the NDJSON parser (tests/test_ndjson.py), full suite green (40), and a live spotify-genre dry-run that fetched a token and queried Spotify through the shared helper. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
271 lines
10 KiB
Python
Executable File
271 lines
10 KiB
Python
Executable File
#!/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.
|
|
|
|
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 = []
|
|
next_url = f"{API}/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=15) 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
|
|
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")
|
|
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)
|
|
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())
|