Plex: Add extended recommendations with "Mixes For You" support

music_assistant/providers/plex/__init__.py

@@ -5,13 +5,15 @@
 import asyncio
 import contextlib
 import logging
+import random
 import warnings
 from asyncio import Task, TaskGroup
 from collections.abc import Awaitable
 from datetime import UTC, datetime
 from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar, cast
 
 import plexapi.exceptions
+import plexapi.utils
 import requests
 import urllib3.exceptions
 from music_assistant_models.config_entries import (
@@ -23,6 +25,7 @@
 from music_assistant_models.enums import (
     ConfigEntryType,
     ContentType,
+    ImageType,
     MediaType,
     ProviderFeature,
     StreamType,
@@ -41,6 +44,7 @@
     ItemMapping,
     MediaItem,
     MediaItemChapter,
+    MediaItemImage,
     MediaItemType,
     Playlist,
     Podcast,
@@ -75,6 +79,7 @@
     CONF_ACTION_GDM,
     CONF_AUTH_TOKEN,
     CONF_COLLECTION_PREFIX,
+    CONF_EXTENDED_RECOMMENDATIONS,
     CONF_HUB_ITEMS_LIMIT,
     CONF_IMPORT_COLLECTIONS,
     CONF_LIBRARY_ID,
@@ -95,6 +100,9 @@
     ERR_NO_ARTIST_FOR_TRACK,
     ERR_TRACK_NOT_FOUND,
     FAKE_ARTIST_PREFIX,
+    MIX_CACHE_EXPIRATION,
+    MIX_ITEM_PREFIX,
+    RECOMMENDATIONS_HUB_PARAMS,
 )
 from music_assistant.providers.plex.helpers import (
     AUDIOBOOK_FEATURES,
@@ -482,6 +490,14 @@ async def get_config_entries(  # noqa: PLR0915
             range=(1, 100),
         )
     )
+    entries.append(
+        ConfigEntry(
+            key=CONF_EXTENDED_RECOMMENDATIONS,
+            type=ConfigEntryType.BOOLEAN,
+            default_value=True,
+            advanced=True,
+        )
+    )
 
     # return all config entries
     return tuple(entries)
@@ -1111,6 +1127,25 @@ async def get_playlist(self, prov_playlist_id: str) -> Playlist:
             plex_collection: PlexObject = await self._get_data(collection_key)
             return await self._parse_collection(plex_collection)
 
+        # "Mixes For You" items use a MIX_ITEM_PREFIX (see _build_mix_playlist).
+        if prov_playlist_id.startswith(MIX_ITEM_PREFIX):
+            mix_key = prov_playlist_id.removeprefix(MIX_ITEM_PREFIX)
+            fields = await self._find_mix_by_key(mix_key)
+            if fields is None:
+                msg = f"Mix {prov_playlist_id} not found"
+                raise MediaNotFoundError(msg)
+            _, title, thumb = fields
+            # Cache title/artwork on interaction so replay from recently-played
+            # still renders after Plex rotates the mix out of the hub.
+            if mix_key:
+                await self.mass.cache.set(
+                    key=mix_key,
+                    data={"title": title, "thumb": thumb},
+                    provider=self.instance_id,
+                    expiration=MIX_CACHE_EXPIRATION,
+                )
+            return self._build_mix_playlist(mix_key, title, thumb)
+
         plex_playlist = await self._get_data(prov_playlist_id, PlexPlaylist)
         return await self._parse_playlist(plex_playlist)
 
@@ -1142,6 +1177,21 @@ async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> lis
                         result.append(album_track)
             return result
 
+        # "Mixes For You" items use a MIX_ITEM_PREFIX. Strip it to recover
+        # the Plex section-query key, append the track type filter to expand
+        # albums into tracks, then shuffle — Plexamp randomizes mix playback
+        # client-side.
+        if prov_playlist_id.startswith(MIX_ITEM_PREFIX):
+            mix_key = prov_playlist_id.removeprefix(MIX_ITEM_PREFIX)
+            tracks_key = f"{mix_key}&type={plexapi.utils.searchType('track')}"
+            plex_tracks = await self._run_async(self._plex_library.fetchItems, tracks_key)
+            random.shuffle(plex_tracks)
+            for index, plex_track in enumerate(plex_tracks, 1):
+                if track := await self._parse_track(plex_track):
+                    track.position = index
+                    result.append(track)
+            return result
+
         plex_playlist: PlexPlaylist = await self._get_data(prov_playlist_id, PlexPlaylist)
         if not (playlist_items := await self._run_async(plex_playlist.items)):
             return result
@@ -1208,18 +1258,19 @@ async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[
             self.logger.warning("Error getting similar tracks for %s: %s", prov_track_id, err)
         return []
 
-    @use_cache(3600 * 3, cache_checksum="v2")  # Cache for 3 hours
+    @use_cache(3600 * 3, cache_checksum="v3")  # Cache for 3 hours
     async def recommendations(self) -> list[RecommendationFolder]:
         """Get recommendations from Plex hubs."""
         try:
             # Get the configured limit for items per hub
             limit_value = self.config.get_value(CONF_HUB_ITEMS_LIMIT)
             limit = int(limit_value) if isinstance(limit_value, (int, float, str)) else 10
 
-            # Fetch hubs from the music library section with count parameter
-            # The section's hubs() method uses /hubs/sections/{key}?includeStations=1
-            # We need to add the count parameter manually to limit items per hub
-            key = f"/hubs/sections/{self._plex_library.key}?includeStations=1&count={limit}"
+            # Build the hubs key manually because plexapi's hubs() method
+            # doesn't accept a count parameter to limit items per hub.
+            extended = self.config.get_value(CONF_EXTENDED_RECOMMENDATIONS)
+            hub_params = RECOMMENDATIONS_HUB_PARAMS if extended else "includeStations=1"
+            key = f"/hubs/sections/{self._plex_library.key}?count={limit}&{hub_params}"
             hubs = await self._run_async(self._plex_library.fetchItems, key)
 
             if not hubs:
@@ -1242,6 +1293,17 @@ async def recommendations(self) -> list[RecommendationFolder]:
                     icon="mdi-music",
                 )
 
+                # Mixes For You are synthetic smart playlists; build them from
+                # their partial hub items (see _mix_playlist_fields).
+                if "music.mixes" in (hub.hubIdentifier or ""):
+                    folder.items.extend(
+                        self._build_mix_playlist(*self._mix_playlist_fields(plex_mix))
+                        for plex_mix in hub._partialItems
+                    )
+                    if folder.items:
+                        folders.append(folder)
+                    continue
+
                 # Parse each item based on its type (limit to configured max)
                 # Use _partialItems to respect the count limit from the hubs() call
                 # rather than hub.items() which fetches ALL items if more is True
@@ -1652,6 +1714,91 @@ async def _parse_collection(self, plex_collection: PlexCollection) -> Playlist:
         playlist.is_editable = False
         return playlist
 
+    def _mix_playlist_fields(self, plex_mix: PlexPlaylist) -> tuple[str, str, str | None]:
+        """
+        Extract (smart-query key, title, centroid thumb) from a 'Mix For You' item.
+
+        :param plex_mix: A Plex Playlist parsed from the 'Mixes For You' hub.
+        """
+        # Read straight from the parsed XML element. These synthetic mix playlists
+        # carry a centroid-derived ratingKey rather than their own, so touching any
+        # attribute that triggers a reload (e.g. .thumb) re-fetches the wrong object
+        # and corrupts it. The smart-query key, title, and centroid artist thumb are
+        # all present on the partial element itself.
+        data = plex_mix._data
+        mix_key = data.get("key") or ""
+        title = data.get("title") or "[Unknown Mix]"
+        thumb = next(
+            (child.get("thumb") for child in data if child.get("centroid") and child.get("thumb")),
+            None,
+        )
+        return mix_key, title, thumb
+
+    def _build_mix_playlist(self, mix_key: str, title: str, thumb: str | None) -> Playlist:
+        """
+        Build a MA Playlist from a Plex 'Mix For You' hub item.
+
+        :param mix_key: The Plex smart-query key identifying the mix.
+        :param title: The mix title.
+        :param thumb: The centroid artist thumb path, if any.
+        """
+        item_id = f"{MIX_ITEM_PREFIX}{mix_key}"
+        playlist = Playlist(
+            item_id=item_id,
+            provider=self.instance_id,
+            name=title,
+            provider_mappings={
+                ProviderMapping(
+                    item_id=item_id,
+                    provider_domain=self.domain,
+                    provider_instance=self.instance_id,
+                )
+            },
+        )
+        if thumb:
+            playlist.metadata.images = UniqueList(
+                [
+                    MediaItemImage(
+                        type=ImageType.THUMB,
+                        path=thumb,
+                        provider=self.instance_id,
+                        remotely_accessible=False,
+                    )
+                ]
+            )
+        playlist.is_editable = False
+        playlist.is_dynamic = True
+        return playlist
+
+    async def _get_mix_playlists(self, count: int) -> list[PlexPlaylist]:
+        """
+        Fetch the 'Mixes For You' hub items as Plex Playlist objects.
+
+        :param count: Maximum number of items per hub.
+        """
+        key = f"/hubs/sections/{self._plex_library.key}?count={count}&{RECOMMENDATIONS_HUB_PARAMS}"
+        hubs = await self._run_async(self._plex_library.fetchItems, key)
+        for hub in hubs:
+            if "music.mixes" in (hub.hubIdentifier or ""):
+                return list(hub._partialItems)
+        return []
+
+    async def _find_mix_by_key(self, mix_key: str) -> tuple[str, str, str | None] | None:
+        """Find a 'Mix For You' by its smart-query key, falling back to cache."""
+        limit_value = self.config.get_value(CONF_HUB_ITEMS_LIMIT)
+        limit = int(limit_value) if isinstance(limit_value, (int, float, str)) else 10
+        for plex_mix in await self._get_mix_playlists(limit):
+            fields = self._mix_playlist_fields(plex_mix)
+            if fields[0] == mix_key:
+                return fields
+        # Plex rotates mixes out of the hub, but the smart-query key remains a
+        # valid section query, so replay from recently-played still works — we
+        # only need the cache to restore the title and artwork.
+        cached = await self.mass.cache.get(key=mix_key, provider=self.instance_id)
+        if isinstance(cached, dict):
+            return mix_key, cached.get("title") or "[Unknown Mix]", cached.get("thumb")
+        return None
+
     async def _parse_track(self, plex_track: PlexTrack) -> Track:
         """Parse a Plex Track response to a Track model object."""
         content = plex_track.media[0].container if plex_track.media else None

music_assistant/providers/plex/constants.py

@@ -20,6 +20,7 @@
 CONF_PLEX_FAVORITE_THRESHOLD = "plex_favorite_threshold"
 CONF_PLEX_UNLIKE_RATING = "plex_unlike_rating"
 CONF_HUB_ITEMS_LIMIT = "hub_items_limit"
+CONF_EXTENDED_RECOMMENDATIONS = "extended_recommendations"
 
 FAKE_ARTIST_PREFIX = "_fake://"
 
@@ -29,6 +30,27 @@
 # item_id prefix used for Plex collections imported as playlists
 COLLECTION_ID_PREFIX = "collection:"
 
+# item_id prefix used for Plex "Mixes For You" playlists
+MIX_ITEM_PREFIX = "mix:"
+
+# Mix title/artwork are cached so replay from recently-played survives Plex
+# rotating the mix out of its hub. 90 days is chosen to outlive MA's
+# playlog retention (see controllers/music.py: _cleanup_database).
+MIX_CACHE_EXPIRATION = 86400 * 90
+
+# Query parameters passed to /hubs/sections when loading recommendations.
+# Some of these are not in public Plex docs but are required to surface
+# the "Mixes For You" hub and library playlists.
+RECOMMENDATIONS_HUB_PARAMS = (
+    "includeMyMixes=1"
+    "&includeStations=1"
+    "&includeLibraryPlaylists=1"
+    "&includeExternalMetadata=1"
+    "&includeAnniversaryReleases=1"
+    "&excludeElements=Similar,Mood"
+    "&excludeFields=summary"
+)
+
 # error messages (templates use str.format)
 ERR_INVALID_CREDENTIALS = "Invalid login credentials"
 ERR_AUTH_FAILED = "Authentication failed"

music_assistant/providers/plex/strings.json

@@ -54,6 +54,10 @@
       "label": "Items per hub",
       "description": "Maximum number of items to load from each hub (default: 10)"
     },
+    "extended_recommendations": {
+      "label": "Extended recommendations",
+      "description": "Show additional recommendation hubs in the dashboard, including 'Mixes For You', recent library playlists, and 'On This Day' releases. Matches the hub set Plex's own clients surface. Changes take effect on the next recommendations refresh."
+    },
     "gdm": {
       "label": "Use Plex GDM to discover local servers",
       "description": "Enable \"GDM\" to discover local Plex servers automatically.",

music_assistant/translations/en.json

@@ -1591,6 +1591,8 @@
   "provider.plex.config_entries.auth_myplex.label": "Authenticate with MyPlex",
   "provider.plex.config_entries.collection_prefix.description": "Prefix to add to collection names when imported as playlists",
   "provider.plex.config_entries.collection_prefix.label": "Collection Prefix",
+  "provider.plex.config_entries.extended_recommendations.description": "Show additional recommendation hubs in the dashboard, including 'Mixes For You', recent library playlists, and 'On This Day' releases. Matches the hub set Plex's own clients surface. Changes take effect on the next recommendations refresh.",
+  "provider.plex.config_entries.extended_recommendations.label": "Extended recommendations",
   "provider.plex.config_entries.gdm.action_label": "Use Plex GDM to discover local servers",
   "provider.plex.config_entries.gdm.description": "Enable \"GDM\" to discover local Plex servers automatically.",
   "provider.plex.config_entries.gdm.label": "Use Plex GDM to discover local servers",