Track Spotify origin (liked/playlist/top); make top tracks opt-in; clean Ctrl+C

- Tag each Spotify track with its origin (liked / from_playlist / from_top,
  combinable) so genuinely-saved tracks are distinguishable from ones only
  surfaced by the top-tracks endpoint.
- Top tracks are no longer added by default (they inflated "missing" with
  unowned listening stats). --include-top pulls them as full objects so they
  carry an ISRC for real ID matching.
- On --spotify-full, prune Spotify rows that are neither liked nor in a
  playlist (top-only / orphaned), cleaning an existing DB in one pass.
- Cached-ingest fallback honours include_top and prunes too.
- In-place SQLite migration adds the new columns (no DB rebuild needed).
- Ctrl+C exits cleanly (130) with a message instead of a traceback;
  incremental progress is already committed, so re-run to continue.
- Surface origin breakdown in the fetch log, match summary, and GUI.

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

Gui.py

@@ -49,6 +49,12 @@ def main():
         c4.metric("Have (text)", s["present_text"])
         coverage = (s["has_mbid"] / s["total"] * 100) if s["total"] else 0
         c5.metric("MBID coverage", f"{coverage:.0f}%")
+        if source == "spotify":
+            st.caption(
+                f"Origins — liked: {s.get('liked', 0)} · "
+                f"in a playlist: {s.get('from_playlist', 0)} · "
+                f"top tracks: {s.get('from_top', 0)}"
+            )
 
     local = summ.get("local", {})
     st.caption(
@@ -80,7 +86,7 @@ def main():
         return
 
     df = pd.DataFrame(rows)[
-        ["status", "source", "artist", "title", "album", "matched_by",
+        ["status", "source", "origin", "artist", "title", "album", "matched_by",
          "isrc", "mbid", "play_count", "spotify_id"]
     ]
     st.write(f"**{len(df)}** tracks")

README.md

@@ -84,6 +84,15 @@ because each source only touches what changed:
   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).
+  - Your library is **liked songs + playlist tracks**. The *top-tracks* endpoint
+    is **not** pulled by default (those are just listening stats, not saved
+    music, and would inflate "missing"). Add `--include-top` to pull them too —
+    they're fetched as full Spotify objects, so they carry an ISRC for real ID
+    matching like everything else.
+  - Each Spotify track records its **origin** (`liked` / `playlist` / `top`,
+    combinable). On a `--spotify-full` run, rows that are *only* there because
+    the top endpoint listed them (neither liked nor in a playlist) are pruned —
+    run `python main.py fetch --spotify-full` once to clean up an existing DB.
 - **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.

main.py

@@ -9,6 +9,7 @@ Helpers: `enrich` (resume MBID resolution), `match` (print the summary).
 """
 import argparse
 import logging
+import sys
 
 
 def _setup_logging():
@@ -20,7 +21,7 @@ def cmd_fetch(args):
     from src.workflows import run_fetch
 
     run_fetch(enrich_limit=args.enrich_limit, enrich=not args.no_enrich,
-              spotify_full=args.spotify_full)
+              spotify_full=args.spotify_full, include_top=args.include_top)
 
 
 def cmd_display(args):
@@ -59,6 +60,9 @@ def build_parser():
                        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)")
+    fetch.add_argument("--include-top", action="store_true",
+                       help="also pull the top-tracks endpoint (off by default; "
+                            "with --spotify-full, top-only tracks are otherwise pruned)")
 
     sub.add_parser("display", help="open the Streamlit GUI")
 
@@ -72,12 +76,19 @@ def build_parser():
 
 def main():
     args = build_parser().parse_args()
-    {
+    try:
-        "fetch": cmd_fetch,
+        {
-        "display": cmd_display,
+            "fetch": cmd_fetch,
-        "enrich": cmd_enrich,
+            "display": cmd_display,
-        "match": cmd_match,
+            "enrich": cmd_enrich,
-    }[args.cmd](args)
+            "match": cmd_match,
+        }[args.cmd](args)
+    except KeyboardInterrupt:
+        # Ctrl+C: stop cleanly, no traceback. Work already committed is kept
+        # (each step writes incrementally), so just re-run to continue.
+        print("\ninterrupted — partial progress saved; re-run to continue.",
+              file=sys.stderr)
+        sys.exit(130)
 
 
 if __name__ == "__main__":

src/db/database.py

@@ -2,7 +2,7 @@
 code runs on SQLite (default) or Postgres by changing one env var."""
 from contextlib import contextmanager
 
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, inspect, text
 from sqlalchemy.orm import sessionmaker
 
 from Env import database_url
@@ -11,16 +11,41 @@ from src.db.models import Base
 _engine = create_engine(database_url, future=True)
 SessionLocal = sessionmaker(bind=_engine, future=True, expire_on_commit=False)
 
+# Columns added after the table first shipped. create_all() never ALTERs an
+# existing table, so we add any missing ones in place (keeps the local-scan /
+# iTunes / MBID-cache data instead of forcing a rebuild).
+_ADDED_COLUMNS = {
+    "track": [
+        ("liked", "BOOLEAN DEFAULT 0"),
+        ("from_playlist", "BOOLEAN DEFAULT 0"),
+        ("from_top", "BOOLEAN DEFAULT 0"),
+    ],
+}
+
 
 def get_engine():
     return _engine
 
 
+def _migrate(engine):
+    inspector = inspect(engine)
+    existing_tables = set(inspector.get_table_names())
+    with engine.begin() as conn:
+        for table, columns in _ADDED_COLUMNS.items():
+            if table not in existing_tables:
+                continue
+            have = {c["name"] for c in inspector.get_columns(table)}
+            for name, ddl in columns:
+                if name not in have:
+                    conn.execute(text(f'ALTER TABLE "{table}" ADD COLUMN {name} {ddl}'))
+
+
 def init_db():
     """Create the storage dir + tables if missing (safe to call every run)."""
     from Env import data_dir
 
     data_dir.mkdir(parents=True, exist_ok=True)
+    _migrate(_engine)
     Base.metadata.create_all(_engine)
 
 

src/db/models.py

@@ -7,6 +7,7 @@ fetched is lost.
 from datetime import datetime
 
 from sqlalchemy import (
+    Boolean,
     Column,
     DateTime,
     Float,
@@ -56,6 +57,13 @@ class Track(Base):
     mbid = Column(String, index=True)          # recording MBID
     norm_key = Column(String, index=True)      # normalized "artist|title" fallback
 
+    # Where a Spotify track came from (a track can be several at once). Used to
+    # tell genuinely-saved tracks (liked / in a playlist) apart from ones only
+    # surfaced by the top-tracks endpoint, and to opt those out of the library.
+    liked = Column(Boolean, default=False, index=True)
+    from_playlist = Column(Boolean, default=False, index=True)
+    from_top = Column(Boolean, default=False, index=True)
+
     # Enrichment bookkeeping so the resolver only touches unresolved rows.
     mbid_status = Column(String, index=True)   # None | found | notfound | error
     mbid_source = Column(String)               # isrc | text
@@ -71,6 +79,14 @@ class Track(Base):
         MBID, fall back to the normalized text key."""
         return self.mbid or self.norm_key
 
+    @property
+    def origin(self) -> str:
+        """Human-readable Spotify origin, e.g. "liked+playlist" or "top"."""
+        parts = [name for flag, name in
+                 (("liked", "liked"), ("from_playlist", "playlist"), ("from_top", "top"))
+                 if getattr(self, flag)]
+        return "+".join(parts)
+
 
 class MbCache(Base):
     """Cache of MusicBrainz lookups (positive AND negative) so the resolver is

src/ingest/spotify_fetch.py

@@ -7,6 +7,10 @@ Only fetches what changed since last run:
     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.
 
+Each track is tagged with its origin (liked / playlist / top). The top-tracks
+endpoint is only consulted when include_top=True, and on a full resync the rows
+that exist *only* because the top endpoint listed them are pruned.
+
 Writes straight to the DB (upsert). Pass full=True to force a complete resync.
 """
 import logging
@@ -18,13 +22,14 @@ 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
+from src.ingest.spotify_raw import prune_top_only, upsert_spotify_track
 
 log = logging.getLogger("musicindexer.spotify")
 
 _LIKED_WATERMARK = "liked_newest_added_at"
 _USER_COVER_URL = "spotify_user_cover_url"
 _COVERS_DIR = "covers"
+_TOP_RANGE = "long_term"
 
 
 def _existing_tracks(session):
@@ -85,7 +90,7 @@ def _sync_liked(session, existing, full):
                 stop = True
                 break
             if track:
-                upsert_spotify_track(session, existing, track, counts)
+                upsert_spotify_track(session, existing, track, counts, origin="liked")
                 if newest is None or (added and added > newest):
                     newest = added
         if len(items) < 50:
@@ -125,7 +130,7 @@ def _sync_playlists(session, existing, full):
             for item in items:
                 track = item.get("track")
                 if track and track.get("id"):
-                    upsert_spotify_track(session, existing, track, counts)
+                    upsert_spotify_track(session, existing, track, counts, origin="playlist")
             changed += 1
             row.snapshot_id = snapshot
         row.name = pl.get("name")
@@ -143,6 +148,19 @@ def _sync_playlists(session, existing, full):
     return counts
 
 
+def _sync_top(session, existing):
+    """Pull the top-tracks endpoint (full objects, so they carry an ISRC just
+    like liked songs) and flag them. Opt-in; off by default."""
+    items = sp.fetch_paginated(
+        f"v1/me/top/tracks?limit=50&time_range={_TOP_RANGE}", desc="Fetching top tracks")
+    counts = {"new": 0, "updated": 0}
+    for track in items:
+        if track and track.get("id"):
+            upsert_spotify_track(session, existing, track, counts, origin="top")
+    log.info("top tracks: %d new, %d updated", counts["new"], counts["updated"])
+    return counts
+
+
 def _sync_user_cover(session):
     profile = sp.fetch("v1/me")
     images = profile.get("images") or []
@@ -152,11 +170,21 @@ def _sync_user_cover(session):
         log.info("user cover updated")
 
 
-def fetch_spotify_incremental(full=False):
+def fetch_spotify_incremental(full=False, include_top=False):
-    """Authenticate and sync liked songs, playlists and cover art incrementally."""
+    """Authenticate and sync liked songs, playlists and cover art incrementally.
+
+    include_top also pulls the top-tracks endpoint. On a full resync the liked
+    and playlist passes re-set every flag, so we then prune any Spotify rows
+    that are neither liked nor in a playlist (top-only, unless include_top)."""
     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)
+        if include_top:
+            _sync_top(session, existing)
         _sync_user_cover(session)
+        if full:
+            pruned = prune_top_only(session, include_top)
+            if pruned:
+                log.info("pruned %d top-only / orphaned Spotify tracks", pruned)

src/ingest/spotify_raw.py

@@ -16,10 +16,18 @@ from src.match.normalize import norm_key
 
 log = logging.getLogger("musicindexer.spotify")
 
-# Files written by SpotifyWebAPI.fetch_data(), each holding full track objects.
+# Files written by SpotifyWebAPI.fetch_data(), each holding full track objects,
-_SOURCES = ("tracks.json", "top_tracks.json", "playlistTracks.json")
+# mapped to the origin flag they set on a Track.
+_SOURCES = {
+    "tracks.json": "liked",
+    "playlistTracks.json": "playlist",
+    "top_tracks.json": "top",
+}
 _STATE_KEY = "spotify_cache_sig"
 
+# origin name -> Track boolean column
+_ORIGIN_FLAG = {"liked": "liked", "playlist": "from_playlist", "top": "from_top"}
+
 
 def _cache_signature(spotify_dir):
     """Latest mtime across the cached payloads; changes whenever a fetch
@@ -29,6 +37,22 @@ def _cache_signature(spotify_dir):
     return str(max(mtimes)) if mtimes else "0"
 
 
+def prune_top_only(session, include_top):
+    """Drop Spotify tracks that are *only* there because the top-tracks endpoint
+    listed them — i.e. neither liked nor in any playlist. Keeps top tracks when
+    the caller opted in. Safe only after a full pass that re-set the flags.
+    Returns the number deleted."""
+    keep = Track.liked.is_(True) | Track.from_playlist.is_(True)
+    if include_top:
+        keep = keep | Track.from_top.is_(True)
+    doomed = (session.query(Track)
+              .filter(Track.source == SOURCE_SPOTIFY)
+              .filter(~keep).all())
+    for track in doomed:
+        session.delete(track)
+    return len(doomed)
+
+
 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")]
@@ -43,24 +67,31 @@ def spotify_track_fields(track_obj):
     )
 
 
-def upsert_spotify_track(session, existing, track_obj, counts):
+def upsert_spotify_track(session, existing, track_obj, counts, origin=None):
     """Upsert one Spotify track into `existing` (source_id -> Track), updating
-    `counts`. Resets MBID enrichment only if the ISRC actually changed."""
+    `counts`. Resets MBID enrichment only if the ISRC actually changed. `origin`
+    (liked / playlist / top) sets the matching flag True and is never cleared,
+    so a track seen from several sources accumulates all of them."""
     spotify_id = track_obj.get("id")
     if not spotify_id:
         return
+    flag = _ORIGIN_FLAG.get(origin)
     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)
+        if flag:
+            setattr(track, flag, True)
         session.add(track)
         existing[spotify_id] = track
         counts["new"] += 1
         return
     for key, value in fields.items():
         setattr(track, key, value)
+    if flag:
+        setattr(track, flag, True)
     if new_isrc and new_isrc != track.isrc:
         track.isrc, track.mbid, track.mbid_status, track.mbid_source = (
             new_isrc, None, None, None)
@@ -69,9 +100,12 @@ def upsert_spotify_track(session, existing, track_obj, counts):
     counts["updated"] += 1
 
 
-def _track_objects(spotify_dir):
+def _track_objects(spotify_dir, include_top):
-    by_id = {}
+    """Yield (track_obj, origin) for every cached track. Top tracks are skipped
-    for fname in _SOURCES:
+    unless include_top, so by default they don't pollute the library."""
+    for fname, origin in _SOURCES.items():
+        if origin == "top" and not include_top:
+            continue
         path = os.path.join(spotify_dir, fname)
         if not os.path.isfile(path):
             continue
@@ -81,36 +115,39 @@ def _track_objects(spotify_dir):
         for item in items:
             track = item.get("track") if isinstance(item, dict) and "track" in item else item
             if track and track.get("id"):
-                by_id[track["id"]] = track
+                yield track, origin
-    return by_id
 
 
-def ingest_spotify(spotify_dir=None, force=False):
+def ingest_spotify(spotify_dir=None, force=False, include_top=False):
     spotify_dir = str(spotify_dir or spotify_cache_dir)
     if not os.path.isdir(spotify_dir):
         log.warning("no Spotify cache at %s — skipping", spotify_dir)
         return {"new": 0, "updated": 0, "skipped": True}
 
-    signature = _cache_signature(spotify_dir)
+    signature = f"{_cache_signature(spotify_dir)}|top={int(include_top)}"
     with session_scope() as session:
         state = session.get(IngestState, _STATE_KEY)
         if state and state.value == signature and not force:
             log.info("Spotify cache unchanged — skipping")
             return {"new": 0, "updated": 0, "skipped": True}
 
-    objects = _track_objects(spotify_dir)
     counts = {"new": 0, "updated": 0}
     with session_scope() as session:
         existing = {t.source_id: t for t in
                     session.query(Track).filter_by(source=SOURCE_SPOTIFY).all()}
-        for t in tqdm(objects.values(), desc="ingest spotify", unit="track"):
+        for track, origin in tqdm(list(_track_objects(spotify_dir, include_top)),
-            upsert_spotify_track(session, existing, t, counts)
+                                  desc="ingest spotify", unit="track"):
+            upsert_spotify_track(session, existing, track, counts, origin=origin)
+
+        # The cache is a complete snapshot, so it's safe to drop top-only rows.
+        pruned = prune_top_only(session, include_top)
 
         state = session.get(IngestState, _STATE_KEY) or IngestState(key=_STATE_KEY)
         state.value = signature
         session.merge(state)
 
-    log.info("spotify: %(new)d new, %(updated)d updated", counts)
+    log.info("spotify: %d new, %d updated, %d top-only pruned",
+             counts["new"], counts["updated"], pruned)
     return counts
 
 

src/match/matcher.py

@@ -40,6 +40,7 @@ def _classify(track, local_isrcs, local_mbids, local_norms):
 def _row(track, method):
     return {
         "source": track.source,
+        "origin": track.origin if track.source == SOURCE_SPOTIFY else "",
         "artist": track.artist,
         "title": track.title,
         "album": track.album,
@@ -76,10 +77,15 @@ def summary():
         for source in WANTED_SOURCES:
             tracks = session.query(Track).filter_by(source=source).all()
             counts = {"total": len(tracks), "present_mbid": 0,
-                      "present_text": 0, "missing": 0, "has_mbid": 0}
+                      "present_text": 0, "missing": 0, "has_mbid": 0,
+                      "liked": 0, "from_playlist": 0, "from_top": 0}
             for track in tracks:
                 if track.mbid:
                     counts["has_mbid"] += 1
+                if source == SOURCE_SPOTIFY:
+                    counts["liked"] += bool(track.liked)
+                    counts["from_playlist"] += bool(track.from_playlist)
+                    counts["from_top"] += bool(track.from_top)
                 has_local, method = _classify(track, local_isrcs, local_mbids, local_norms)
                 if not has_local:
                     counts["missing"] += 1

src/workflows.py

@@ -33,26 +33,27 @@ def _step(title):
     log.info("== %s", title)
 
 
-def run_fetch(enrich_limit=None, enrich=True, spotify_full=False):
+def run_fetch(enrich_limit=None, enrich=True, spotify_full=False, include_top=False):
     started = time.time()
     init_db()
     log.info("MusicIndexer — incremental fetch")
     log.info("  local library : %s", local_library_path)
     log.info("  itunes xml    : %s", itunes_xml_path)
     log.info("  spotify cache : %s", spotify_cache_dir)
+    log.info("  top tracks    : %s", "included" if include_top else "excluded (liked + playlists only)")
 
     _step("Spotify")
     if client_id and client_secret:
         log.info("credentials present -> incremental live fetch%s",
                  " (full resync)" if spotify_full else "")
         try:
-            fetch_spotify_incremental(full=spotify_full)
+            fetch_spotify_incremental(full=spotify_full, include_top=include_top)
         except Exception as exc:  # keep going on cached data
             log.error("live fetch failed (%s); falling back to cached responses", exc)
-            ingest_spotify()
+            ingest_spotify(include_top=include_top)
     else:
         log.info("no SPOTIFY_CLIENT_ID/SECRET -> using cached responses")
-        ingest_spotify()
+        ingest_spotify(include_top=include_top)
 
     _step("iTunes")
     ingest_itunes()
@@ -78,6 +79,10 @@ def run_fetch(enrich_limit=None, enrich=True, spotify_full=False):
         if s:
             log.info("%-7s total=%-5d MISSING=%-5d have(id)=%-5d have(text)=%-5d",
                      source, s["total"], s["missing"], s["present_mbid"], s["present_text"])
+    sp = summ.get("spotify")
+    if sp:
+        log.info("spotify origins: liked=%d playlist=%d top=%d",
+                 sp.get("liked", 0), sp.get("from_playlist", 0), sp.get("from_top", 0))
     local = summ.get("local", {})
     log.info("local   total=%-5d with_mbid=%d", local.get("total", 0), local.get("has_mbid", 0))