spotify

maple-py/src/maple/collection/__init__.py

@@ -0,0 +1,27 @@
+"""Sync data/collection with one or more Spotify playlists.
+
+Pipeline (one shot, in `maple.collection.cli.main_async`):
+  1. Async-fetch each Spotify playlist via aiohttp (concurrently).
+  2. Diff against the Collection's index (sldl `index.sldl.csv` + tag scan).
+  3. Write the missing rows as a CSV and shell out to `sockseek` to download
+     them (streaming `--progress-json` events back so each completed track
+     updates its playlist's m3u8 immediately).
+  4. Validate every track that now resolves to a file: ffprobe-decodable,
+     duration within 5s of Spotify, fuzzy tag match, embedded cover art.
+     If artwork is missing, fetch the largest image from
+     `track.album.images` and embed it.
+  5. Write one m3u8 per playlist into `<root>/_playlists/<name>.m3u8` with
+     paths relative to the m3u8 file (portable across synced devices).
+
+Usage:
+  uv run python -m maple.collection <id-or-url ...> [--collection PATH]
+                                    [--no-download] [--no-validate]
+
+Env (from .env, loaded by just):
+  MAPLE_SPOTIFY_CLIENT_ID, MAPLE_SPOTIFY_CLIENT_SECRET (spotify.py reads)
+  SOCKSEEK_USER, SOCKSEEK_PASS (forwarded to sockseek)
+"""
+from maple.collection.core import Collection, TrackIssue
+from maple.collection.index import CollectionIndex
+
+__all__ = ["Collection", "CollectionIndex", "TrackIssue"]

maple-py/src/maple/collection/__main__.py

@@ -0,0 +1,4 @@
+"""Entrypoint for `python -m maple.collection`."""
+from maple.collection.cli import main
+
+raise SystemExit(main())

maple-py/src/maple/collection/cli.py

@@ -0,0 +1,192 @@
+"""`just collection` CLI entrypoint.
+
+End-to-end pipeline: fetch Spotify playlists, diff against the on-disk
+collection, run sockseek for the missing tracks, validate the result,
+write per-playlist m3u8s, and emit a report.
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import sys
+from pathlib import Path
+
+import aiohttp
+from loguru import logger
+
+from maple.collection.core import Collection, TrackIssue
+from maple.collection.index import first_artist, norm
+from maple.log import configure as configure_logging
+from maple.spotify import fetch_playlists
+
+
+def parse_args(argv: list[str]) -> argparse.Namespace:
+    p = argparse.ArgumentParser(prog="maple.collection", description="Sync data/collection with Spotify playlists.")
+    p.add_argument("ids", nargs="+", help="Spotify playlist ids, URLs, or spotify: URIs")
+    p.add_argument("--collection", help="Override collection root (default: <repo>/data/collection)")
+    p.add_argument("--no-download", action="store_true", help="Skip the sockseek download step")
+    p.add_argument("--no-validate", action="store_true", help="Skip ffprobe/duration/tag/art validation")
+    return p.parse_args(argv)
+
+
+async def main_async(argv: list[str]) -> int:
+    args = parse_args(argv)
+    configure_logging()
+    collection = Collection(Path(args.collection).resolve()) if args.collection else Collection.from_repo()
+    collection.ensure_dirs()
+    collection.attach_report_sink()
+    logger.info("collection root: {}", collection.root)
+
+    # Startup hygiene: drop unfinished downloads, prune the sldl index, and
+    # rewrite any stale entries out of existing m3u8s.
+    removed_incomplete = collection.remove_unfinished_downloads()
+    if removed_incomplete:
+        logger.warning("removed {} unfinished download(s)", removed_incomplete)
+    kept, removed = collection.validate_index_file()
+    if removed:
+        logger.warning("index: pruned {} row(s) ({} kept)", removed, kept)
+    pruned_pl, rewritten_pl = collection.validate_playlists()
+    if pruned_pl or rewritten_pl:
+        logger.warning(
+            "playlists: pruned {} stale entr(ies), rewrote {} absolute path(s)",
+            pruned_pl, rewritten_pl,
+        )
+
+    playlists = await fetch_playlists(args.ids)
+    total = sum(len(p["tracks"]) for p in playlists)
+    logger.info("fetched {} playlist(s), {} tracks", len(playlists), total)
+
+    collection.refresh_index()
+    logger.info(
+        "existing in collection: {} strict / {} loose keys",
+        len(collection.index.by_strict),
+        len(collection.index.by_loose),
+    )
+
+    missing = collection.find_missing(playlists)
+    logger.info("missing tracks to download: {}", len(missing))
+
+    # Issues accumulate from both the streaming callback (per-track on
+    # completion) and the end-of-run validate sweep. process_track dedups
+    # paths so neither path double-reports.
+    issues: list[TrackIssue] = []
+
+    if missing and not args.no_download:
+        # (artist_norm, title_norm) -> [(playlist_dict, spotify_track)] so the
+        # callback can both rewrite the right m3u8 and reconcile tags against
+        # the canonical Spotify entry (sockseek echoes back only artist/title).
+        track_to_playlists: dict[tuple[str, str], list[tuple[dict, dict]]] = {}
+        for pl in playlists:
+            for t in pl["tracks"]:
+                key = (
+                    norm(first_artist(", ".join(t["artists"]))),
+                    norm(t["title"]),
+                )
+                track_to_playlists.setdefault(key, []).append((pl, t))
+
+        def _human_size(n: int) -> str:
+            for unit in ("B", "KiB", "MiB", "GiB"):
+                if n < 1024:
+                    return f"{n:.0f}{unit}" if unit == "B" else f"{n:.1f}{unit}"
+                n /= 1024
+            return f"{n:.1f}TiB"
+
+        async def on_event(evt: dict) -> None:
+            etype = evt.get("type")
+            data = evt.get("data") or {}
+
+            if etype == "search_start":
+                logger.info(
+                    "search: {} - {}",
+                    data.get("artist") or "?", data.get("title") or "?",
+                )
+                return
+
+            if etype == "download_start":
+                size = data.get("size") or 0
+                logger.info(
+                    "downloading: {} - {} ({} {}, from {})",
+                    data.get("artist") or "?",
+                    data.get("title") or "?",
+                    _human_size(size) if size else "?",
+                    (data.get("extension") or "").lstrip(".") or "?",
+                    data.get("username") or "?",
+                )
+                return
+
+            if etype != "track_state":
+                return
+
+            outcome = (data.get("terminalOutcome") or "").lower()
+            skip_reason = (data.get("skipReason") or "").lower()
+            failure_reason = (data.get("failureReason") or "").strip()
+            artist = data.get("artist") or ""
+            title = data.get("title") or ""
+            success = outcome == "succeeded" or (
+                outcome == "skipped" and skip_reason == "alreadyexists"
+            )
+            if not success:
+                # Surface failures/skips so the user sees why a track didn't land.
+                if outcome == "skipped":
+                    logger.info("skipped: {} - {} ({})", artist, title, skip_reason or "skipped")
+                else:
+                    logger.warning(
+                        "failed: {} - {} ({})",
+                        artist, title, failure_reason or outcome or "unknown",
+                    )
+                return
+
+            download_path = data.get("downloadPath")
+            if not download_path:
+                return
+            path = Path(download_path)
+            if not path.exists():
+                logger.debug("track_state success but path missing: {}", download_path)
+                return
+            collection.index.add(artist, title, path)
+            key = (norm(first_artist(artist)), norm(title))
+            affected = track_to_playlists.get(key, [])
+            if not affected:
+                logger.info("downloaded {} - {} ({})", artist, title, path.name)
+                return
+            # Reconcile against the Spotify track (writes Spotify tags
+            # + records any mismatches). One spotify track might appear
+            # in several playlists; process_track dedups via collection._reconciled.
+            spotify_track = affected[0][1]
+            new_issues = collection.process_track(
+                affected[0][0]["name"] or affected[0][0]["id"],
+                spotify_track,
+                path,
+            )
+            for issue in new_issues:
+                logger.warning(
+                    "{} - {}: {}", issue.artist, issue.title, issue.reason,
+                )
+            issues.extend(new_issues)
+            for pl, _ in affected:
+                out = collection.write_m3u8_for(pl)
+                logger.info(
+                    "downloaded {} - {} → updated {}",
+                    artist, title, out.name,
+                )
+
+        rc = await collection.download(missing, on_event=on_event)
+        if rc != 0:
+            logger.warning("sockseek exited with {}", rc)
+        collection.refresh_index()
+
+    if not args.no_validate:
+        timeout = aiohttp.ClientTimeout(total=30)
+        async with aiohttp.ClientSession(timeout=timeout) as session:
+            issues.extend(await collection.validate(session, playlists))
+
+    for out in collection.write_m3u8s(playlists, issues):
+        logger.info("wrote {}", out)
+
+    collection.report_issues(issues)
+    logger.info("{} issue(s) recorded in {}", len(issues), collection.report_path)
+    return 0
+
+
+def main() -> int:
+    return asyncio.run(main_async(sys.argv[1:]))

maple-py/src/maple/collection/core.py

@@ -0,0 +1,582 @@
+"""`Collection` dataclass: the on-disk music collection rooted at `root`.
+
+Owns dir layout, the in-memory `CollectionIndex` of (artist, title) →
+file paths, startup validation (drop unfinished downloads, prune the
+sldl index, prune stale m3u8 entries), the missing-track diff, validation
+of downloaded tracks, the per-playlist m3u8 writer, and the loguru file
+sink used for `collection-report.txt`.
+
+Sockseek subprocess plumbing lives in `maple.collection.sockseek`; this
+module just calls into it from `Collection.download()`.
+"""
+from __future__ import annotations
+
+import asyncio
+import csv
+import os
+import re
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import aiohttp
+from loguru import logger
+from mutagen import File as MutagenFile
+
+from maple.collection.index import (
+    CollectionIndex,
+    first_artist,
+    fuzzy_ratio,
+    norm,
+    norm_loose,
+)
+from maple.collection.sockseek import download as run_sockseek
+from maple.collection.validate import (
+    ARTWORK_CONCURRENCY,
+    DURATION_TOLERANCE_S,
+    embed_art,
+    fetch_jpeg,
+    ffprobe_ok,
+    file_duration,
+    has_embedded_art,
+)
+from maple.scan import AUDIO_EXTS
+
+
+@dataclass
+class TrackIssue:
+    playlist: str
+    artist: str
+    title: str
+    reason: str
+
+
+_M3U_BAD = re.compile(r"[\\/\x00]+")
+
+
+def _sanitise_name(name: str) -> str:
+    s = _M3U_BAD.sub("_", name or "playlist").strip().lstrip(".")
+    return (s or "playlist")[:200]
+
+
+@dataclass
+class Collection:
+    """A music collection rooted at `root` on disk."""
+
+    root: Path
+    index: CollectionIndex = field(default_factory=CollectionIndex)
+    # Paths whose tags have already been reconciled against Spotify this run.
+    # `process_track` consults this so the on_event hot path and the
+    # end-of-run `validate` sweep don't double-handle the same file.
+    _reconciled: set[Path] = field(default_factory=set, repr=False)
+
+    # --- paths ------------------------------------------------------------- #
+
+    @classmethod
+    def from_repo(cls) -> "Collection":
+        """Default location: `<repo>/data/collection`."""
+        here = Path(__file__).resolve()
+        # src/maple/collection/core.py → src/maple/collection → src/maple → src → maple-py → repo
+        return cls((here.parents[4] / "data" / "collection").resolve())
+
+    @property
+    def metadata_dir(self) -> Path:
+        return self.root / "_metadata"
+
+    @property
+    def playlists_dir(self) -> Path:
+        return self.root / "_playlists"
+
+    @property
+    def index_csv(self) -> Path:
+        return self.metadata_dir / "index.sldl.csv"
+
+    @property
+    def report_path(self) -> Path:
+        return self.metadata_dir / "collection-report.txt"
+
+    def ensure_dirs(self) -> None:
+        self.root.mkdir(parents=True, exist_ok=True)
+        self.metadata_dir.mkdir(parents=True, exist_ok=True)
+        self.playlists_dir.mkdir(parents=True, exist_ok=True)
+
+    # --- startup validation ----------------------------------------------- #
+
+    def remove_unfinished_downloads(self) -> int:
+        """Delete sldl's `*.incomplete` leftover files. Returns count removed."""
+        if not self.root.exists():
+            return 0
+        count = 0
+        for p in self.root.rglob("*.incomplete"):
+            if "_metadata" in p.parts or "_playlists" in p.parts:
+                continue
+            try:
+                p.unlink()
+                count += 1
+            except OSError as e:
+                logger.warning("failed to remove {}: {}", p, e)
+        return count
+
+    def validate_index_file(self) -> tuple[int, int]:
+        """Rewrite `index.sldl.csv` dropping duplicate rows and orphaned-path rows.
+
+        Returns (kept, removed). Failure rows (no filepath) are kept so
+        sldl can still honour `--skip-not-found` on the next run.
+        """
+        if not self.index_csv.exists():
+            return (0, 0)
+        with self.index_csv.open(newline="", encoding="utf-8", errors="replace") as fh:
+            rows = list(csv.reader(fh))
+        if len(rows) <= 1:
+            return (max(0, len(rows) - 1), 0)
+        header = rows[0]
+        cols = {h.strip().lower(): i for i, h in enumerate(header)}
+        path_i = cols.get("filepath", 0)
+
+        kept: list[list[str]] = []
+        seen: set[tuple[str, ...]] = set()
+        removed = 0
+        for row in rows[1:]:
+            key = tuple(row)
+            if key in seen:
+                removed += 1
+                continue
+            seen.add(key)
+            rel = row[path_i] if path_i < len(row) else ""
+            if rel:
+                p = self.root / rel if not Path(rel).is_absolute() else Path(rel)
+                if not p.exists():
+                    removed += 1
+                    continue
+            kept.append(row)
+
+        if removed > 0:
+            with self.index_csv.open("w", newline="", encoding="utf-8") as fh:
+                w = csv.writer(fh)
+                w.writerow(header)
+                w.writerows(kept)
+        return (len(kept), removed)
+
+    def validate_playlists(self) -> tuple[int, int]:
+        """Sweep `_playlists/*.m3u8`. Two passes per entry:
+
+        1. Lines whose file doesn't resolve are replaced with a
+           `# NOT FOUND: <title>` comment so the playlist still tells you
+           what's missing.
+        2. Lines that resolve but were written with an absolute path get
+           rewritten relative to the m3u8's parent dir — playlists need to
+           survive being synced to another machine where the collection
+           sits at a different root.
+
+        Returns (pruned, rewritten)."""
+        if not self.playlists_dir.exists():
+            return (0, 0)
+        total_pruned = 0
+        total_rewritten = 0
+        for m3u in self.playlists_dir.glob("*.m3u8"):
+            lines = m3u.read_text(encoding="utf-8").splitlines()
+            out: list[str] = []
+            pending_extinf: str | None = None
+            pruned_here = 0
+            rewritten_here = 0
+            for line in lines:
+                s = line.strip()
+                if not s:
+                    out.append(line)
+                    continue
+                if s.startswith("#EXTINF"):
+                    pending_extinf = line
+                    continue
+                if s.startswith("#"):
+                    if pending_extinf is not None:
+                        out.append(pending_extinf)
+                        pending_extinf = None
+                    out.append(line)
+                    continue
+                # Bare file path (relative to m3u or absolute).
+                was_absolute = Path(s).is_absolute()
+                target = Path(s) if was_absolute else (m3u.parent / s)
+                if target.exists():
+                    if pending_extinf is not None:
+                        out.append(pending_extinf)
+                        pending_extinf = None
+                    if was_absolute:
+                        rel = os.path.relpath(target.resolve(), start=m3u.parent.resolve())
+                        out.append(rel)
+                        rewritten_here += 1
+                    else:
+                        out.append(line)
+                else:
+                    desc = s
+                    if pending_extinf and "," in pending_extinf:
+                        desc = pending_extinf.split(",", 1)[1].strip() or s
+                    out.append(f"# NOT FOUND: {desc}")
+                    pending_extinf = None
+                    pruned_here += 1
+            if pruned_here or rewritten_here:
+                m3u.write_text("\n".join(out) + "\n", encoding="utf-8")
+                if pruned_here:
+                    logger.warning("pruned {} missing entries from {}", pruned_here, m3u.name)
+                if rewritten_here:
+                    logger.warning("rewrote {} absolute path(s) → relative in {}", rewritten_here, m3u.name)
+                total_pruned += pruned_here
+                total_rewritten += rewritten_here
+        return (total_pruned, total_rewritten)
+
+    # --- report sink ------------------------------------------------------- #
+
+    def attach_report_sink(self) -> int:
+        """Install a loguru sink that mirrors every reported issue to disk.
+
+        Filters on `extra["report"] is True` so unrelated log lines don't
+        leak into the report. Returns the loguru handle so callers can
+        detach it later.
+        """
+        self.metadata_dir.mkdir(parents=True, exist_ok=True)
+        if self.report_path.exists():
+            self.report_path.unlink()
+        return logger.add(
+            self.report_path,
+            format="{message}",
+            filter=lambda r: r["extra"].get("report") is True,
+            level="INFO",
+        )
+
+    def report_issues(self, issues: list[TrackIssue]) -> None:
+        """Emit each issue through the report sink; emit `ok` if none."""
+        bound = logger.bind(report=True)
+        if not issues:
+            bound.info("ok")
+            return
+        for i in issues:
+            bound.warning(f"[{i.playlist}] {i.artist} - {i.title}: {i.reason}")
+
+    # --- indexing ---------------------------------------------------------- #
+
+    def refresh_index(self) -> CollectionIndex:
+        """Rebuild `self.index` from sldl's csv + an on-disk tag scan."""
+        idx = CollectionIndex()
+        self._load_sldl_index(idx)
+        self._scan_collection_tags(idx)
+        self.index = idx
+        return idx
+
+    def _load_sldl_index(self, idx: CollectionIndex) -> None:
+        if not self.index_csv.exists():
+            return
+        with self.index_csv.open(newline="", encoding="utf-8", errors="replace") as fh:
+            reader = csv.reader(fh)
+            try:
+                header = next(reader)
+            except StopIteration:
+                return
+            cols = {h.strip().lower(): i for i, h in enumerate(header)}
+            path_i = cols.get("filepath") or cols.get("path") or 0
+            artist_i = cols.get("artist")
+            title_i = cols.get("title")
+            state_i = cols.get("state")
+            for row in reader:
+                if not row:
+                    continue
+                if state_i is not None and state_i < len(row):
+                    state = row[state_i].strip().lower()
+                    if state in {"2", "failed", "notfounderror", "error"}:
+                        continue
+                try:
+                    rel = row[path_i]
+                except IndexError:
+                    continue
+                if not rel:
+                    continue
+                path = self.root / rel if not Path(rel).is_absolute() else Path(rel)
+                if not path.exists():
+                    continue
+                artist = row[artist_i] if artist_i is not None and artist_i < len(row) else ""
+                title = row[title_i] if title_i is not None and title_i < len(row) else ""
+                idx.add(artist, title, path)
+
+    def _scan_collection_tags(self, idx: CollectionIndex) -> None:
+        if not self.root.exists():
+            return
+        for p in self.root.rglob("*"):
+            if not p.is_file() or p.suffix.lower() not in AUDIO_EXTS:
+                continue
+            if "_metadata" in p.parts or "_playlists" in p.parts:
+                continue
+            try:
+                audio = MutagenFile(p, easy=True)
+            except Exception:
+                continue
+            if audio is None:
+                continue
+            tags = audio.tags or {}
+            artist = (tags.get("artist") or [""])[0] if hasattr(tags, "get") else ""
+            title = (tags.get("title") or [""])[0] if hasattr(tags, "get") else ""
+            if not title:
+                title = p.stem
+            idx.add(str(artist), str(title), p)
+
+    # --- planning + downloads --------------------------------------------- #
+
+    def find_missing(self, playlists: list[dict]) -> list[dict]:
+        """Return the Spotify tracks not currently resolvable in this collection."""
+        seen: set[tuple[str, str]] = set()
+        missing: list[dict] = []
+        for pl in playlists:
+            for t in pl["tracks"]:
+                artist = ", ".join(t["artists"])
+                if self.index.lookup(artist, t["title"]) is not None:
+                    continue
+                key = (norm(first_artist(artist)), norm(t["title"]))
+                if key in seen:
+                    continue
+                seen.add(key)
+                missing.append(t)
+        return missing
+
+    async def download(
+        self,
+        missing: list[dict],
+        on_event: Callable[[dict], Awaitable[None]] | None = None,
+    ) -> int:
+        """Thin wrapper around `maple.collection.sockseek.download`."""
+        return await run_sockseek(self.root, self.index_csv, missing, on_event)
+
+    # --- per-track Spotify reconciliation --------------------------------- #
+
+    def compare_against_spotify(
+        self,
+        playlist_name: str,
+        track: dict,
+        path: Path,
+    ) -> list[TrackIssue]:
+        """Compare a downloaded file's tags + duration against Spotify.
+
+        Spotify is the source of truth. Each mismatched field yields a
+        `TrackIssue` whose `reason` reads as
+        `<field> mismatch: spotify=<sv> file=<fv>`. Empty/missing file
+        tags count as mismatches against a non-empty Spotify value.
+        """
+        sp_artist = ", ".join(track.get("artists") or [])
+        sp_title = (track.get("title") or "").strip()
+        sp_album = (track.get("album") or "").strip()
+        sp_isrc = (track.get("isrc") or "").strip()
+        sp_duration = (track.get("duration_ms") or 0) / 1000.0
+        sp_date = (track.get("release_date") or "").strip()
+
+        try:
+            audio = MutagenFile(path, easy=True)
+            tags = audio.tags if audio is not None else None
+        except Exception:
+            tags = None
+        f_artist = (tags.get("artist") or [""])[0] if tags and hasattr(tags, "get") else ""
+        f_title = (tags.get("title") or [""])[0] if tags and hasattr(tags, "get") else ""
+        f_album = (tags.get("album") or [""])[0] if tags and hasattr(tags, "get") else ""
+        f_isrc = (tags.get("isrc") or [""])[0] if tags and hasattr(tags, "get") else ""
+        f_date = (tags.get("date") or [""])[0] if tags and hasattr(tags, "get") else ""
+        f_duration = file_duration(path)
+
+        issues: list[TrackIssue] = []
+        def mismatch(field: str, sv: object, fv: object) -> None:
+            issues.append(TrackIssue(
+                playlist_name, sp_artist, sp_title,
+                f"{field} mismatch: spotify={sv!r} file={fv!r} ({path.name})",
+            ))
+
+        # Artist + title use fuzzy compare because tags & spotify quote
+        # remixers / features differently. Album + ISRC + date are exact-ish.
+        if sp_artist and fuzzy_ratio(norm(first_artist(sp_artist)), norm(first_artist(f_artist))) < 0.5:
+            mismatch("artist", sp_artist, f_artist)
+        if sp_title and fuzzy_ratio(norm_loose(sp_title), norm_loose(f_title)) < 0.5:
+            mismatch("title", sp_title, f_title)
+        if sp_album and f_album and fuzzy_ratio(norm_loose(sp_album), norm_loose(f_album)) < 0.5:
+            mismatch("album", sp_album, f_album)
+        elif sp_album and not f_album:
+            mismatch("album", sp_album, f_album)
+        if sp_isrc and f_isrc and sp_isrc.upper() != f_isrc.upper():
+            mismatch("isrc", sp_isrc, f_isrc)
+        # Compare only the year portion of `date` — file tags are usually "YYYY"
+        # while Spotify is "YYYY[-MM[-DD]]".
+        if sp_date and f_date and sp_date[:4] != f_date[:4]:
+            mismatch("date", sp_date, f_date)
+        if sp_duration > 0 and abs(f_duration - sp_duration) > DURATION_TOLERANCE_S:
+            mismatch("duration", f"{sp_duration:.1f}s", f"{f_duration:.1f}s")
+
+        return issues
+
+    def write_spotify_tags(self, track: dict, path: Path) -> None:
+        """Overwrite the file's tags with the Spotify values.
+
+        Spotify is canonical: title, artist (joined ", "), album, date,
+        genre(s), and ISRC are written. Missing Spotify values leave the
+        existing tag untouched.
+        """
+        try:
+            audio = MutagenFile(path, easy=True)
+        except Exception as e:
+            logger.warning("write_spotify_tags: cannot open {}: {}", path.name, e)
+            return
+        if audio is None:
+            return
+        if audio.tags is None:
+            try:
+                audio.add_tags()
+            except Exception as e:
+                logger.warning("write_spotify_tags: cannot init tags on {}: {}", path.name, e)
+                return
+
+        title = (track.get("title") or "").strip()
+        if title:
+            audio.tags["title"] = title
+        artists = track.get("artists") or []
+        if artists:
+            audio.tags["artist"] = ", ".join(artists)
+        album = (track.get("album") or "").strip()
+        if album:
+            audio.tags["album"] = album
+        date = (track.get("release_date") or "").strip()
+        if date:
+            try:
+                audio.tags["date"] = date
+            except (KeyError, ValueError) as e:
+                logger.debug("date tag write skipped on {}: {}", path.name, e)
+        genres = track.get("genres") or []
+        if genres:
+            try:
+                audio.tags["genre"] = list(genres)
+            except (KeyError, ValueError) as e:
+                logger.debug("genre tag write skipped on {}: {}", path.name, e)
+        isrc = (track.get("isrc") or "").strip()
+        if isrc:
+            try:
+                audio.tags["isrc"] = isrc
+            except (KeyError, ValueError) as e:
+                logger.debug("isrc tag write skipped on {}: {}", path.name, e)
+
+        try:
+            audio.save()
+        except Exception as e:
+            logger.warning("write_spotify_tags: save failed for {}: {}", path.name, e)
+
+    def process_track(
+        self,
+        playlist_name: str,
+        track: dict,
+        path: Path,
+    ) -> list[TrackIssue]:
+        """Reconcile a single downloaded file against Spotify.
+
+        Steps:
+          1. ffprobe sanity check (one issue + early return on failure).
+          2. Compare every relevant field; collect mismatch issues.
+          3. Write Spotify's values back into the file's tags.
+
+        Idempotent within a run — paths already processed are no-ops.
+        """
+        if path in self._reconciled:
+            return []
+        self._reconciled.add(path)
+
+        sp_artist = ", ".join(track.get("artists") or [])
+        sp_title = track.get("title") or ""
+        if not ffprobe_ok(path):
+            return [TrackIssue(playlist_name, sp_artist, sp_title, f"ffprobe failed: {path}")]
+
+        issues = self.compare_against_spotify(playlist_name, track, path)
+        self.write_spotify_tags(track, path)
+        return issues
+
+    # --- validation -------------------------------------------------------- #
+
+    async def validate(
+        self,
+        session: aiohttp.ClientSession,
+        playlists: list[dict],
+    ) -> list[TrackIssue]:
+        """Per-track Spotify reconciliation + artwork embed.
+
+        Spotify is the source of truth: any field that differs from the
+        downloaded file's tags is recorded as a mismatch issue, then the
+        file's tags are overwritten with the Spotify values. Missing
+        embedded artwork is queued for an async batch fetch.
+        """
+        resolved: list[tuple[str, dict, Path]] = []
+        for pl in playlists:
+            for t in pl["tracks"]:
+                path = self.index.lookup(", ".join(t["artists"]), t["title"])
+                if path is not None:
+                    resolved.append((pl["name"] or pl["id"], t, path))
+
+        issues: list[TrackIssue] = []
+        sem = asyncio.Semaphore(ARTWORK_CONCURRENCY)
+        art_jobs: list[tuple[str, dict, Path]] = []
+
+        for playlist_name, track, path in resolved:
+            issues.extend(self.process_track(playlist_name, track, path))
+            if not has_embedded_art(path) and track.get("album_image_url"):
+                art_jobs.append((playlist_name, track, path))
+
+        async def _embed(job: tuple[str, dict, Path]) -> None:
+            playlist_name, track, path = job
+            async with sem:
+                data = await fetch_jpeg(session, track["album_image_url"])
+            if not data:
+                issues.append(TrackIssue(
+                    playlist_name, ", ".join(track["artists"]), track["title"],
+                    "artwork download failed",
+                ))
+                return
+            try:
+                embed_art(path, data)
+            except Exception as e:
+                issues.append(TrackIssue(
+                    playlist_name, ", ".join(track["artists"]), track["title"],
+                    f"artwork embed failed: {e}",
+                ))
+
+        if art_jobs:
+            await asyncio.gather(*(_embed(j) for j in art_jobs))
+        return issues
+
+    # --- output ------------------------------------------------------------ #
+
+    def write_m3u8_for(
+        self,
+        playlist: dict,
+        issues: list[TrackIssue] | None = None,
+    ) -> Path:
+        """Write a single playlist's m3u8 against the current index.
+
+        Track paths are emitted relative to the m3u8's parent directory so
+        the playlist is portable across devices that sync the collection.
+
+        Pass `issues=None` for incremental rewrites during download — the
+        list of unresolved tracks gets rebuilt on the final pass instead.
+        """
+        self.playlists_dir.mkdir(parents=True, exist_ok=True)
+        playlist_name = playlist["name"] or playlist["id"]
+        out_path = self.playlists_dir / f"{_sanitise_name(playlist_name)}.m3u8"
+        lines = ["#EXTM3U"]
+        for t in playlist["tracks"]:
+            artist = ", ".join(t["artists"])
+            title = t["title"]
+            secs = round(t["duration_ms"] / 1000)
+            path = self.index.lookup(artist, title)
+            if path is None:
+                lines.append(f"# NOT FOUND: {artist} - {title}")
+                if issues is not None:
+                    issues.append(TrackIssue(playlist_name, artist, title, "unresolved in collection"))
+                continue
+            rel = os.path.relpath(path.resolve(), start=out_path.parent.resolve())
+            lines.append(f"#EXTINF:{secs},{artist} - {title}")
+            lines.append(rel)
+        out_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+        return out_path
+
+    def write_m3u8s(
+        self,
+        playlists: list[dict],
+        issues: list[TrackIssue],
+    ) -> list[Path]:
+        """One m3u8 per playlist in `<root>/_playlists/`. Returns the paths written."""
+        return [self.write_m3u8_for(pl, issues) for pl in playlists]

maple-py/src/maple/collection/index.py

@@ -0,0 +1,81 @@
+"""Artist/title normalisation + `CollectionIndex` lookup table.
+
+No I/O. Pure functions plus a small dataclass keyed on normalised
+(artist, title) tuples so we can answer "is this track already in the
+collection?" against an in-memory map.
+"""
+from __future__ import annotations
+
+import re
+import unicodedata
+from dataclasses import dataclass, field
+from pathlib import Path
+
+FUZZY_THRESHOLD = 0.85
+
+_PARENS = re.compile(r"\s*[\(\[][^)\]]*[\)\]]\s*")
+_NONWORD = re.compile(r"[^\w\s]+", flags=re.UNICODE)
+_WS = re.compile(r"\s+")
+
+
+def norm(s: str) -> str:
+    """Lowercase + NFKC + strip non-word, preserves remix/version suffixes."""
+    s = unicodedata.normalize("NFKC", s or "").lower()
+    s = _NONWORD.sub(" ", s)
+    s = _WS.sub(" ", s).strip()
+    return s
+
+
+def norm_loose(s: str) -> str:
+    """Same as norm() but with parenthesised suffixes removed."""
+    s = unicodedata.normalize("NFKC", s or "").lower()
+    s = _PARENS.sub(" ", s)
+    s = _NONWORD.sub(" ", s)
+    s = _WS.sub(" ", s).strip()
+    return s
+
+
+def first_artist(artists: str) -> str:
+    """Split sldl/spotify multi-artist strings on common delimiters."""
+    if not artists:
+        return ""
+    for sep in (",", " feat.", " ft.", " & ", "&", " and ", ";"):
+        if sep in artists:
+            return artists.split(sep, 1)[0].strip()
+    return artists.strip()
+
+
+def fuzzy_ratio(a: str, b: str) -> float:
+    """Cheap token-overlap ratio in [0, 1]. No external dep."""
+    ta, tb = set(a.split()), set(b.split())
+    if not ta or not tb:
+        return 0.0
+    return len(ta & tb) / max(len(ta), len(tb))
+
+
+@dataclass
+class CollectionIndex:
+    """Maps normalised (artist, title) pairs to file paths in the collection."""
+
+    by_strict: dict[tuple[str, str], Path] = field(default_factory=dict)
+    by_loose: dict[tuple[str, str], Path] = field(default_factory=dict)
+
+    def add(self, artist: str, title: str, path: Path) -> None:
+        a, t = norm(first_artist(artist)), norm(title)
+        if a and t:
+            self.by_strict.setdefault((a, t), path)
+        la, lt = norm_loose(first_artist(artist)), norm_loose(title)
+        if la and lt:
+            self.by_loose.setdefault((la, lt), path)
+
+    def lookup(self, artist: str, title: str) -> Path | None:
+        a, t = norm(first_artist(artist)), norm(title)
+        if (a, t) in self.by_strict:
+            return self.by_strict[(a, t)]
+        la, lt = norm_loose(first_artist(artist)), norm_loose(title)
+        if (la, lt) in self.by_loose:
+            return self.by_loose[(la, lt)]
+        for (ka, kt), p in self.by_loose.items():
+            if ka == la and fuzzy_ratio(kt, lt) >= FUZZY_THRESHOLD:
+                return p
+        return None

maple-py/src/maple/collection/sockseek.py

@@ -0,0 +1,149 @@
+"""Sockseek subprocess plumbing.
+
+Owns:
+    * the CLI argument shape (`args_for`) used to run sockseek/sldl,
+    * the temp-CSV writer that feeds tracks in,
+    * the async `download()` entrypoint that streams `--progress-json`
+      events out and forwards sockseek stderr through loguru.
+
+`Collection.download()` is a thin wrapper around `download()` here so the
+subprocess concerns stay in one file.
+"""
+from __future__ import annotations
+
+import asyncio
+import csv
+import json
+import os
+import re
+import shutil
+import sys
+import tempfile
+from collections.abc import Awaitable, Callable
+from pathlib import Path
+
+from loguru import logger
+
+NAME_FORMAT = "{artist(/)album(/)track( )title|(_missing_tags/)slsk-foldername(/)slsk-filename}"
+
+_ANSI_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")
+_LEVEL_RE = re.compile(
+    r"^\s*\[?(?P<lvl>TRACE|TRC|DEBUG|DBG|INFO|INF|WARN|WRN|ERROR|ERR|FATAL|CRITICAL|CRT)\]?\b",
+    re.IGNORECASE,
+)
+_LEVEL_MAP = {
+    "TRC": "TRACE", "TRACE": "TRACE",
+    "DBG": "DEBUG", "DEBUG": "DEBUG",
+    "INF": "INFO",  "INFO":  "INFO",
+    "WRN": "WARNING", "WARN": "WARNING",
+    "ERR": "ERROR", "ERROR": "ERROR",
+    "FATAL": "CRITICAL", "CRT": "CRITICAL", "CRITICAL": "CRITICAL",
+}
+
+
+def log_level_for(line: str) -> str:
+    """Map a sockseek/.NET-style log prefix to a loguru level. Default WARNING."""
+    m = _LEVEL_RE.match(line)
+    if not m:
+        return "WARNING"
+    return _LEVEL_MAP.get(m.group("lvl").upper(), "WARNING")
+
+
+def args_for(root: Path, index_csv: Path, user: str, password: str) -> list[str]:
+    """Mirror of src/sockseek.py:config_args(), pinned for FLAC-strict."""
+    return [
+        "--path", str(root),
+        "--index-path", str(index_csv),
+        "--user", user,
+        "--pass", password,
+        "--name-format", NAME_FORMAT,
+        "--format", "flac",
+        "--pref-format", "flac",
+        "--pref-min-bitrate", "200",
+        "--pref-max-bitrate", "2500",
+        "--strict-conditions",
+        "--fast-search",
+        "--concurrent-jobs", "4",
+    ]
+
+
+def write_download_csv(missing: list[dict], fh) -> None:
+    """sldl-compatible CSV (Artist,Title,Album,Length)."""
+    w = csv.writer(fh)
+    w.writerow(["Artist", "Title", "Album", "Length"])
+    for t in missing:
+        w.writerow([
+            ", ".join(t["artists"]),
+            t["title"],
+            t["album"],
+            round(t["duration_ms"] / 1000),
+        ])
+
+
+async def download(
+    root: Path,
+    index_csv: Path,
+    missing: list[dict],
+    on_event: Callable[[dict], Awaitable[None]] | None = None,
+) -> int:
+    """Shell out to sockseek for every track in `missing`.
+
+    When `on_event` is provided, sockseek is run with `--progress-json` and
+    every NDJSON envelope on stdout is parsed and awaited through the
+    callback. stderr is always piped and forwarded line-by-line through
+    loguru with a best-effort log-level guess.
+    """
+    if not missing:
+        return 0
+    binary = shutil.which("sockseek") or shutil.which("sldl")
+    if binary is None:
+        logger.error("sockseek/sldl not on PATH; install slsk-batchdl first")
+        sys.exit(1)
+    user = os.environ.get("SOCKSEEK_USER")
+    password = os.environ.get("SOCKSEEK_PASS")
+    if not (user and password):
+        logger.error("set SOCKSEEK_USER and SOCKSEEK_PASS for downloads")
+        sys.exit(1)
+    index_csv.parent.mkdir(parents=True, exist_ok=True)
+    with tempfile.NamedTemporaryFile("w", suffix=".csv", newline="", delete=False) as tf:
+        write_download_csv(missing, tf)
+        tf_path = Path(tf.name)
+    try:
+        args = [binary, str(tf_path), *args_for(root, index_csv, user, password)]
+        if on_event is not None:
+            args.append("--progress-json")
+        masked = ["***" if a == password else a for a in args]
+        logger.info("$ {}", " ".join(masked))
+        proc = await asyncio.create_subprocess_exec(
+            *args,
+            stdout=asyncio.subprocess.PIPE if on_event is not None else None,
+            stderr=asyncio.subprocess.PIPE,
+        )
+
+        async def _forward_stderr() -> None:
+            assert proc.stderr is not None
+            async for raw in proc.stderr:
+                line = _ANSI_RE.sub("", raw.decode("utf-8", errors="replace")).rstrip()
+                if not line:
+                    continue
+                logger.log(log_level_for(line), "sockseek: {}", line)
+
+        stderr_task = asyncio.create_task(_forward_stderr())
+
+        if on_event is not None:
+            assert proc.stdout is not None
+            async for raw in proc.stdout:
+                line = raw.decode("utf-8", errors="replace").strip()
+                if not line or not line.startswith("{"):
+                    continue
+                try:
+                    evt = json.loads(line)
+                except ValueError:
+                    continue
+                await on_event(evt)
+
+        rc = await proc.wait()
+        await stderr_task
+        return rc
+    finally:
+        tf_path.unlink(missing_ok=True)

maple-py/src/maple/collection/validate.py

@@ -0,0 +1,103 @@
+"""Per-track validation helpers: ffprobe sanity, duration probe, embedded
+artwork inspection + embed, async JPEG fetch for missing cover art.
+
+Pure helpers — no Collection coupling.
+"""
+from __future__ import annotations
+
+import io
+import shutil
+import subprocess
+from pathlib import Path
+
+import aiohttp
+from mutagen import File as MutagenFile
+from mutagen.flac import FLAC, Picture
+from mutagen.id3 import APIC, ID3, ID3NoHeaderError
+from mutagen.mp4 import MP4, MP4Cover
+
+DURATION_TOLERANCE_S = 5.0
+ARTWORK_CONCURRENCY = 8
+
+
+def ffprobe_ok(path: Path) -> bool:
+    if not shutil.which("ffprobe"):
+        return True  # ffprobe absent → don't fail the run
+    r = subprocess.run(
+        ["ffprobe", "-v", "error", "-select_streams", "a:0",
+         "-show_entries", "stream=codec_type", "-of", "csv=p=0", str(path)],
+        capture_output=True, text=True,
+    )
+    return r.returncode == 0 and "audio" in r.stdout
+
+
+def file_duration(path: Path) -> float:
+    audio = MutagenFile(path)
+    info = getattr(audio, "info", None)
+    return float(getattr(info, "length", 0.0) or 0.0)
+
+
+def has_embedded_art(path: Path) -> bool:
+    suf = path.suffix.lower()
+    if suf == ".flac":
+        try:
+            return bool(FLAC(path).pictures)
+        except Exception:
+            return False
+    if suf == ".mp3":
+        try:
+            return bool(ID3(path).getall("APIC"))
+        except Exception:
+            return False
+    if suf in (".m4a", ".mp4"):
+        try:
+            tags = MP4(path).tags
+            return bool(tags and tags.get("covr"))
+        except Exception:
+            return False
+    return True  # unknown format → don't complain
+
+
+def embed_art(path: Path, jpeg_bytes: bytes) -> None:
+    suf = path.suffix.lower()
+    if suf == ".flac":
+        f = FLAC(path)
+        pic = Picture()
+        pic.data = jpeg_bytes
+        pic.type = 3  # FRONT_COVER
+        pic.mime = "image/jpeg"
+        f.clear_pictures()
+        f.add_picture(pic)
+        f.save()
+    elif suf == ".mp3":
+        try:
+            tags = ID3(path)
+        except ID3NoHeaderError:
+            tags = ID3()
+        tags.delall("APIC")
+        tags.add(APIC(encoding=3, mime="image/jpeg", type=3, desc="Cover", data=jpeg_bytes))
+        tags.save(path)
+    elif suf in (".m4a", ".mp4"):
+        f = MP4(path)
+        f.tags = f.tags or {}
+        f["covr"] = [MP4Cover(jpeg_bytes, imageformat=MP4Cover.FORMAT_JPEG)]
+        f.save()
+
+
+async def fetch_jpeg(session: aiohttp.ClientSession, url: str) -> bytes | None:
+    try:
+        async with session.get(url) as resp:
+            resp.raise_for_status()
+            data = await resp.read()
+    except aiohttp.ClientError:
+        return None
+    if data[:3] == b"\xff\xd8\xff":
+        return data
+    try:
+        from PIL import Image
+        img = Image.open(io.BytesIO(data)).convert("RGB")
+        buf = io.BytesIO()
+        img.save(buf, format="JPEG", quality=90)
+        return buf.getvalue()
+    except Exception:
+        return None

maple-py/src/maple/log.py

@@ -0,0 +1,20 @@
+"""Loguru sink config for maple's CLIs.
+
+`configure(level)` (re)installs a single stderr sink with a compact format.
+Idempotent — safe to call from every CLI entrypoint.
+"""
+import os
+import sys
+
+from loguru import logger
+
+_FMT = "<green>{time:HH:mm:ss}</green> <level>{level:<7}</level> {message}"
+
+
+def configure(level: str | None = None) -> None:
+    level = (level or os.environ.get("MAPLE_LOG_LEVEL") or "INFO").upper()
+    logger.remove()
+    logger.add(sys.stderr, format=_FMT, level=level, colorize=True, enqueue=False)
+
+
+__all__ = ["configure", "logger"]

maple-py/src/maple/spotify.py

@@ -0,0 +1,249 @@
+"""Dump tracks from one or more Spotify playlists (Client Credentials flow).
+
+    just spotify                       # uses the placeholder ids below
+    just spotify "<url-or-id> ..."     # explicit playlists
+    just spotify "--file lists.txt --json"
+    just spotify "<url-or-id> --csv"   # single sockseek CSV (stdout)
+    just spotify "<url-or-id> --csv --out tracks.csv"
+
+Playlists can be ids, full URLs, or spotify: URIs. Works for any public
+playlist; no user login needed. Requires SPOTIFY_CLIENT_ID /
+SPOTIFY_CLIENT_SECRET in the environment (loaded from .env by just).
+
+--csv writes one combined CSV with Artist,Title,Album,Length columns (Length
+in seconds), the format consumed by sockseek/sldl. Each playlist is preceded
+by a `# <name>` comment line as a separator. Writes to stdout, or to a file
+with --out <path>.
+
+Networking is async via aiohttp: playlists are fetched concurrently.
+"""
+import asyncio
+import base64
+import csv
+import json
+import os
+import re
+import sys
+import urllib.parse
+
+import aiohttp
+
+
+async def get_token(session: aiohttp.ClientSession, client_id: str, client_secret: str) -> str:
+    auth = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
+    async with session.post(
+        "https://accounts.spotify.com/api/token",
+        data={"grant_type": "client_credentials"},
+        headers={"Authorization": f"Basic {auth}"},
+    ) as resp:
+        resp.raise_for_status()
+        body = await resp.json()
+    return body["access_token"]
+
+
+async def api_get(session: aiohttp.ClientSession, url: str, token: str) -> dict:
+    async with session.get(url, headers={"Authorization": f"Bearer {token}"}) as resp:
+        resp.raise_for_status()
+        return await resp.json()
+
+
+def parse_playlist_id(arg: str) -> str:
+    m = re.search(r"playlist[/:]([A-Za-z0-9]+)", arg)
+    return m.group(1) if m else arg.strip()
+
+
+def read_list_file(path: str) -> list[str]:
+    out = []
+    with open(path) as fh:
+        for line in fh:
+            line = line.split("#", 1)[0].strip()
+            if line:
+                out.append(line)
+    return out
+
+
+def _largest_image(images: list[dict] | None) -> str | None:
+    if not images:
+        return None
+    # Spotify returns images sorted largest first, but defend against absent width.
+    ranked = sorted(images, key=lambda im: (im.get("width") or 0), reverse=True)
+    return ranked[0].get("url")
+
+
+async def fetch_playlist(session: aiohttp.ClientSession, pid: str, token: str) -> dict:
+    meta = await api_get(
+        session,
+        f"https://api.spotify.com/v1/playlists/{pid}"
+        f"?fields={urllib.parse.quote('name,owner(display_name)')}",
+        token,
+    )
+    fields = (
+        "items(track(id,name,artists(id,name),album(name,images,release_date,release_date_precision),"
+        "duration_ms,external_ids(isrc),track_number,disc_number)),next"
+    )
+    url = (
+        f"https://api.spotify.com/v1/playlists/{pid}/tracks"
+        f"?limit=100&fields={urllib.parse.quote(fields)}"
+    )
+    tracks = []
+    while url:
+        page = await api_get(session, url, token)
+        for item in page.get("items", []):
+            t = item.get("track")
+            if not t:
+                continue
+            tracks.append(
+                {
+                    "id": t.get("id"),
+                    "title": t["name"],
+                    "artists": [a["name"] for a in t["artists"]],
+                    "artist_ids": [a.get("id") for a in t["artists"] if a.get("id")],
+                    "album": t["album"]["name"],
+                    "album_image_url": _largest_image(t["album"].get("images")),
+                    "release_date": t["album"].get("release_date"),
+                    "release_date_precision": t["album"].get("release_date_precision"),
+                    "duration_ms": t["duration_ms"],
+                    "isrc": (t.get("external_ids") or {}).get("isrc"),
+                    "track_number": t.get("track_number"),
+                    "disc_number": t.get("disc_number"),
+                    "genres": [],
+                }
+            )
+        url = page.get("next")
+    return {"id": pid, "name": meta.get("name"), "tracks": tracks}
+
+
+async def fetch_artist_genres(
+    session: aiohttp.ClientSession,
+    artist_ids: list[str],
+    token: str,
+) -> dict[str, list[str]]:
+    """Batch `/v1/artists?ids=...` (max 50 ids per call). Returns id → genres."""
+    out: dict[str, list[str]] = {}
+    seen: set[str] = set()
+    deduped = [a for a in artist_ids if a and not (a in seen or seen.add(a))]
+    for i in range(0, len(deduped), 50):
+        batch = deduped[i:i + 50]
+        body = await api_get(
+            session,
+            f"https://api.spotify.com/v1/artists?ids={','.join(batch)}",
+            token,
+        )
+        for a in body.get("artists", []) or []:
+            if a and a.get("id"):
+                out[a["id"]] = a.get("genres") or []
+    return out
+
+
+def _enrich_track_genres(playlists: list[dict], genres_by_artist: dict[str, list[str]]) -> None:
+    """Populate each track's `genres` field with the union (dedup'd, preserving order)
+    of its artists' genres."""
+    for pl in playlists:
+        for t in pl["tracks"]:
+            seen: set[str] = set()
+            union: list[str] = []
+            for aid in t.get("artist_ids") or []:
+                for g in genres_by_artist.get(aid, []):
+                    if g not in seen:
+                        seen.add(g)
+                        union.append(g)
+            t["genres"] = union
+
+
+def write_csv(playlists, fh):
+    """Single combined sockseek/sldl CSV, one `# <name>` comment per playlist."""
+    w = csv.writer(fh)
+    w.writerow(["Artist", "Title", "Album", "Length"])
+    for pl in playlists:
+        fh.write(f"# {pl['name']}\n")
+        for t in pl["tracks"]:
+            w.writerow(
+                [
+                    ", ".join(t["artists"]),
+                    t["title"],
+                    t["album"],
+                    round(t["duration_ms"] / 1000),
+                ]
+            )
+
+
+async def fetch_playlists(ids: list[str], with_genres: bool = True) -> list[dict]:
+    """Resolve env, open a session, fetch every playlist concurrently.
+
+    When `with_genres` is true (default), follow up with a batched
+    `/v1/artists` call to populate each track's `genres` field.
+    """
+    client_id = os.environ.get("MAPLE_SPOTIFY_CLIENT_ID") or os.environ.get("SPOTIFY_CLIENT_ID")
+    client_secret = os.environ.get("MAPLE_SPOTIFY_CLIENT_SECRET") or os.environ.get("SPOTIFY_CLIENT_SECRET")
+    if not (client_id and client_secret):
+        sys.exit("set MAPLE_SPOTIFY_CLIENT_ID and MAPLE_SPOTIFY_CLIENT_SECRET")
+
+    timeout = aiohttp.ClientTimeout(total=30)
+    async with aiohttp.ClientSession(timeout=timeout) as session:
+        token = await get_token(session, client_id, client_secret)
+        playlists = await asyncio.gather(
+            *(fetch_playlist(session, parse_playlist_id(x), token) for x in ids)
+        )
+        if with_genres:
+            artist_ids: list[str] = []
+            for pl in playlists:
+                for t in pl["tracks"]:
+                    artist_ids.extend(t.get("artist_ids") or [])
+            genres = await fetch_artist_genres(session, artist_ids, token)
+            _enrich_track_genres(playlists, genres)
+        return playlists
+
+
+async def main_async() -> None:
+    argv = sys.argv[1:]
+    as_json = "--json" in argv
+    as_csv = "--csv" in argv
+    argv = [a for a in argv if a not in ("--json", "--csv")]
+
+    inputs: list[str] = []
+    out_path: str | None = None
+    i = 0
+    while i < len(argv):
+        if argv[i] == "--file":
+            inputs += read_list_file(argv[i + 1])
+            i += 2
+        elif argv[i] == "--out":
+            out_path = argv[i + 1]
+            i += 2
+        else:
+            inputs.append(argv[i])
+            i += 1
+
+    if not inputs:
+        sys.exit(
+            "usage: spotify.py [--json|--csv] <url-or-id ...> "
+            "[--file list.txt] [--out path]"
+        )
+
+    playlists = await fetch_playlists(inputs)
+
+    if as_json:
+        print(json.dumps(playlists, indent=2, ensure_ascii=False))
+        return
+
+    if as_csv:
+        if out_path:
+            with open(out_path, "w", newline="") as fh:
+                write_csv(playlists, fh)
+            print(f"wrote {sum(len(p['tracks']) for p in playlists)} tracks to {out_path}")
+        else:
+            write_csv(playlists, sys.stdout)
+        return
+
+    for pl in playlists:
+        print(f"\n=== {pl['name']} ({len(pl['tracks'])} tracks) ===")
+        for n, t in enumerate(pl["tracks"], 1):
+            print(f"{n:>3}. {', '.join(t['artists'])} - {t['title']}")
+
+
+def main() -> None:
+    asyncio.run(main_async())
+
+
+if __name__ == "__main__":
+    main()

maple-py/src/spotify.py

@@ -1,167 +0,0 @@
-#!/usr/bin/env -S uv run --script
-# /// script
-# requires-python = ">=3.14"
-# dependencies = []
-# ///
-"""Dump tracks from one or more Spotify playlists (Client Credentials flow).
-
-    just spotify                       # uses the placeholder ids below
-    just spotify "<url-or-id> ..."     # explicit playlists
-    just spotify "--file lists.txt --json"
-    just spotify "<url-or-id> --csv"   # single sockseek CSV (stdout)
-    just spotify "<url-or-id> --csv --out tracks.csv"
-
-Playlists can be ids, full URLs, or spotify: URIs. Works for any public
-playlist; no user login needed. Requires SPOTIFY_CLIENT_ID /
-SPOTIFY_CLIENT_SECRET in the environment (loaded from .env by just).
-
---csv writes one combined CSV with Artist,Title,Album,Length columns (Length
-in seconds), the format consumed by sockseek/sldl. Each playlist is preceded
-by a `# <name>` comment line as a separator. Writes to stdout, or to a file
-with --out <path>.
-"""
-import base64
-import csv
-import json
-import os
-import re
-import sys
-import urllib.parse
-import urllib.request
-
-
-def get_token(client_id, client_secret):
-    auth = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
-    data = urllib.parse.urlencode({"grant_type": "client_credentials"}).encode()
-    req = urllib.request.Request(
-        "https://accounts.spotify.com/api/token",
-        data=data,
-        headers={"Authorization": f"Basic {auth}"},
-    )
-    with urllib.request.urlopen(req) as resp:
-        return json.load(resp)["access_token"]
-
-
-def api_get(url, token):
-    req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"})
-    with urllib.request.urlopen(req) as resp:
-        return json.load(resp)
-
-
-def parse_playlist_id(arg):
-    m = re.search(r"playlist[/:]([A-Za-z0-9]+)", arg)
-    return m.group(1) if m else arg.strip()
-
-
-def read_list_file(path):
-    out = []
-    with open(path) as fh:
-        for line in fh:
-            line = line.split("#", 1)[0].strip()
-            if line:
-                out.append(line)
-    return out
-
-
-def fetch_playlist(pid, token):
-    meta = api_get(
-        f"https://api.spotify.com/v1/playlists/{pid}"
-        f"?fields={urllib.parse.quote('name,owner(display_name)')}",
-        token,
-    )
-    fields = "items(track(name,artists(name),album(name),duration_ms)),next"
-    url = (
-        f"https://api.spotify.com/v1/playlists/{pid}/tracks"
-        f"?limit=100&fields={urllib.parse.quote(fields)}"
-    )
-    tracks = []
-    while url:
-        page = api_get(url, token)
-        for item in page.get("items", []):
-            t = item.get("track")
-            if not t:
-                continue
-            tracks.append(
-                {
-                    "title": t["name"],
-                    "artists": [a["name"] for a in t["artists"]],
-                    "album": t["album"]["name"],
-                    "duration_ms": t["duration_ms"],
-                }
-            )
-        url = page.get("next")
-    return {"id": pid, "name": meta.get("name"), "tracks": tracks}
-
-
-def write_csv(playlists, fh):
-    """Single combined sockseek/sldl CSV, one `# <name>` comment per playlist."""
-    w = csv.writer(fh)
-    w.writerow(["Artist", "Title", "Album", "Length"])
-    for pl in playlists:
-        fh.write(f"# {pl['name']}\n")
-        for t in pl["tracks"]:
-            w.writerow(
-                [
-                    ", ".join(t["artists"]),
-                    t["title"],
-                    t["album"],
-                    round(t["duration_ms"] / 1000),
-                ]
-            )
-
-
-def main():
-    argv = sys.argv[1:]
-    as_json = "--json" in argv
-    as_csv = "--csv" in argv
-    argv = [a for a in argv if a not in ("--json", "--csv")]
-
-    inputs = []
-    out_path = None
-    i = 0
-    while i < len(argv):
-        if argv[i] == "--file":
-            inputs += read_list_file(argv[i + 1])
-            i += 2
-        elif argv[i] == "--out":
-            out_path = argv[i + 1]
-            i += 2
-        else:
-            inputs.append(argv[i])
-            i += 1
-
-    if not inputs:
-        sys.exit(
-            "usage: spotify.py [--json|--csv] <url-or-id ...> "
-            "[--file list.txt] [--out path]"
-        )
-
-    client_id = os.environ.get("SPOTIFY_CLIENT_ID")
-    client_secret = os.environ.get("SPOTIFY_CLIENT_SECRET")
-    if not (client_id and client_secret):
-        sys.exit("set SPOTIFY_CLIENT_ID and SPOTIFY_CLIENT_SECRET")
-
-    token = get_token(client_id, client_secret)
-    playlists = [fetch_playlist(parse_playlist_id(x), token) for x in inputs]
-
-    if as_json:
-        print(json.dumps(playlists, indent=2, ensure_ascii=False))
-        return
-
-    if as_csv:
-        if out_path:
-            with open(out_path, "w", newline="") as fh:
-                write_csv(playlists, fh)
-            print(f"wrote {sum(len(p['tracks']) for p in playlists)} tracks to {out_path}")
-        else:
-            write_csv(playlists, sys.stdout)
-        return
-
-    for pl in playlists:
-        print(f"\n=== {pl['name']} ({len(pl['tracks'])} tracks) ===")
-        for n, t in enumerate(pl["tracks"], 1):
-            print(f"{n:>3}. {', '.join(t['artists'])} - {t['title']}")
-
-
-if __name__ == "__main__":
-    main()