andrew/alembic
1
0
Fork 0
Code Pull Requests Actions Releases 12 Wiki Activity
Files
62251d764ed066796ee8af7fc605831414c6f634
alembic/pipeline/lib/fix-track-metadata.py
T
andrew 62251d764e 0.6.9: Finish the Stage 4 path revisit in the three remaining scripts 
The 2026-07-08 path rewrite changed beets' displayed paths from the
transitional /music/... mount to real /data/music/Library/... paths.
dedup-library.sh was fixed in 0.6.8; three more scripts still assumed
the old prefix, each failing silently:

- replace-with-better.sh resolved every library match to a doubled
  nonexistent path, logged [stale-db], skipped the in-place replace,
  and then imported the staged FLAC as a NEW track via the
  leftover-import step. Net effect: the mp3-to-flac upgrade flow
  produced flac+mp3 twin pairs in the library (surfacing in the dedup
  queue) instead of replacing the MP3 in place.

- fix-track-metadata.py queried beets only by the legacy /music/...
  form, matched nothing, and silently skipped beet update/move after
  retagging.

- export-laptop-playlists.py filtered out every library track (no
  displayed path starts with /music/ anymore), producing empty
  exports.

All three now use the real displayed path and keep the /music/... form
only as a legacy fallback.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-16 11:51:55 -06:00

517 lines
19 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""
fix-track-metadata.py — interactively rewrite a single track's metadata
from a Spotify or iTunes URL.
Prompts for an audio file path and a Spotify/iTunes URL, then:
1. Fetches canonical metadata (artist, album, title, track#, date, cover).
2. Writes tags to the file (artist semicolon-joined, albumartist primary
only — matches the rest of the booksandtunes pipeline).
3. Strips SoundCloud / upload-junk frames (WWWAUDIOFILE, comment,
compatible_brands, major_brand, minor_version).
4. Replaces embedded cover with the URL's artwork and drops a cover.jpg
sidecar in the album folder.
5. Runs `beet update` so beets's DB matches.
6. Runs `beet move` to relocate the file under
/music/<albumartist>/<album>/<title>.<ext>.
7. Strips GENRE and runs spotify-genre.py to refill from the artist's
Spotify per-artist genres (use --keep-genre to skip this).
8. Triggers a Navidrome rescan via the Subsonic API.
Usage:
./fix-track-metadata.py # interactive
./fix-track-metadata.py <file> <url> # non-interactive
./fix-track-metadata.py --keep-genre <f> <u> # leave existing GENRE alone
The <file> arg accepts any of:
- absolute host path: $MUSIC_DATA_DIR/Library/Foo/Bar.flac
- beets container path: /music/Foo/Bar.flac
- library-relative path: Foo/Bar.flac
(resolved against $MUSIC_DATA_DIR/Library/)
Supported URLs:
https://open.spotify.com/track/<id>...
https://open.spotify.com/album/<id>... (single-track albums only)
https://music.apple.com/.../id<col>?i=<track> (iTunes track URL)
"""
import sys, os, re, json, base64, argparse, subprocess, unicodedata, urllib.parse, urllib.request
from _spotify_auth import get_token
from mutagen import File as MFile
from mutagen.id3 import ID3, ID3NoHeaderError, TPE1, TPE2, TALB, TIT2, TRCK, TDRC, TCON, APIC
from mutagen.flac import FLAC, Picture
from mutagen.mp4 import MP4, MP4Cover
from mutagen.oggopus import OggOpus
from mutagen.oggvorbis import OggVorbis
_ALEMBIC_CONFIG_DIR = os.environ.get("ALEMBIC_CONFIG_DIR", "/config")
_MUSIC_DATA_DIR = os.environ.get("MUSIC_DATA_DIR", "/data/music")
LIBRARY = f"{_MUSIC_DATA_DIR}/Library"
# Legacy prefix from the migration's transitional beets mount. Since the
# 2026-07-08 path rewrite beets displays real LIBRARY paths, so this only
# matters for accepting pasted /music/... paths as input and as a fallback
# beets query form for any stale pre-rewrite DB entry.
CONTAINER_PREFIX = "/music"
SPOTIFY_ENV = f"{_ALEMBIC_CONFIG_DIR}/pipeline/_spotify.env"
SPOTIFY_GENRE = f"{os.environ.get('PIPELINE_DIR', '/app/pipeline')}/lib/spotify-genre.py"
def _load_navidrome_env():
path = f"{_ALEMBIC_CONFIG_DIR}/pipeline/navidrome/admin.env"
env = {}
if os.path.exists(path):
for line in open(path):
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
k, v = line.split("=", 1)
env[k] = v.strip().strip("'").strip('"')
return env
_nd_env = _load_navidrome_env()
NAVIDROME_URL = f"{_nd_env.get('ND_BASE', 'http://navidrome:4533')}/rest/startScan.view"
NAVIDROME_USER = _nd_env.get("ND_USER", "")
NAVIDROME_PASS = _nd_env.get("ND_PASS", "")
VORBIS_LIKE = (FLAC, OggOpus, OggVorbis)
JUNK_TXXX = {"WWWAUDIOFILE", "comment", "compatible_brands",
"major_brand", "minor_version"}
# ---------- Spotify ----------
def _spotify_token():
env = {}
with open(SPOTIFY_ENV) as f:
for line in f:
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
k, v = line.split("=", 1)
env[k] = v.strip().strip("'").strip('"')
return get_token(env["SPOTIFY_CLIENT_ID"], env["SPOTIFY_CLIENT_SECRET"],
env.get("SPOTIFY_REFRESH_TOKEN") or None)
def _spotify_get(path, token):
req = urllib.request.Request(f"https://api.spotify.com/v1{path}",
headers={"Authorization": f"Bearer {token}"})
return json.loads(urllib.request.urlopen(req, timeout=15).read())
def fetch_spotify_track(track_id):
tok = _spotify_token()
t = _spotify_get(f"/tracks/{track_id}", tok)
a = t["album"]
return {
"title": t["name"],
"artists": [ar["name"] for ar in t["artists"]],
"album": a["name"],
"albumartists": [ar["name"] for ar in a["artists"]],
"date": a["release_date"],
"track": t["track_number"],
"tracktotal": a["total_tracks"],
"cover_url": a["images"][0]["url"] if a["images"] else None,
}
def fetch_spotify_album(album_id):
tok = _spotify_token()
a = _spotify_get(f"/albums/{album_id}", tok)
if a["total_tracks"] != 1:
raise SystemExit(f"album has {a['total_tracks']} tracks — pass a track URL instead")
t = a["tracks"]["items"][0]
return {
"title": t["name"],
"artists": [ar["name"] for ar in t["artists"]],
"album": a["name"],
"albumartists": [ar["name"] for ar in a["artists"]],
"date": a["release_date"],
"track": 1,
"tracktotal": 1,
"cover_url": a["images"][0]["url"] if a["images"] else None,
}
# ---------- iTunes ----------
def fetch_itunes_track(track_id):
url = f"https://itunes.apple.com/lookup?id={track_id}&entity=song"
data = json.loads(urllib.request.urlopen(url, timeout=15).read())
items = [r for r in data.get("results", []) if r.get("wrapperType") == "track"]
if not items:
raise SystemExit(f"iTunes lookup returned no track for id={track_id}")
r = items[0]
# iTunes only gives one artist string — the project convention is
# semicolon-joined for ARTIST. The user can split manually after if needed.
artists = [r["artistName"]]
cover = r.get("artworkUrl100", "")
if cover:
# 100x100 → 600x600 by URL substitution (well-known iTunes hack)
cover = re.sub(r"/\d+x\d+bb\.jpg$", "/600x600bb.jpg", cover)
return {
"title": r["trackName"],
"artists": artists,
"album": r["collectionName"],
"albumartists": artists,
"date": (r.get("releaseDate") or "")[:10],
"track": r.get("trackNumber", 1),
"tracktotal": r.get("trackCount", 1),
"cover_url": cover or None,
}
# ---------- URL dispatch ----------
def fetch_metadata(url):
m = re.search(r"open\.spotify\.com/track/([A-Za-z0-9]+)", url)
if m:
return fetch_spotify_track(m.group(1))
m = re.search(r"open\.spotify\.com/album/([A-Za-z0-9]+)", url)
if m:
return fetch_spotify_album(m.group(1))
m = re.search(r"music\.apple\.com/.+?[?&]i=(\d+)", url)
if m:
return fetch_itunes_track(m.group(1))
m = re.search(r"music\.apple\.com/.+?/song/[^/]+/(\d+)", url)
if m:
return fetch_itunes_track(m.group(1))
raise SystemExit(f"unrecognized URL: {url}")
# ---------- Tagging ----------
def write_tags(filepath, meta):
"""Write artist/albumartist/album/title/track/date and embedded cover.
Drops watermark/junk TXXX frames as a side effect.
"""
artist_joined = "; ".join(meta["artists"])
albumartist = meta["albumartists"][0] if meta["albumartists"] else meta["artists"][0]
cover_bytes = _download_cover(meta["cover_url"]) if meta["cover_url"] else None
date = meta["date"] # YYYY-MM-DD or YYYY
m = MFile(filepath)
if m is None:
raise SystemExit(f"unsupported audio format: {filepath}")
if isinstance(m, VORBIS_LIKE):
_write_vorbis(m, artist_joined, albumartist, meta, date, cover_bytes)
elif isinstance(m, MP4):
_write_mp4(m, artist_joined, albumartist, meta, date, cover_bytes)
else:
_write_id3(filepath, artist_joined, albumartist, meta, date, cover_bytes)
def _write_vorbis(m, artist_joined, albumartist, meta, date, cover_bytes):
m.tags["ARTIST"] = artist_joined
m.tags["ALBUMARTIST"] = albumartist
m.tags["ALBUM"] = meta["album"]
m.tags["TITLE"] = meta["title"]
m.tags["TRACKNUMBER"] = str(meta["track"])
m.tags["TRACKTOTAL"] = str(meta["tracktotal"])
m.tags["DATE"] = date
if cover_bytes and isinstance(m, FLAC):
m.clear_pictures()
pic = Picture()
pic.type = 3 # Cover (front)
pic.mime = "image/jpeg"
pic.data = cover_bytes
m.add_picture(pic)
m.save()
def _write_mp4(m, artist_joined, albumartist, meta, date, cover_bytes):
m.tags["\xa9ART"] = artist_joined
m.tags["aART"] = albumartist
m.tags["\xa9alb"] = meta["album"]
m.tags["\xa9nam"] = meta["title"]
m.tags["\xa9day"] = date
m.tags["trkn"] = [(int(meta["track"]), int(meta["tracktotal"]))]
if cover_bytes:
m.tags["covr"] = [MP4Cover(cover_bytes, imageformat=MP4Cover.FORMAT_JPEG)]
m.save()
def _write_id3(filepath, artist_joined, albumartist, meta, date, cover_bytes):
try:
t = ID3(filepath)
except ID3NoHeaderError:
t = ID3()
t.delall("TPE1"); t.add(TPE1(encoding=3, text=artist_joined))
t.delall("TPE2"); t.add(TPE2(encoding=3, text=albumartist))
t.delall("TALB"); t.add(TALB(encoding=3, text=meta["album"]))
t.delall("TIT2"); t.add(TIT2(encoding=3, text=meta["title"]))
t.delall("TRCK"); t.add(TRCK(encoding=3, text=f"{meta['track']}/{meta['tracktotal']}"))
t.delall("TDRC"); t.add(TDRC(encoding=3, text=date))
# Strip junk TXXX frames
for frame in list(t.getall("TXXX")):
if frame.desc in JUNK_TXXX:
t.delall(f"TXXX:{frame.desc}")
if cover_bytes:
t.delall("APIC")
t.add(APIC(encoding=3, mime="image/jpeg", type=3, desc="Cover", data=cover_bytes))
t.save(filepath)
def _download_cover(url):
if not url:
return None
return urllib.request.urlopen(url, timeout=20).read()
# ---------- Beets / Navidrome ----------
def host_to_container(host_path):
if host_path.startswith(LIBRARY):
return CONTAINER_PREFIX + host_path[len(LIBRARY):]
return host_path
def container_to_host(cp):
if cp.startswith(CONTAINER_PREFIX + "/"):
return LIBRARY + cp[len(CONTAINER_PREFIX):]
return cp
def resolve_input_path(arg):
"""Accept a host path, container path, or library-relative path; return
the absolute host path that actually exists on disk.
Falls back through NFC/NFD Unicode normalizations because terminal pastes
of Japanese/Korean/etc. paths often come in decomposed form even when the
filesystem stores them composed (or vice versa)."""
arg = arg.strip().strip('"').strip("'")
if arg.startswith(CONTAINER_PREFIX + "/"):
candidate = container_to_host(arg)
elif os.path.isabs(arg):
candidate = arg
else:
candidate = os.path.join(LIBRARY, arg)
if os.path.exists(candidate):
return candidate
for form in ("NFC", "NFD", "NFKC", "NFKD"):
norm = unicodedata.normalize(form, candidate)
if os.path.exists(norm):
return norm
# Last resort: walk component by component, picking the entry whose NFC
# form matches our NFC form. Handles the common case where one directory
# in the path differs in normalization but the rest is fine.
if not candidate.startswith("/"):
return candidate
parts = candidate.lstrip("/").split("/")
cur = "/"
for want in parts:
if not os.path.isdir(cur):
return candidate
want_nfc = unicodedata.normalize("NFC", want)
match = next(
(n for n in os.listdir(cur)
if unicodedata.normalize("NFC", n) == want_nfc),
None)
if match is None:
return candidate
cur = os.path.join(cur, match)
return cur
def beets_id_for(host_path):
# Real path first (how beets displays everything since the 2026-07-08
# path rewrite), legacy /music/... form second for any stale entry.
# Querying only the legacy form (the old behavior) matched nothing after
# the rewrite, so retag-from-url silently skipped beet update/move.
for cp in (host_path, host_to_container(host_path)):
r = subprocess.run(
["beet", "ls", "-f", "$id|||$path", f"path:{cp}"],
capture_output=True, text=True, check=True)
for line in r.stdout.splitlines():
if "|||" in line:
id_str, p = line.split("|||", 1)
if p == cp:
return int(id_str)
return None
def beet_update(beets_id):
# `beet update` prompts y/n on each diff; `yes y` auto-confirms.
subprocess.run(
f"yes y | beet update id:{beets_id}",
shell=True, check=False)
def beet_move(beets_id):
"""Beets path-format relocates the file under LIBRARY/<albumartist>/<album>/."""
subprocess.run(["beet", "move", f"id:{beets_id}"], check=False)
def beet_path(beets_id):
r = subprocess.run(
["beet", "ls", "-f", "$path", f"id:{beets_id}"],
capture_output=True, text=True, check=True)
cp = r.stdout.strip()
return container_to_host(cp) if cp else None
def strip_genre(filepath):
"""Blank GENRE so spotify-genre.py will refill it (skip-existing logic)."""
m = MFile(filepath)
if m is None or m.tags is None:
return
if isinstance(m, VORBIS_LIKE):
if "genre" in m.tags:
del m.tags["genre"]
m.save()
elif isinstance(m, MP4):
if "\xa9gen" in m.tags:
del m.tags["\xa9gen"]
m.save()
else:
try:
t = ID3(filepath)
t.delall("TCON")
t.save(filepath)
except ID3NoHeaderError:
pass
def refill_genre(beets_id):
subprocess.run(
[SPOTIFY_GENRE, "--apply", "--force", "--query", f"id:{beets_id}"],
check=False)
def write_cover_sidecar(filepath, cover_bytes):
if not cover_bytes:
return
sidecar = os.path.join(os.path.dirname(filepath), "cover.jpg")
with open(sidecar, "wb") as f:
f.write(cover_bytes)
def navidrome_scan():
qs = urllib.parse.urlencode({
"u": NAVIDROME_USER, "p": NAVIDROME_PASS,
"v": "1.16.0", "c": "fix-track-metadata", "f": "json",
})
try:
urllib.request.urlopen(f"{NAVIDROME_URL}?{qs}", timeout=10).read()
print("[navidrome] rescan triggered")
except Exception as e:
print(f"[navidrome] rescan failed: {e}")
# ---------- Main ----------
def main():
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("file", nargs="?",
help="audio file: host path, /music/... container path, "
"or library-relative path (Foo/Bar.flac)")
ap.add_argument("url", nargs="?", help="Spotify or iTunes URL")
ap.add_argument("--keep-genre", action="store_true",
help="don't strip + refill GENRE (default: refill via spotify-genre.py)")
args = ap.parse_args()
# Tab-completion on the file prompt — pasting unicode paths is finicky,
# tab-completing them is not.
try:
import readline
readline.set_completer_delims(" \t\n;")
def _path_complete(text, state):
base = text if text.startswith("/") else os.path.join(LIBRARY, text)
d, prefix = os.path.split(base)
if not os.path.isdir(d):
return None
matches = [n for n in os.listdir(d) if n.startswith(prefix)]
matches.sort()
if state >= len(matches):
return None
full = os.path.join(d, matches[state])
# Strip the LIBRARY prefix back off if user was typing relative
display = full[len(LIBRARY) + 1:] if (
not text.startswith("/") and full.startswith(LIBRARY + "/")) else full
return display + ("/" if os.path.isdir(full) else "")
readline.set_completer(_path_complete)
readline.parse_and_bind("tab: complete")
except ImportError:
pass
raw_file = args.file or input("file path: ").strip()
url = args.url or input("spotify/itunes URL: ").strip()
filepath = resolve_input_path(raw_file)
if not os.path.isfile(filepath):
# Diagnostics: show the bytes we tried + what's actually in the parent.
print(f"[error] not a file: {filepath}", file=sys.stderr)
print(f"[error] input bytes: {raw_file.encode('utf-8')!r}", file=sys.stderr)
print(f"[error] resolved bytes: {filepath.encode('utf-8')!r}", file=sys.stderr)
parent = os.path.dirname(filepath)
if os.path.isdir(parent):
print(f"[error] parent dir contents:", file=sys.stderr)
for n in sorted(os.listdir(parent)):
print(f" {n!r} ({n.encode('utf-8')!r})", file=sys.stderr)
else:
# Walk up to the deepest existing ancestor and list it
anc = parent
while anc and anc != "/" and not os.path.isdir(anc):
anc = os.path.dirname(anc)
if anc and os.path.isdir(anc):
print(f"[error] deepest existing dir: {anc}", file=sys.stderr)
print(f"[error] its contents:", file=sys.stderr)
for n in sorted(os.listdir(anc)):
print(f" {n!r}", file=sys.stderr)
raise SystemExit(1)
print(f"[fetch] {url}")
meta = fetch_metadata(url)
print(f" title: {meta['title']}")
print(f" artists: {'; '.join(meta['artists'])}")
print(f" album: {meta['album']}")
print(f" albumartist: {meta['albumartists'][0] if meta['albumartists'] else '?'}")
print(f" date: {meta['date']}")
print(f" track: {meta['track']}/{meta['tracktotal']}")
print(f" cover: {meta['cover_url'] or '(none)'}")
if not args.file:
ok = input("apply? [y/N] ").strip().lower()
if ok != "y":
print("aborted")
return
print("[tags] writing")
cover_bytes = _download_cover(meta["cover_url"]) if meta["cover_url"] else None
write_tags(filepath, meta)
write_cover_sidecar(filepath, cover_bytes)
bid = beets_id_for(filepath)
if bid is None:
print("[beets] file not in beets DB — skipping update/move/genre")
else:
print(f"[beets] update id:{bid}")
beet_update(bid)
print(f"[beets] move id:{bid}")
beet_move(bid)
new_path = beet_path(bid)
if new_path:
filepath = new_path
# Re-drop sidecar in the (possibly new) folder
write_cover_sidecar(filepath, cover_bytes)
print(f"[beets] now at {filepath}")
if not args.keep_genre:
print("[genre] stripping + refilling via spotify-genre.py")
strip_genre(filepath)
beet_update(bid)
refill_genre(bid)
beet_update(bid)
navidrome_scan()
print("[done]")
if __name__ == "__main__":
main()
View Git Blame Copy Permalink