alembic/pipeline/lib/clean-sldl-index.py

#!/usr/bin/env python3
"""
clean-sldl-index.py — for each playlist's _sldl.m3u8, drop entries whose
track is no longer anywhere in the beets library. sldl uses the m3u as its
"already downloaded" skip list, so anything left in there is skipped on
the next playlist run. By removing entries for files we deleted (placeholders,
quarantined junk), we let sldl retry them fresh.

Conservative matching — we only drop a m3u line if no library track even
loosely matches its (artist, title). If we drop too aggressively, sldl
re-downloads good tracks we still have. If we keep too liberally, deleted
placeholders linger in the m3u — but that's a smaller problem.

Usage:
  clean-sldl-index.py              # dry run
  clean-sldl-index.py --apply      # rewrite m3us in place (.bak.<ts> backup)
"""
import sys, re, subprocess, argparse, os, time, shutil
from pathlib import Path
from difflib import SequenceMatcher

DROPBOX = f"{os.environ.get('MUSIC_DATA_DIR', '/data/music')}/sldl-dropbox"


def normalize(s):
    """Lowercase, strip parens content + feat/ft suffix, collapse whitespace.
    Keep ALL alphanumeric (including Unicode/Japanese) — strip only spaces and
    common punctuation."""
    s = s.lower()
    s = re.sub(r"\s*\([^)]*\)", "", s)
    s = re.sub(r"\s*\[[^\]]*\]", "", s)
    s = re.sub(r"\s*feat\.?\s.*", "", s)
    s = re.sub(r"\s*ft\.?\s.*", "", s)
    s = re.sub(r"[\s\-_,;:&'\"!?.()\[\]/]+", "", s)
    return s


def fetch_beets_index():
    """Return list of (norm_artist_title, norm_artist, norm_title) tuples for
    every track in beets, normalized for comparison."""
    r = subprocess.run(
        ["beet", "ls", "-f", "$artist‖$title"],
        capture_output=True, text=True, timeout=120
    )
    out = []
    for line in r.stdout.splitlines():
        if "‖" not in line:
            continue
        artist, title = line.split("‖", 1)
        n_a = normalize(artist)
        n_t = normalize(title)
        out.append((n_a + n_t, n_a, n_t))
    return out


def parse_m3u_line(line):
    """Extract (artist, title) from a m3u filename like 'Artist - Title.ext'.
    If the line is FAIL-prefixed or comment, return (None, None)."""
    s = line.rstrip("\n")
    if not s or s.startswith("#"):
        return None, None
    stem = re.sub(r"\.[a-zA-Z0-9]{1,5}$", "", s)
    if " - " in stem:
        # First " - " split: typical "Artist - Title"
        a, _, t = stem.partition(" - ")
        return a.strip(), t.strip()
    return None, stem


def m3u_matches_beets(line, beets_idx):
    """Return True if some beets track plausibly matches this m3u line."""
    artist, title = parse_m3u_line(line)
    if artist is None and title is None:
        return True  # comment/blank — keep
    n_a = normalize(artist or "")
    n_t = normalize(title or "")
    n_full = n_a + n_t

    if not n_full:
        return True  # nothing to match against — be safe and keep

    for beets_full, beets_a, beets_t in beets_idx:
        if not beets_full:
            continue
        # Cheap exact full match
        if n_full == beets_full:
            return True
        # Mutual substring (handles multi-artist credits added to one side)
        if n_full in beets_full or beets_full in n_full:
            return True
        # Title-only match if artist sides are very different (multi-artist case)
        if n_t and beets_t and (n_t == beets_t or n_t in beets_t or beets_t in n_t):
            # require some artist overlap (substring either way) to avoid false hits
            if n_a and beets_a and (n_a in beets_a or beets_a in n_a):
                return True
            # Romaji ↔ Japanese cases: the m3u uses Spotify's romaji artist
            # ("natori") but the file has the original kanji ("なとり"). Exact
            # title match with a non-trivial length is enough.
            if n_t == beets_t and len(n_t) > 6:
                return True
        # Last-ditch fuzzy: very similar overall strings
        if len(n_full) > 6 and len(beets_full) > 6:
            ratio = SequenceMatcher(None, n_full, beets_full).quick_ratio()
            if ratio > 0.85:
                return True
    return False


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--apply", action="store_true")
    args = ap.parse_args()

    print(f"[clean-index] mode={'APPLY' if args.apply else 'DRY RUN'}")
    print("[clean-index] loading beets library index...")
    beets_idx = fetch_beets_index()
    print(f"[clean-index] indexed {len(beets_idx)} tracks")

    # Safety guard: an empty index means `beet ls` failed (missing/locked DB,
    # bad config) — NOT that the library is empty. Without this check,
    # --apply would drop every entry from every _sldl.m3u8 and the next
    # playlist runs would re-download the entire library from Soulseek.
    if not beets_idx:
        print("[clean-index] ABORT: beets index came back empty "
              "(beet ls failed) — refusing to touch any m3u",
              file=sys.stderr)
        return 1

    total_kept = 0
    total_dropped = 0
    total_files = 0

    for m3u in sorted(Path(DROPBOX).glob("*/_sldl.m3u8")):
        playlist = m3u.parent.name
        with open(m3u) as f:
            lines = f.readlines()
        kept, dropped = [], []
        for line in lines:
            if m3u_matches_beets(line, beets_idx):
                kept.append(line)
            else:
                dropped.append(line)
        print(f"\n[{playlist}] {len(kept)} kept, {len(dropped)} dropped")
        for d in dropped[:8]:
            print(f"  DROP {d.rstrip()}")
        if len(dropped) > 8:
            print(f"  ...and {len(dropped) - 8} more")

        total_kept += len(kept)
        total_dropped += len(dropped)
        total_files += 1

        if args.apply and dropped:
            backup = f"{m3u}.bak.{int(time.time())}"
            shutil.copy(m3u, backup)
            with open(m3u, "w") as f:
                f.writelines(kept)
            print(f"  [{playlist}] rewritten; backup at {backup}")

    print(f"\n[clean-index] {total_files} m3us, total {total_kept} kept, {total_dropped} dropped")
    if not args.apply:
        print("[clean-index] DRY RUN — re-run with --apply to rewrite the m3us")


if __name__ == "__main__":
    sys.exit(main())