Incremental live Spotify fetch (snapshot/watermark)

The live fetch no longer re-pulls everything each run:
- playlists: re-fetch a playlist's tracks only when its snapshot_id changed
  (new/deleted playlists handled); state in spotify_playlist table.
- liked songs: page newest-first, stop at the newest added_at already seen
  (watermark in ingest_state).
- cover art: download only when the file is missing or the URL changed.

New src/ingest/spotify_fetch.py drives this against the low-level
SpotifyWebAPI client and upserts straight to the DB; fetch uses it when creds
are set (falls back to the cached-JSON ingest otherwise). `fetch --spotify-full`
forces a complete resync. Shared upsert helper extracted into spotify_raw.

Verified with a mocked API: unchanged re-run does 0 playlist-track fetches and
one liked page; a snapshot change re-fetches exactly that playlist.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

README.md

@@ -78,8 +78,17 @@ python main.py display    # open the GUI to see what you're missing
 **`fetch`** pulls Spotify, scans the local library, parses the iTunes export,
 resolves MBIDs, and matches — logging progress at each step. The **first run is
 slow** (full scan + MusicBrainz lookups at ~1 req/s); **later runs are quick**
-because only changed files, the iTunes export if it changed, and still-unresolved
+because each source only touches what changed:
-MBIDs are touched. Useful flags/helpers:
+
+- **Spotify** — playlists are re-fetched only when their `snapshot_id` changes;
+  liked songs are paged newest-first and stop at the newest one already stored;
+  cover art downloads only when missing or its URL changed (`--spotify-full`
+  forces a complete resync).
+- **Local** — only files whose mtime/size changed; deleted files are pruned.
+- **iTunes** — skipped entirely if the export's mtime is unchanged.
+- **MBID enrichment** — only still-unresolved tracks.
+
+Useful flags/helpers:
 
 ```bash
 python main.py fetch --enrich-limit 500   # cap MBID lookups this run

main.py

@@ -19,7 +19,8 @@ def cmd_fetch(args):
     _setup_logging()
     from src.workflows import run_fetch
 
-    run_fetch(enrich_limit=args.enrich_limit, enrich=not args.no_enrich)
+    run_fetch(enrich_limit=args.enrich_limit, enrich=not args.no_enrich,
+              spotify_full=args.spotify_full)
 
 
 def cmd_display(args):
@@ -56,6 +57,8 @@ def build_parser():
                        help="cap MBID lookups this run (default: resolve all)")
     fetch.add_argument("--no-enrich", action="store_true",
                        help="skip MusicBrainz enrichment; use MBIDs already on tracks")
+    fetch.add_argument("--spotify-full", action="store_true",
+                       help="force a full Spotify resync (ignore snapshot/watermark)")
 
     sub.add_parser("display", help="open the Streamlit GUI")
 

src/db/models.py

@@ -99,11 +99,25 @@ class FileState(Base):
 
 
 class IngestState(Base):
-    """Small key/value table for incremental-fetch bookkeeping, e.g. playlist
+    """Small key/value table for incremental-fetch bookkeeping, e.g. the newest
-    snapshot ids or the newest liked-track id seen on the last run."""
+    liked-track timestamp or cover URLs seen on the last run."""
 
     __tablename__ = "ingest_state"
 
     key = Column(String, primary_key=True)
     value = Column(String)
     updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
+
+
+class SpotifyPlaylist(Base):
+    """Per-playlist sync state so unchanged playlists are skipped: Spotify's
+    snapshot_id changes whenever a playlist's contents change, and cover_url
+    changes when its image changes."""
+
+    __tablename__ = "spotify_playlist"
+
+    playlist_id = Column(String, primary_key=True)
+    name = Column(String)
+    snapshot_id = Column(String)
+    cover_url = Column(String)
+    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

src/ingest/spotify_fetch.py

@@ -0,0 +1,162 @@
+"""Incremental live Spotify fetch.
+
+Only fetches what changed since last run:
+  - liked songs : returned newest-first, so we page until we reach the newest
+    timestamp already stored, then stop.
+  - playlists   : each carries a snapshot_id; a playlist's tracks are only
+    re-fetched when its snapshot_id changed (new/removed playlists handled too).
+  - cover art   : downloaded only when the image is missing or its URL changed.
+
+Writes straight to the DB (upsert). Pass full=True to force a complete resync.
+"""
+import logging
+import os
+
+import requests
+
+import src.backends.spotify.SpotifyWebAPI as sp
+from Env import client_id, client_secret, spotify_cache_dir
+from src.db.database import session_scope
+from src.db.models import IngestState, SOURCE_SPOTIFY, SpotifyPlaylist, Track
+from src.ingest.spotify_raw import upsert_spotify_track
+
+log = logging.getLogger("musicindexer.spotify")
+
+_LIKED_WATERMARK = "liked_newest_added_at"
+_USER_COVER_URL = "spotify_user_cover_url"
+_COVERS_DIR = "covers"
+
+
+def _existing_tracks(session):
+    return {t.source_id: t for t in
+            session.query(Track).filter_by(source=SOURCE_SPOTIFY).all()}
+
+
+def _get_state(session, key):
+    row = session.get(IngestState, key)
+    return row.value if row else None
+
+
+def _set_state(session, key, value):
+    row = session.get(IngestState, key) or IngestState(key=key)
+    row.value = value
+    session.merge(row)
+
+
+def _cover_path(name):
+    return os.path.join(str(spotify_cache_dir), _COVERS_DIR, f"{name}.jpg")
+
+
+def _download_cover(name, url, prev_url):
+    """Download only if the file is missing or the URL changed. Returns True
+    if a download happened."""
+    if not url:
+        return False
+    path = _cover_path(name)
+    if url == prev_url and os.path.exists(path):
+        return False
+    try:
+        resp = requests.get(url, timeout=30)
+        resp.raise_for_status()
+    except requests.RequestException as exc:
+        log.warning("cover download failed for %s: %s", name, exc)
+        return False
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, "wb") as fh:
+        fh.write(resp.content)
+    return True
+
+
+def _sync_liked(session, existing, full):
+    watermark = None if full else _get_state(session, _LIKED_WATERMARK)
+    counts = {"new": 0, "updated": 0}
+    newest = watermark
+    offset = 0
+    stop = False
+    while not stop:
+        page = sp.fetch(f"v1/me/tracks?limit=50&offset={offset}")
+        items = page.get("items", [])
+        if not items:
+            break
+        for item in items:
+            added = item.get("added_at")
+            track = item.get("track")
+            if watermark and added and added <= watermark:
+                stop = True
+                break
+            if track:
+                upsert_spotify_track(session, existing, track, counts)
+                if newest is None or (added and added > newest):
+                    newest = added
+        if len(items) < 50:
+            break
+        offset += 50
+    if newest:
+        _set_state(session, _LIKED_WATERMARK, newest)
+    log.info("liked songs: %d new, %d updated%s", counts["new"], counts["updated"],
+             "" if watermark is None else " (incremental)")
+    return counts
+
+
+def _sync_playlists(session, existing, full):
+    playlists = sp.fetch_paginated("v1/me/playlists?limit=50")
+    states = {p.playlist_id: p for p in session.query(SpotifyPlaylist).all()}
+    counts = {"new": 0, "updated": 0}
+    changed = unchanged = covers = 0
+    seen = set()
+
+    for pl in playlists:
+        pid = pl.get("id")
+        if not pid:
+            continue
+        seen.add(pid)
+        snapshot = pl.get("snapshot_id")
+        cover_url = (pl.get("images") or [{}])[0].get("url")
+        row = states.get(pid)
+        if row is None:
+            row = SpotifyPlaylist(playlist_id=pid)
+            session.add(row)
+            states[pid] = row
+
+        if row.snapshot_id == snapshot and not full:
+            unchanged += 1
+        else:
+            items = sp.fetch_paginated(f"v1/playlists/{pid}/tracks?limit=50")
+            for item in items:
+                track = item.get("track")
+                if track and track.get("id"):
+                    upsert_spotify_track(session, existing, track, counts)
+            changed += 1
+            row.snapshot_id = snapshot
+        row.name = pl.get("name")
+
+        if _download_cover(pid, cover_url, row.cover_url):
+            covers += 1
+        row.cover_url = cover_url
+
+    for pid, row in states.items():  # playlists deleted on Spotify
+        if pid not in seen:
+            session.delete(row)
+
+    log.info("playlists: %d changed, %d unchanged; tracks %d new, %d updated; "
+             "%d covers fetched", changed, unchanged, counts["new"], counts["updated"], covers)
+    return counts
+
+
+def _sync_user_cover(session):
+    profile = sp.fetch("v1/me")
+    images = profile.get("images") or []
+    url = images[-1]["url"] if images else None
+    if _download_cover("user", url, _get_state(session, _USER_COVER_URL)):
+        _set_state(session, _USER_COVER_URL, url)
+        log.info("user cover updated")
+
+
+def fetch_spotify_incremental(full=False):
+    """Authenticate and sync liked songs, playlists and cover art incrementally."""
+    sp.update_access_token(client_id, client_secret)
+    with session_scope() as session:
+        existing = _existing_tracks(session)
+        _sync_liked(session, existing, full)
+        _sync_playlists(session, existing, full)
+        _sync_user_cover(session)

src/ingest/spotify_raw.py

@@ -29,6 +29,46 @@ def _cache_signature(spotify_dir):
     return str(max(mtimes)) if mtimes else "0"
 
 
+def spotify_track_fields(track_obj):
+    """Map a Spotify full-track object to our Track columns."""
+    names = [a.get("name") for a in track_obj.get("artists", []) if a.get("name")]
+    primary = names[0] if names else None
+    title = track_obj.get("name")
+    return dict(
+        title=title, artist=primary, artists=", ".join(names),
+        album=(track_obj.get("album") or {}).get("name"),
+        duration_ms=track_obj.get("duration_ms"),
+        isrc=(track_obj.get("external_ids") or {}).get("isrc"),
+        norm_key=norm_key(primary or "", title or ""),
+    )
+
+
+def upsert_spotify_track(session, existing, track_obj, counts):
+    """Upsert one Spotify track into `existing` (source_id -> Track), updating
+    `counts`. Resets MBID enrichment only if the ISRC actually changed."""
+    spotify_id = track_obj.get("id")
+    if not spotify_id:
+        return
+    fields = spotify_track_fields(track_obj)
+    new_isrc = fields.pop("isrc")
+    track = existing.get(spotify_id)
+    if track is None:
+        track = Track(source=SOURCE_SPOTIFY, source_id=spotify_id,
+                      isrc=new_isrc, **fields)
+        session.add(track)
+        existing[spotify_id] = track
+        counts["new"] += 1
+        return
+    for key, value in fields.items():
+        setattr(track, key, value)
+    if new_isrc and new_isrc != track.isrc:
+        track.isrc, track.mbid, track.mbid_status, track.mbid_source = (
+            new_isrc, None, None, None)
+    elif new_isrc:
+        track.isrc = new_isrc
+    counts["updated"] += 1
+
+
 def _track_objects(spotify_dir):
     by_id = {}
     for fname in _SOURCES:
@@ -63,31 +103,8 @@ def ingest_spotify(spotify_dir=None, force=False):
     with session_scope() as session:
         existing = {t.source_id: t for t in
                     session.query(Track).filter_by(source=SOURCE_SPOTIFY).all()}
-        for spotify_id, t in tqdm(objects.items(), desc="ingest spotify", unit="track"):
+        for t in tqdm(objects.values(), desc="ingest spotify", unit="track"):
-            names = [a.get("name") for a in t.get("artists", []) if a.get("name")]
+            upsert_spotify_track(session, existing, t, counts)
-            primary = names[0] if names else None
-            title = t.get("name")
-            fields = dict(
-                title=title, artist=primary, artists=", ".join(names),
-                album=(t.get("album") or {}).get("name"),
-                duration_ms=t.get("duration_ms"),
-                isrc=(t.get("external_ids") or {}).get("isrc"),
-                norm_key=norm_key(primary or "", title or ""),
-            )
-            track = existing.get(spotify_id)
-            if track is None:
-                session.add(Track(source=SOURCE_SPOTIFY, source_id=spotify_id, **fields))
-                counts["new"] += 1
-            else:
-                new_isrc = fields.pop("isrc")
-                for key, value in fields.items():
-                    setattr(track, key, value)
-                if new_isrc and new_isrc != track.isrc:  # stale enrichment
-                    track.isrc, track.mbid, track.mbid_status, track.mbid_source = (
-                        new_isrc, None, None, None)
-                elif new_isrc:
-                    track.isrc = new_isrc
-                counts["updated"] += 1
 
         state = session.get(IngestState, _STATE_KEY) or IngestState(key=_STATE_KEY)
         state.value = signature

src/workflows.py

@@ -21,7 +21,8 @@ from src.db.database import init_db
 from src.enrich.mbid import resolve_mbids, unresolved_count
 from src.ingest.itunes_xml import ingest_itunes
 from src.ingest.local_scan import scan_local
-from src.ingest.spotify_raw import fetch_spotify_live, ingest_spotify
+from src.ingest.spotify_fetch import fetch_spotify_incremental
+from src.ingest.spotify_raw import ingest_spotify
 from src.match.matcher import summary
 
 log = logging.getLogger("musicindexer")
@@ -32,7 +33,7 @@ def _step(title):
     log.info("== %s", title)
 
 
-def run_fetch(enrich_limit=None, enrich=True):
+def run_fetch(enrich_limit=None, enrich=True, spotify_full=False):
     started = time.time()
     init_db()
     log.info("MusicIndexer — incremental fetch")
@@ -42,11 +43,13 @@ def run_fetch(enrich_limit=None, enrich=True):
 
     _step("Spotify")
     if client_id and client_secret:
-        log.info("credentials present -> live fetch")
+        log.info("credentials present -> incremental live fetch%s",
+                 " (full resync)" if spotify_full else "")
         try:
-            fetch_spotify_live()
+            fetch_spotify_incremental(full=spotify_full)
         except Exception as exc:  # keep going on cached data
-            log.error("live fetch failed (%s); using cached responses", exc)
+            log.error("live fetch failed (%s); falling back to cached responses", exc)
+            ingest_spotify()
     else:
         log.info("no SPOTIFY_CLIENT_ID/SECRET -> using cached responses")
         ingest_spotify()