[mix] one track per album, and only 75 long

src/mopidy_smartplaylists/__init__.py

@@ -23,6 +23,8 @@ class Extension(ext.Extension):
         schema["refresh_interval"] = config.Integer(optional=True, minimum=0)
         schema["search_uris"] = config.List(optional=True)
         schema["playlist_dir"] = config.String(optional=True)
+        schema["max_tracks"] = config.Integer(optional=True, minimum=0)
+        schema["max_per_album"] = config.Integer(optional=True, minimum=0)
         return schema
 
     def validate_environment(self) -> None:

src/mopidy_smartplaylists/ext.conf

@@ -22,3 +22,7 @@ search_uris =
 # Directory to write .m3u8 playlist files into (e.g. for mopidy-m3u).
 # Leave empty to use Mopidy's core.playlists.save() API instead.
 playlist_dir =
+# Maximum number of tracks per generated playlist (0 = no limit)
+max_tracks = 75
+# Maximum tracks per album in a mix (0 = no limit)
+max_per_album = 1

src/mopidy_smartplaylists/generators.py

@@ -32,8 +32,30 @@ def _search(
     return _extract_tracks(result)
 
 
+def _mix_tracks(
+    tracks: list[Track], max_tracks: int = 0, max_per_album: int = 0,
+) -> list[Track]:
+    if not tracks:
+        return []
+    if max_per_album > 0:
+        seen: dict[str, int] = {}
+        result: list[Track] = []
+        for t in tracks:
+            key = t.album.uri if t.album and t.album.uri else str(id(t))
+            count = seen.get(key, 0)
+            if count >= max_per_album:
+                continue
+            seen[key] = count + 1
+            result.append(t)
+        tracks = result
+    if max_tracks > 0:
+        tracks = tracks[:max_tracks]
+    return tracks
+
+
 def build_decade_mix(
     core: CoreProxy, decade: str, uris: list[Uri] | None = None,
+    max_tracks: int = 0, max_per_album: int = 0,
 ) -> list[Track]:
     query = cast("Query[SearchField]", {"date": [parse_decade(decade)]})
     try:
@@ -41,21 +63,23 @@ def build_decade_mix(
     except Exception:
         logger.exception("Decade search failed for %s", decade)
         return []
-    tracks = _extract_tracks(result)
+    tracks = _mix_tracks(_extract_tracks(result), max_tracks, max_per_album)
     logger.info("Found %d tracks for decade %s", len(tracks), decade)
     return tracks
 
 
 def build_genre_mix(
     core: CoreProxy, genre: str, uris: list[Uri] | None = None,
+    max_tracks: int = 0, max_per_album: int = 0,
 ) -> list[Track]:
-    return _search(core, "genre", genre, uris=uris)
+    return _mix_tracks(_search(core, "genre", genre, uris=uris), max_tracks, max_per_album)
 
 
 def build_artist_mix(
     core: CoreProxy, artist: str, uris: list[Uri] | None = None,
+    max_tracks: int = 0, max_per_album: int = 0,
 ) -> list[Track]:
-    return _search(core, "artist", artist, uris=uris)
+    return _mix_tracks(_search(core, "artist", artist, uris=uris), max_tracks, max_per_album)
 
 
 def build_album_mix(core: CoreProxy, album_uri: str) -> list[Track]:
@@ -186,13 +210,16 @@ def refresh_smart_playlists(core: CoreProxy, config_dict: dict) -> None:
     if raw:
         uris = cast("list[Uri]", [u.strip() for u in raw.split(",") if u.strip()])
     playlist_dir = config_dict.get("playlist_dir") or None
+    max_tracks = int(config_dict.get("max_tracks", 0) or 0)
+    max_per_album = int(config_dict.get("max_per_album", 0) or 0)
 
     decades_raw = config_dict.get("decades", "")
     if decades_raw:
         decades = [d.strip() for d in decades_raw.split(",") if d.strip()]
         for decade in decades:
             try:
-                tracks = build_decade_mix(core, decade, uris=uris)
+                tracks = build_decade_mix(core, decade, uris=uris,
+                                          max_tracks=max_tracks, max_per_album=max_per_album)
             except Exception:
                 logger.exception("Failed to build decade mix for %s", decade)
                 continue
@@ -204,7 +231,8 @@ def refresh_smart_playlists(core: CoreProxy, config_dict: dict) -> None:
         genres = [d.strip() for d in genres_raw.split(",") if d.strip()]
         for genre in genres:
             try:
-                tracks = build_genre_mix(core, genre, uris=uris)
+                tracks = build_genre_mix(core, genre, uris=uris,
+                                         max_tracks=max_tracks, max_per_album=max_per_album)
             except Exception:
                 logger.exception("Failed to build genre mix for %s", genre)
                 continue
@@ -216,7 +244,8 @@ def refresh_smart_playlists(core: CoreProxy, config_dict: dict) -> None:
         artists = [a.strip() for a in artists_raw.split(",") if a.strip()]
         for artist in artists:
             try:
-                tracks = build_artist_mix(core, artist, uris=uris)
+                tracks = build_artist_mix(core, artist, uris=uris,
+                                          max_tracks=max_tracks, max_per_album=max_per_album)
             except Exception:
                 logger.exception("Failed to build artist mix for %s", artist)
                 continue

src/mopidy_smartplaylists/web.py

@@ -29,11 +29,14 @@ class DecadeMixHandler(tornado.web.RequestHandler):
     def initialize(
         self, core: CoreProxy, prefix: str, uris: list[Uri] | None = None,
         playlist_dir: str | None = None,
+        max_tracks: int = 0, max_per_album: int = 0,
     ) -> None:
         self.core = core
         self.prefix = prefix
         self.uris = uris
         self.playlist_dir = playlist_dir
+        self.max_tracks = max_tracks
+        self.max_per_album = max_per_album
 
     def post(self) -> None:
         data = json.loads(self.request.body)
@@ -42,7 +45,10 @@ class DecadeMixHandler(tornado.web.RequestHandler):
             self.set_status(400)
             self.write({"error": "Missing 'decade' in request body"})
             return
-        tracks = build_decade_mix(self.core, decade, uris=self.uris)
+        tracks = build_decade_mix(
+            self.core, decade, uris=self.uris,
+            max_tracks=self.max_tracks, max_per_album=self.max_per_album,
+        )
         if not tracks:
             self.write({"playlist": None, "tracks": 0})
             return
@@ -65,11 +71,14 @@ class GenreMixHandler(tornado.web.RequestHandler):
     def initialize(
         self, core: CoreProxy, prefix: str, uris: list[Uri] | None = None,
         playlist_dir: str | None = None,
+        max_tracks: int = 0, max_per_album: int = 0,
     ) -> None:
         self.core = core
         self.prefix = prefix
         self.uris = uris
         self.playlist_dir = playlist_dir
+        self.max_tracks = max_tracks
+        self.max_per_album = max_per_album
 
     def post(self) -> None:
         data = json.loads(self.request.body)
@@ -78,7 +87,10 @@ class GenreMixHandler(tornado.web.RequestHandler):
             self.set_status(400)
             self.write({"error": "Missing 'genre' in request body"})
             return
-        tracks = build_genre_mix(self.core, genre, uris=self.uris)
+        tracks = build_genre_mix(
+            self.core, genre, uris=self.uris,
+            max_tracks=self.max_tracks, max_per_album=self.max_per_album,
+        )
         if not tracks:
             self.write({"playlist": None, "tracks": 0})
             return
@@ -101,11 +113,14 @@ class ArtistMixHandler(tornado.web.RequestHandler):
     def initialize(
         self, core: CoreProxy, prefix: str, uris: list[Uri] | None = None,
         playlist_dir: str | None = None,
+        max_tracks: int = 0, max_per_album: int = 0,
     ) -> None:
         self.core = core
         self.prefix = prefix
         self.uris = uris
         self.playlist_dir = playlist_dir
+        self.max_tracks = max_tracks
+        self.max_per_album = max_per_album
 
     def post(self) -> None:
         data = json.loads(self.request.body)
@@ -114,7 +129,10 @@ class ArtistMixHandler(tornado.web.RequestHandler):
             self.set_status(400)
             self.write({"error": "Missing 'artist' in request body"})
             return
-        tracks = build_artist_mix(self.core, artist, uris=self.uris)
+        tracks = build_artist_mix(
+            self.core, artist, uris=self.uris,
+            max_tracks=self.max_tracks, max_per_album=self.max_per_album,
+        )
         if not tracks:
             self.write({"playlist": None, "tracks": 0})
             return
@@ -287,18 +305,30 @@ def _parse_playlist_dir(config: Config) -> str | None:
     return raw.strip() if raw else None
 
 
+def _parse_int(config: Config, key: str, default: int) -> int:
+    try:
+        return int(config.get("smartplaylists", {}).get(key, default) or default)
+    except (ValueError, TypeError):
+        return default
+
+
 def app_factory(config: Config, core: CoreProxy) -> list[tuple]:
     prefix = config.get("smartplaylists", {}).get("playlist_prefix", "[Smart]")
     uris = _parse_search_uris(config)
     playlist_dir = _parse_playlist_dir(config)
+    max_tracks = _parse_int(config, "max_tracks", 0)
+    max_per_album = _parse_int(config, "max_per_album", 0)
 
     return [
         (r"/decade", DecadeMixHandler,
-         {"core": core, "prefix": prefix, "uris": uris, "playlist_dir": playlist_dir}),
+         {"core": core, "prefix": prefix, "uris": uris, "playlist_dir": playlist_dir,
+          "max_tracks": max_tracks, "max_per_album": max_per_album}),
         (r"/genre", GenreMixHandler,
-         {"core": core, "prefix": prefix, "uris": uris, "playlist_dir": playlist_dir}),
+         {"core": core, "prefix": prefix, "uris": uris, "playlist_dir": playlist_dir,
+          "max_tracks": max_tracks, "max_per_album": max_per_album}),
         (r"/artist", ArtistMixHandler,
-         {"core": core, "prefix": prefix, "uris": uris, "playlist_dir": playlist_dir}),
+         {"core": core, "prefix": prefix, "uris": uris, "playlist_dir": playlist_dir,
+          "max_tracks": max_tracks, "max_per_album": max_per_album}),
         (r"/album", AlbumMixHandler,
          {"core": core, "prefix": prefix, "playlist_dir": playlist_dir}),
         (r"/instant-mix", InstantMixHandler,