Stage 0: migrate pipeline scripts from /opt/sldl, scaffold repo
Moves all ~30 pipeline scripts, configs, and the vendored sldl binary into this repo (source /opt/sldl left untouched). Removes all docker exec/docker compose dependencies now that beets and sldl run in-process/as a subprocess of this container instead of via soulbeet/on-demand sldl containers. Replaces hardcoded host paths, Navidrome credentials, and Spotify credential sourcing with env-var-driven paths and shared credential loaders. Adds Dockerfile, entrypoint.sh, requirements.txt, docker-compose.snippet.yml, and the initial app DB schema.
This commit is contained in:
andrew
Executable
+282
@@ -0,0 +1,282 @@
|
||||
#!/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, base64, urllib.request, urllib.parse, subprocess
|
||||
from pathlib import Path
|
||||
from difflib import SequenceMatcher
|
||||
|
||||
API = "https://api.spotify.com/v1"
|
||||
|
||||
|
||||
def get_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_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())
|
||||
Reference in New Issue
Block a user