feat: add FastAPI webapp for Spotify/Last.FM library management

CLAUDE.md

@@ -51,6 +51,11 @@ A Python library and CLI tool for Spotify and Last.FM API interaction. Focuses o
 - `spotfm/sqlite.py` - SQLite singleton connection management
 - `spotfm/utils.py` - Config, caching, string sanitization
 
+**Layered Separation Rule (core vs. presentation):**
+- **Core modules** (`spotfm/spotify/`, `spotfm/lastfm.py`): return structured data (dicts/lists), use `logging` for observability, **never call `print()`**
+- **Presentation layer** (`spotfm/cli.py`, `spotfm/web/`): format and display data; **never own business logic**
+- Shared orchestration logic used by both CLI and web belongs in the core module, not duplicated in each caller
+
 ## Code Style & Requirements
 
 - **Python**: 3.14+ (PEP 758 bracketless exception syntax)

Dockerfile

@@ -0,0 +1,8 @@
+FROM python:3.14-slim
+WORKDIR /app
+RUN pip install --no-cache-dir uv
+COPY pyproject.toml LICENSE README.md ./
+COPY spotfm/ spotfm/
+RUN uv pip install --no-cache-dir --system ".[web]"
+EXPOSE 8000
+CMD ["uvicorn", "spotfm.web.app:create_app", "--factory", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]

docker-compose.yml

@@ -0,0 +1,9 @@
+services:
+  spotfm-web:
+    build: .
+    ports:
+      - "8000:8000"
+    volumes:
+      - ~/.spotfm:/root/.spotfm
+      - ~/.cache/spotfm:/root/.cache/spotfm
+    restart: unless-stopped

pyproject.toml

@@ -28,12 +28,21 @@ dev = [
     "ruff",
     "twine",
 ]
+web = [
+    "fastapi>=0.136.3",
+    "httpx>=0.28.1",
+    "itsdangerous>=2.2.0",
+    "jinja2>=3.1.6",
+    "python-multipart>=0.0.30",
+    "uvicorn[standard]>=0.48.0",
+]
 
 [project.urls]
 homepage = "https://github.com/jmlrt/spotfm"
 
 [project.scripts]
 spfm = "spotfm.cli:main"
+spfm-web = "spotfm.web.app:run"
 
 [tool.ruff]
 line-length = 120
@@ -66,6 +75,7 @@ skip-magic-trailing-comma = false
 line-ending = "auto"
 
 [tool.pytest.ini_options]
+asyncio_mode = "auto"
 testpaths = ["tests"]
 python_files = ["test_*.py"]
 python_classes = ["Test*"]
@@ -109,3 +119,8 @@ exclude_lines = [
     "if __name__ == .__main__.:",
     "if TYPE_CHECKING:",
 ]
+
+[dependency-groups]
+dev = [
+    "pytest-asyncio>=1.4.0",
+]

spotfm.example.toml

@@ -1,3 +1,6 @@
+[web]
+api_key = ""  # any random string; required to access the webapp
+
 [lastfm]
 api_key = ""
 api_secret = ""

spotfm/cli.py

@@ -8,7 +8,7 @@
 from logging.handlers import RotatingFileHandler
 
 from spotfm import lastfm, utils
-from spotfm.lastfm import read_lastfm_state, save_lastfm_state
+from spotfm.lastfm import fetch_recent_scrobbles, read_lastfm_state
 from spotfm.spotify import client as spotify_client
 from spotfm.spotify import constants as spotify_constants
 from spotfm.spotify import dupes as spotify_dupes
@@ -31,44 +31,45 @@ def _non_negative_int(value):
     return ivalue
 
 
-def recent_scrobbles(user, limit, scrobbles_minimum, period, period_minimum, interactive):
-    current_count = user.get_playcount()
-    scrobble_count_to_save = current_count  # Track what state to save
+def _format_track(track_dict):
+    """Format a track dict as 'artist - title - period_scrobbles - total_scrobbles - url'."""
+    return f"{track_dict['artist']} - {track_dict['title']} - {track_dict['period_scrobbles']} - {track_dict['total_scrobbles']} - {track_dict['url']}"
 
-    state = read_lastfm_state()
-    if state is None:
-        # First run: initialize state with current count, fetch --limit scrobbles
-        print(f"Initializing scrobble tracking. Fetching up to {limit} recent scrobbles.")
-    else:
-        # Subsequent runs: fetch all new scrobbles since last run
-        last_scrobble_count = None
-        if isinstance(state, dict):
-            last_scrobble_count = state.get("last_scrobble_count")
-        if not isinstance(last_scrobble_count, int):
-            print("Invalid previous state. Re-initializing scrobble tracking.")
-            save_lastfm_state(current_count)
-            return
-        computed_limit = current_count - last_scrobble_count
-        if computed_limit <= 0:
-            print("No new scrobbles since last run.")
-            save_lastfm_state(current_count)
-            return
-        # Fetch all new scrobbles on subsequent runs
-        limit = computed_limit
-        print(f"Fetching {limit} new scrobbles (was {last_scrobble_count}, now {current_count}).")
 
-    # Get scrobbles generator
-    scrobbles_gen = user.get_recent_tracks_scrobbles(limit, scrobbles_minimum, period, period_minimum)
+def recent_scrobbles(user, limit, scrobbles_minimum, period, period_minimum, interactive, config):
+    # Check if this is first run BEFORE fetching (once fetch_recent_scrobbles runs, state is saved)
+    is_first_run = read_lastfm_state() is None
 
-    if interactive:
-        # Collect all results first so exceptions/logs surface before editor opens
-        scrobbles = list(scrobbles_gen)
-        lines = sorted(set(scrobbles))
-        if not lines:
-            print("No results to open in editor.")
-            save_lastfm_state(scrobble_count_to_save)
-            return
+    # Fetch scrobbles with incremental state management (orchestration in lastfm module)
+    tracks, mode = fetch_recent_scrobbles(
+        user, config, limit=limit, scrobbles_minimum=scrobbles_minimum, period=period, period_minimum=period_minimum
+    )
+
+    # CLI-layer presentation: handle user feedback and interactive mode
+    if mode == "no_new":
+        print("No new scrobbles since last run.")
+        return
+    elif mode == "incremental":
+        state = read_lastfm_state()
+        last_count = state.get("last_scrobble_count") if isinstance(state, dict) else None
+        current_count = user.get_playcount()
+        if last_count:
+            print(f"Fetching {len(tracks)} new scrobbles (was {last_count}, now {current_count}).")
+        else:
+            print(f"Initializing scrobble tracking. Fetching up to {limit} recent scrobbles.")
+    else:  # mode == "full"
+        # First run shows initialization message; subsequent runs show fetch message
+        if is_first_run:
+            print(f"Initializing scrobble tracking. Fetching up to {limit} recent scrobbles.")
+        else:
+            print(f"Fetching {len(tracks)} recent scrobbles.")
+
+    if not tracks:
+        print("No results found.")
+        return
 
+    if interactive:
+        lines = sorted(set(_format_track(s) for s in tracks))
         editor = os.environ.get("VISUAL") or os.environ.get("EDITOR", "vim")
         editor_args = shlex.split(editor)
         with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False, encoding="utf-8", newline="\n") as f:
@@ -80,14 +81,12 @@ def recent_scrobbles(user, limit, scrobbles_minimum, period, period_minimum, int
             os.unlink(tmp)
         if result.returncode != 0:
             print(f"Editor command {' '.join(editor_args)} exited with status {result.returncode}.")
-            print("Not advancing Last.FM state; please fix the editor issue and retry.")
+            print("Last.FM state was already advanced; please fix the editor issue.")
             return
     else:
         # Stream output in non-interactive mode
-        for scrobble in scrobbles_gen:
-            print(scrobble)
-
-    save_lastfm_state(scrobble_count_to_save)
+        for track in tracks:
+            print(_format_track(track))
 
 
 def count_tracks(playlists_pattern=None):
@@ -126,7 +125,7 @@ def lastfm_cli(args, config):
             if period_minimum is None:
                 period_minimum = config.get("lastfm", {}).get("period_minimum")
 
-            recent_scrobbles(user, args.limit, scrobbles_minimum, args.period, period_minimum, args.interactive)
+            recent_scrobbles(user, args.limit, scrobbles_minimum, args.period, period_minimum, args.interactive, config.get("lastfm", {}))
 
 
 def spotify_cli(args, config):
@@ -184,7 +183,10 @@ def spotify_cli(args, config):
             spotify_dupes.find_duplicate_names(excluded_playlist_ids=excluded, threshold=threshold)
         case "find-relinked-tracks":
             excluded = config["spotify"].get("excluded_playlists", [])
-            spotify_misc.find_relinked_tracks(client.client, excluded_playlist_ids=excluded, output_file=args.output)
+            relinked = spotify_misc.find_relinked_tracks(client.client, excluded_playlist_ids=excluded, output_file=args.output)
+            if not args.output:
+                for track in relinked:
+                    print(f"Relinked - {track['playlist_name']} - {track['original_track']} -> {track['replacement_track']} - {track['original_id']} - {track['replacement_id']}")
         case "list-playlists-with-track-counts":
             playlists = spotify_misc.list_playlists_with_track_counts()
             total_entries = 0

spotfm/lastfm.py

@@ -125,9 +125,15 @@ def get_recent_tracks_scrobbles(self, limit=10, scrobbles_minimum=0, period=90,
             # Rate limiting: 0.2s between API calls (5 req/sec = Last.FM limit)
             sleep(0.2)
 
-        # Yield results (same format as before - backward compatible)
+        # Yield structured track data for callers to format as needed
         for track, period_scrobbles, total_scrobbles, url in tracks_data:
-            yield f"{track} - {period_scrobbles} - {total_scrobbles} - {url}"
+            yield {
+                "artist": track.artist,
+                "title": track.title,
+                "period_scrobbles": period_scrobbles,
+                "total_scrobbles": total_scrobbles,
+                "url": url,
+            }
 
 
 def read_lastfm_state(state_file=None):
@@ -168,3 +174,63 @@ def save_lastfm_state(scrobble_count, state_file=None):
             with contextlib.suppress(OSError):
                 tmp_path.unlink()
         logging.error(f"Failed to save Last.FM state file at {path}: {e}")
+
+
+def fetch_recent_scrobbles(user, config, *, limit=None, scrobbles_minimum=None, period=90, period_minimum=None):
+    """Fetch recent scrobbles with incremental state management.
+
+    Handles the orchestration of reading saved state, computing effective limits,
+    and saving state after fetch. This ensures consistent behavior across CLI and web.
+
+    Args:
+        user: Last.FM User instance
+        config: Last.FM config dict with keys: limit, scrobbles_minimum, period_minimum
+        limit: Explicit limit override (if None, uses incremental mode with saved state)
+        scrobbles_minimum: Minimum total scrobbles filter (if None, uses config)
+        period: Period in days for scrobble counting (default 90)
+        period_minimum: Minimum scrobbles in period window (if None, uses config)
+
+    Returns:
+        tuple: (tracks: list[dict], mode: str) where mode is "incremental", "full", or "no_new"
+    """
+    current_count = user.get_playcount()
+    scrobble_count_to_save = current_count
+
+    # Resolve config defaults
+    effective_limit = limit if limit is not None else config.get("limit", 50)
+    effective_scrobbles_minimum = (
+        scrobbles_minimum if scrobbles_minimum is not None else config.get("scrobbles_minimum", 4)
+    )
+    effective_period_minimum = period_minimum if period_minimum is not None else config.get("period_minimum")
+
+    # Check for saved state first. If it exists, use incremental mode (ignore explicit limit)
+    state = read_lastfm_state()
+    mode = "full"
+    if state is not None and isinstance(state, dict):
+        last_scrobble_count = state.get("last_scrobble_count")
+        if isinstance(last_scrobble_count, int):
+            # Saved state exists: always use incremental mode, compute limit from diff
+            mode = "incremental"
+            computed_limit = current_count - last_scrobble_count
+            if computed_limit <= 0:
+                save_lastfm_state(current_count)
+                return [], "no_new"
+            effective_limit = computed_limit
+    else:
+        # No saved state: use full mode with provided limit or config default
+        mode = "full" if limit is not None else "full"
+
+    # Fetch scrobbles
+    tracks = list(
+        user.get_recent_tracks_scrobbles(
+            limit=effective_limit,
+            scrobbles_minimum=effective_scrobbles_minimum,
+            period=period,
+            period_minimum=effective_period_minimum,
+        )
+    )
+
+    # Save state after successful fetch
+    save_lastfm_state(scrobble_count_to_save)
+
+    return tracks, mode

spotfm/spotify/misc.py

@@ -43,7 +43,6 @@
 
 import csv
 import logging
-import sys
 from datetime import datetime
 from pathlib import Path
 from time import sleep
@@ -277,7 +276,7 @@ def discover_from_playlists(client, discover_playlist_id, sources_playlists_ids)
     total_playlists = len(sources_playlists_ids)
 
     for idx, playlist_id in enumerate(sources_playlists_ids, 1):
-        print(f"fetching playlist {playlist_id} ({idx}/{total_playlists})", file=sys.stderr, flush=True)
+        logging.info(f"Fetching playlist {playlist_id} ({idx}/{total_playlists})")
         playlist = Playlist.get_playlist(playlist_id, client.client, refresh=True, sync_to_db=False)
         logging.info(f"Looking for new tracks into {playlist.id} - {playlist.name}")
 
@@ -323,9 +322,7 @@ def discover_from_playlists(client, discover_playlist_id, sources_playlists_ids)
                 # Track exists and is in other playlists
                 logging.debug(f"Skipping track {track.id} (already in playlists)")
 
-        print(f"discovered {new_this_playlist} new tracks from playlist {playlist.name}", file=sys.stderr, flush=True)
-
-    print(f"total discovered from all playlists: {len(new_tracks)} new tracks", file=sys.stderr, flush=True)
+        logging.info(f"Discovered {new_this_playlist} new tracks from playlist {playlist.name}")
 
     if len(new_tracks) > 0:
         logging.info(
@@ -473,11 +470,6 @@ def find_relinked_tracks(client, excluded_playlist_ids=None, output_file=None):
     # Output results
     if output_file:
         write_relinked_tracks_csv(relinked_tracks, output_file)
-    else:
-        for track in relinked_tracks:
-            print(
-                f"Relinked - {track['playlist_name']} - {track['original_track']} -> {track['replacement_track']} - {track['original_id']} - {track['replacement_id']}"
-            )
 
     return relinked_tracks
 

spotfm/spotify/playlist.py

@@ -56,6 +56,7 @@ def __init__(self, playlist_id, client=None, refresh=True):
         self.tracks = None  # [Track] after hydration; None until update_from_api() is called
         self.updated = None
         self.snapshot_id = None  # Spotify snapshot ID to detect unchanged playlists
+        self.snapshot_unchanged = False  # True when snapshot matched DB — skip redundant track sync
         # TODO: self._tracks_names
         # TODO: self._sorted_tracks
 
@@ -141,6 +142,7 @@ def update_from_db(self):
         return True
 
     def update_from_api(self, client):
+        self.snapshot_unchanged = False  # Reset before checking; instance may be reused across refreshes
         playlist = client.playlist(self.id, fields="name,owner.id,snapshot_id", market=MARKET)
         self.name = utils.sanitize_string(playlist["name"])
         logging.info("Fetching playlist %s - %s from api", self.id, self.name)
@@ -175,6 +177,7 @@ def update_from_api(self, client):
                 self.tracks = self.get_tracks(client, sync_to_db=False)
             # Update the last-updated marker to reflect a successful refresh
             self.updated = str(date.today())
+            self.snapshot_unchanged = True
             return
 
         self.snapshot_id = new_snapshot
@@ -220,6 +223,12 @@ def sync_to_db(self, client):
                 f"INSERT OR REPLACE INTO playlists (id, name, owner, updated_at) VALUES ('{self.id}', '{self.name}', '{self.owner}', '{self.updated}')"
             )
 
+        if self.snapshot_unchanged:
+            # Playlist content didn't change — only update metadata, skip track re-sync
+            logging.info("Playlist %s - %s snapshot unchanged, skipping track sync", self.id, self.name)
+            sqlite.query_db(sqlite.DATABASE, queries)
+            return
+
         # Delete all existing tracks for this playlist to handle removed tracks
         queries.append(f"DELETE FROM playlists_tracks WHERE playlist_id = '{self.id}'")
         # Sync all unique tracks first