alembic/pipeline/lib/fix-track-metadata.py

#!/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 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"
# Still "/music" through the migration's Stage 0-3 transitional beets mount;
# revisit at Stage 4 once beets' directory: becomes MUSIC_DATA_DIR/Library.
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('"')
    cid, csec = env["SPOTIFY_CLIENT_ID"], env["SPOTIFY_CLIENT_SECRET"]
    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"})
    return json.loads(urllib.request.urlopen(req, timeout=15).read())["access_token"]


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):
    cp = 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()