feat: have playlist covers auto-generate if not set

Author: lukehmcc

pyproject.toml

@@ -11,6 +11,7 @@ dependencies = [
   "mpris-server>=0.8.1",
   "platformdirs>=4.0.0",
   "spotifyscraper>=3.9.0",
+  "Pillow>=10.0.0",
   "pyobjc-framework-MediaPlayer>=10.0; platform_system == 'Darwin'",
   "pyobjc-framework-Cocoa>=10.0; platform_system == 'Darwin'",
 ]

skimmer/playlist.py

@@ -104,6 +104,66 @@ def resolve_cover(playlist: Playlist) -> str | None:
     return None
 
 
+def _find_album_cover(track: PlaylistTrack) -> str | None:
+    if not track.file_path:
+        return None
+    album_dir = os.path.dirname(track.file_path)
+    if not os.path.isdir(album_dir):
+        return None
+    for name in ("cover.jpg", "cover.png", "front.jpg", "folder.jpg", "Cover.jpg"):
+        path = os.path.join(album_dir, name)
+        if os.path.exists(path):
+            return path
+    return None
+
+
+def generate_playlist_cover(playlist: Playlist, music_dir: str) -> str | None:
+    """Generate cover from track album covers:
+    - 1-3 tracks: first track's cover (256x256)
+    - 4+ tracks: 2x2 grid of first 4 covers (512x512)
+    Returns path to saved cover in COVERS_DIR/{uuid}.jpg, or None if no covers found.
+    """
+    if not playlist.tracks:
+        return None
+
+    covers = []
+    for track in playlist.tracks[:4]:
+        cover_path = _find_album_cover(track)
+        if cover_path:
+            covers.append(cover_path)
+
+    if not covers:
+        return None
+
+    try:
+        from PIL import Image
+    except ImportError:
+        return None
+
+    COVERS_DIR.mkdir(parents=True, exist_ok=True)
+    out_path = COVERS_DIR / f"{playlist.uuid}.jpg"
+
+    if len(covers) == 1:
+        img = Image.open(covers[0]).convert("RGB")
+        img = img.resize((256, 256), Image.Resampling.LANCZOS)
+    else:
+        grid_size = 512
+        tile_size = 256
+        img = Image.new("RGB", (grid_size, grid_size), (0x22, 0x22, 0x22))
+        for i, cp in enumerate(covers[:4]):
+            try:
+                tile = Image.open(cp).convert("RGB")
+                tile = tile.resize((tile_size, tile_size), Image.Resampling.LANCZOS)
+            except Exception:
+                continue
+            x = (i % 2) * tile_size
+            y = (i // 2) * tile_size
+            img.paste(tile, (x, y))
+
+    img.save(out_path, "JPEG", quality=85)
+    return str(out_path)
+
+
 def export_m3u8(playlist: Playlist, out_path: str, paths: list[str] | None = None):
     """Write an .m3u8 file.
 

skimmer/playlists_ui.py

@@ -15,6 +15,7 @@
     Playlist,
     PlaylistTrack,
     export_m3u8,
+    generate_playlist_cover,
     load_playlists,
     parse_m3u8,
     resolve_cover,
@@ -472,6 +473,7 @@ def _remove_track_at(self, idx):
         del pl.tracks[idx]
         pl.last_modified = time.time()
         self._save()
+        self._maybe_regenerate_cover(pl)
         self._build_detail_for(self._detail_playlist_index)
         self._rebuild_grid()
 
@@ -638,6 +640,20 @@ def _do_delete(self, dialog, response, idx):
             self.stack.set_visible_child_name("grid")
         self._rebuild_grid()
 
+    def _maybe_regenerate_cover(self, pl: Playlist):
+        """Regenerate playlist cover from track album covers if not manually set."""
+        if pl.cover_path:
+            return  # Preserve manual/Spotify cover
+        music_dir = self.config.get("music_dir", "")
+        if not music_dir:
+            return
+        cover_path = generate_playlist_cover(pl, music_dir)
+        if cover_path:
+            pl.cover_path = cover_path
+            self._save()
+            self._build_detail_for(self._detail_playlist_index)
+            self._rebuild_grid()
+
     def _on_add_tracks(self, *args):
         if self._detail_playlist_index < 0:
             return
@@ -648,6 +664,7 @@ def add_tracks(tracks):
                 pl.tracks.append(t)
             pl.last_modified = time.time()
             self._save()
+            self._maybe_regenerate_cover(pl)
             self._build_detail_for(self._detail_playlist_index)
             self._rebuild_grid()
 
@@ -743,9 +760,11 @@ def _add_imported(self, pl):
         if existing:
             existing[0].tracks = pl.tracks
             existing[0].last_modified = time.time()
+            pl = existing[0]
         else:
             self._playlists.append(pl)
         self._save()
+        self._maybe_regenerate_cover(pl)
         self._rebuild_grid()
 
     def _on_import_spotify(self, *args):

skimmer/spotify_import.py

@@ -87,9 +87,16 @@ def import_playlist(self, url):
         self._status(f'Found playlist "{playlist_name}" — {total} tracks')
 
         lib = None
-        if os.path.exists(beets_db):
+        try:
+            os.makedirs(music_dir, exist_ok=True)
             beets_context.set_music_dir(bytestring_path(music_dir))
             lib = Library(beets_db, directory=music_dir)
+        except Exception as e:
+            log.error(f"[skimmer] Failed to open beets library {beets_db}: {e}")
+            lib = None
+
+        if lib:
+            self._cleanup_orphans(lib, music_dir)
 
         matched = 0
         to_download = []
@@ -162,7 +169,43 @@ def import_playlist(self, url):
         self._status("Import complete")
         return result
 
+    def _cleanup_orphans(self, lib, music_dir):
+        """Delete audio files under music_dir that are not indexed in beets."""
+        try:
+            known = {
+                os.fsdecode(i.path)
+                for i in lib.items()
+                if i.path
+            }
+        except Exception as e:
+            log.warning(f"[skimmer] Could not read beets items during cleanup: {e}")
+            return
+
+        removed = 0
+        for root, _dirs, files in os.walk(music_dir):
+            for fname in files:
+                if fname.startswith(".") and fname.endswith(".partial"):
+                    continue
+                if not fname.lower().endswith((".mp3", ".m4a", ".flac", ".opus")):
+                    continue
+                fpath = os.path.join(root, fname)
+                if fpath in known or os.path.realpath(fpath) in known:
+                    continue
+                try:
+                    os.remove(fpath)
+                    removed += 1
+                    log.info(f"[skimmer] Removed orphan: {fpath}")
+                except OSError as e:
+                    log.warning(f"[skimmer] Failed to remove orphan {fpath}: {e}")
+        if removed:
+            log.info(f"[skimmer] Cleanup removed {removed} orphan audio file(s)")
+
     def _download_tracks(self, to_download, music_dir, lib, failed):
+        if lib is None:
+            raise SpotifyImportError(
+                "Beets library is unavailable; refusing to import so files are not orphaned."
+            )
+
         temp_dir = tempfile.mkdtemp(prefix="skimmer-spotify-")
         log.info(f"[skimmer] Download temp dir: {temp_dir}")
 
@@ -256,7 +299,9 @@ def _download_tracks(self, to_download, music_dir, lib, failed):
                     os.makedirs(dest_dir, exist_ok=True)
 
                     dest_path = os.path.join(dest_dir, os.path.basename(found_file))
-                    shutil.copy2(found_file, dest_path)
+                    partial_path = os.path.join(dest_dir, f".{os.path.basename(found_file)}.partial")
+                    shutil.copy2(found_file, partial_path)
+                    os.replace(partial_path, dest_path)
                     entry["file_path"] = dest_path
 
                     # Download album cover
@@ -272,6 +317,19 @@ def _download_tracks(self, to_download, music_dir, lib, failed):
 
                     log.info(f"[skimmer] Copied to: {dest_path}")
 
+                    # Index immediately so a mid-import abort never leaves an orphan file.
+                    try:
+                        self._import_one_track(lib, music_dir, entry, dest_path)
+                    except Exception as e:
+                        log.warning(f"[skimmer] Beets index failed for {dest_path}: {e}")
+                        try:
+                            os.remove(dest_path)
+                        except OSError:
+                            pass
+                        failed.append(f"{artist} - {title}")
+                        entry["file_path"] = ""
+                        continue
+
                 except SpotifyImportError:
                     raise
                 except Exception as e:
@@ -280,20 +338,6 @@ def _download_tracks(self, to_download, music_dir, lib, failed):
                     log.warning(f"[skimmer] Download failed for {artist} - {title}: {e}")
                     failed.append(f"{artist} - {title}")
 
-            # Import new files into beets
-            if lib:
-                new_tracks = [
-                    (t["file_path"], t["artist"], t["album"])
-                    for t in to_download
-                    if t["file_path"] and t["file_path"].startswith(music_dir)
-                ]
-                if new_tracks:
-                    self._status("Importing into beets library...")
-                    try:
-                        self._import_to_beets(lib, music_dir, new_tracks)
-                    except Exception as e:
-                        log.warning(f"[skimmer] Beets import error: {e}", exc_info=True)
-
         except SpotifyImportError:
             raise
         except Exception as e:
@@ -342,53 +386,33 @@ def _tag_file(self, fpath, entry):
         except Exception as e:
             log.warning(f"[skimmer] Failed to tag {fpath}: {e}")
 
-    def _import_to_beets(self, lib, music_dir, track_info_list):
-        seen_albums = {}
-        for fpath, artist, album in track_info_list:
-            if not os.path.exists(fpath) or os.path.getsize(fpath) == 0:
-                continue
-            try:
-                item = Item.from_path(fpath)
-                item.add(lib)
-                key = (artist, album)
-                if key not in seen_albums:
-                    seen_albums[key] = []
-                seen_albums[key].append(item)
-            except Exception as e:
-                log.warning(f"[skimmer] Failed to add {fpath}: {e}")
-
-        for (artist, album_title), items in seen_albums.items():
-            if items:
-                try:
-                    album_obj = lib.add_album(items)
-                    album_obj.genre = "Spotify Import"
-                    album_obj.store()
-                    log.info(f"[skimmer] Created album '{album_obj.album}' (id={album_obj.id})")
-                    try:
-                        from beets.autotag.match import tag_album
-
-                        album_items = list(album_obj.items())
-                        _, _, proposal = tag_album(
-                            album_items, search_artist=artist, search_name=album_title
-                        )
-                        if proposal and proposal.candidates:
-                            match = proposal.candidates[0]
-                            match.apply_metadata()
-                            for item in match.mapping:  # pyright: ignore[reportAttributeAccessIssue]
-                                item.try_write()
-                            album_obj.albumartist = match.info.artist
-                            album_obj.album = match.info.album
-                            album_obj.store()
-                            log.info(
-                                f"[skimmer] Autotagged: {match.info.artist} - {match.info.album}"
-                            )
-                    except Exception as e:
-                        log.warning(f"[skimmer] Autotag skipped: {e}")
-                except Exception as e:
-                    log.warning(f"[skimmer] Failed to create album {album_title}: {e}")
-
-        if lib:
-            try:
-                lib.store()
-            except Exception:
-                pass
+    def _import_one_track(self, lib, music_dir, entry, fpath):
+        artist = entry["artist"]
+        album_title = entry["album"] or entry["title"]
+        item = Item.from_path(fpath)
+        item.add(lib)
+        album_obj = lib.add_album([item])
+        album_obj.genre = "Spotify Import"
+        album_obj.store()
+        log.info(
+            f"[skimmer] Indexed '{item.title}' -> album "
+            f"'{album_obj.album}' (id={album_obj.id})"
+        )
+        try:
+            from beets.autotag.match import tag_album
+
+            album_items = list(album_obj.items())
+            _, _, proposal = tag_album(
+                album_items, search_artist=artist, search_name=album_title
+            )
+            if proposal and proposal.candidates:
+                match = proposal.candidates[0]
+                match.apply_metadata()
+                for item in match.mapping:  # pyright: ignore[reportAttributeAccessIssue]
+                    item.try_write()
+                album_obj.albumartist = match.info.artist
+                album_obj.album = match.info.album
+                album_obj.store()
+                log.info(f"[skimmer] Autotagged: {match.info.artist} - {match.info.album}")
+        except Exception as e:
+            log.warning(f"[skimmer] Autotag skipped: {e}")

tests/test_spotify_import.py

@@ -0,0 +1,57 @@
+import os
+import tempfile
+
+import pytest
+from beets.library import Item, Library
+
+from skimmer.spotify_import import SpotifyImporter, SpotifyImportError
+
+
+def _tmp_lib():
+    tmp = tempfile.mkdtemp()
+    db = os.path.join(tmp, "b.db")
+    music = os.path.join(tmp, "music")
+    os.makedirs(music, exist_ok=True)
+    lib = Library(db, directory=music)
+    return tmp, music, lib
+
+
+def _write_audio(path):
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, "wb") as f:
+        f.write(b"\x00" * 128)
+
+
+def test_cleanup_orphans_removes_unindexed_files():
+    _t, music, lib = _tmp_lib()
+    orphan = os.path.join(music, "Orphan Artist", "Orphan Album", "orphan.mp3")
+    _write_audio(orphan)
+    importer = SpotifyImporter({})
+    importer._cleanup_orphans(lib, music)
+    assert not os.path.exists(orphan)
+
+
+def test_cleanup_orphans_keeps_indexed_files():
+    _t, music, lib = _tmp_lib()
+    known = os.path.join(music, "Artist", "Album", "track.mp3")
+    _write_audio(known)
+    item = Item(path=os.fsencode(known))
+    lib.add(item)
+    importer = SpotifyImporter({})
+    importer._cleanup_orphans(lib, music)
+    assert os.path.exists(known)
+
+
+def test_cleanup_orphans_ignores_partial_files():
+    _t, music, lib = _tmp_lib()
+    partial = os.path.join(music, ".track.mp3.partial")
+    _write_audio(partial)
+    importer = SpotifyImporter({})
+    importer._cleanup_orphans(lib, music)
+    assert os.path.exists(partial)
+
+
+def test_download_tracks_aborts_without_lib():
+    importer = SpotifyImporter({})
+    with pytest.raises(SpotifyImportError):
+        importer._download_tracks([{"artist": "A", "title": "T"}], "/tmp", None, [])