music-assistant-server

1"""Plex musicprovider support for MusicAssistant."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import random
8import warnings
9from asyncio import Task, TaskGroup
10from collections.abc import Awaitable
11from datetime import UTC, datetime
12from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar, cast
13
14import plexapi.exceptions
15import plexapi.utils
16import requests
17import urllib3.exceptions
18from music_assistant_models.config_entries import (
19    ConfigEntry,
20    ProviderConfig,
21)
22from music_assistant_models.enums import (
23    ConfigEntryType,
24    ContentType,
25    ImageType,
26    MediaType,
27    ProviderFeature,
28    StreamType,
29)
30from music_assistant_models.errors import (
31    InvalidDataError,
32    LoginFailed,
33    MediaNotFoundError,
34    SetupFailedError,
35)
36from music_assistant_models.media_items import (
37    Album,
38    Artist,
39    Audiobook,
40    AudioFormat,
41    BrowseFolder,
42    ItemMapping,
43    MediaItem,
44    MediaItemChapter,
45    MediaItemImage,
46    MediaItemType,
47    Playlist,
48    Podcast,
49    PodcastEpisode,
50    ProviderMapping,
51    RecommendationFolder,
52    SearchResults,
53    Track,
54    UniqueList,
55)
56from music_assistant_models.streamdetails import MultiPartPath, StreamDetails
57from plexapi.audio import Album as PlexAlbum
58from plexapi.audio import Artist as PlexArtist
59from plexapi.audio import Track as PlexTrack
60from plexapi.base import PlexObject
61from plexapi.myplex import MyPlexAccount
62from plexapi.playlist import Playlist as PlexPlaylist
63from plexapi.server import PlexServer
64
65from music_assistant.constants import DB_TABLE_PROVIDER_MAPPINGS, UNKNOWN_ARTIST
66from music_assistant.controllers.cache import use_cache
67from music_assistant.helpers.tags import async_parse_tags, clean_mbid
68from music_assistant.helpers.util import parse_title_and_version
69from music_assistant.models.music_provider import MusicProvider
70from music_assistant.models.recommendation_payload import RecommendationPayloadMixin
71from music_assistant.providers.plex.constants import (
72    AUTH_TOKEN_UNAUTH,
73    COLLECTION_ID_PREFIX,
74    CONF_AUTH_TOKEN,
75    CONF_COLLECTION_PREFIX,
76    CONF_EXTENDED_RECOMMENDATIONS,
77    CONF_HUB_ITEMS_LIMIT,
78    CONF_IMPORT_COLLECTIONS,
79    CONF_LIBRARY_ID,
80    CONF_LOCAL_SERVER_IP,
81    CONF_LOCAL_SERVER_PORT,
82    CONF_LOCAL_SERVER_SSL,
83    CONF_LOCAL_SERVER_VERIFY_CERT,
84    CONF_PLEX_FAVORITE_THRESHOLD,
85    CONF_PLEX_LIKE_RATING,
86    CONF_PLEX_UNLIKE_RATING,
87    ERR_ARTIST_INVALID_ID,
88    ERR_ARTIST_NOT_FOUND,
89    ERR_AUTH_FAILED,
90    ERR_INVALID_CREDENTIALS,
91    ERR_ITEM_NOT_FOUND,
92    ERR_NO_ARTIST_FOR_TRACK,
93    ERR_TRACK_NOT_FOUND,
94    FAKE_ARTIST_PREFIX,
95    MAX_TOP_TRACKS,
96    MIX_CACHE_EXPIRATION,
97    MIX_ITEM_PREFIX,
98    RECOMMENDATIONS_HUB_PARAMS,
99)
100from music_assistant.providers.plex.helpers import (
101    AUDIOBOOK_FEATURES,
102    CONF_LIBRARY_TYPE,
103    LIBRARY_TYPE_AUDIOBOOKS,
104    LIBRARY_TYPE_MUSIC,
105    LIBRARY_TYPE_PODCASTS,
106    LIBRARY_TYPE_TO_MEDIA_TYPES,
107    PODCAST_FEATURES,
108    SUPPORTED_FEATURES,
109    extract_library_name,
110    get_explicit,
111    get_favorite_from_rating,
112    get_musicbrainz_id,
113    get_thumbnail_images,
114    parse_plex_lyrics_payload,
115)
116
117# Public surface of the provider package. With mypy's no_implicit_reexport,
118# names imported into this module (e.g. CONF_LIBRARY_ID from .constants) are
119# only re-exported when listed here.
120__all__ = [
121    "CONF_LIBRARY_ID",
122    "PlexProvider",
123    "setup",
124]
125
126if TYPE_CHECKING:
127    from collections.abc import AsyncGenerator, Callable, Coroutine
128
129    from music_assistant_models.provider import ProviderManifest
130    from plexapi.library import LibraryMediaTag as PlexCollection
131    from plexapi.library import MusicSection as PlexMusicSection
132    from plexapi.media import AudioStream as PlexAudioStream
133    from plexapi.media import Media as PlexMedia
134    from plexapi.media import MediaPart as PlexMediaPart
135
136    from music_assistant.mass import MusicAssistant
137    from music_assistant.models import ProviderInstanceType
138
139_LOGGER = logging.getLogger(__name__)
140
141UNKNOWN_NAME = "[Unknown]"
142PODCAST_PREFIX = "podcast:"
143PODCAST_EPISODE_PREFIX = "podcast_episode:"
144AUDIOBOOK_PREFIX = "audiobook:"
145CHAPTER_PREFIX = "Chapter"
146EPISODE_PREFIX = "Episode"
147
148
149async def setup(
150    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
151) -> ProviderInstanceType:
152    """Initialize provider(instance) with given configuration."""
153    # the token lives in setup_data for new installs, or (pre-flow) in the legacy config values
154    if not (config.setup_data.get(CONF_AUTH_TOKEN) or config.get_value(CONF_AUTH_TOKEN)):
155        raise LoginFailed(ERR_INVALID_CREDENTIALS)
156
157    return PlexProvider(mass, manifest, config, SUPPORTED_FEATURES)
158
159
160Param = ParamSpec("Param")
161RetType = TypeVar("RetType")
162PlexObjectT = TypeVar("PlexObjectT", bound=PlexObject)
163MediaItemT = TypeVar("MediaItemT", bound=MediaItem)
164
165
166class PlexProvider(RecommendationPayloadMixin, MusicProvider):
167    """Provider for a plex music library."""
168
169    # keep the pre-refactor 3h refresh interval for the hubs payload
170    recommendation_payload_ttl = 3600 * 3
171
172    _plex_server: PlexServer = None
173    _plex_library: PlexMusicSection = None
174    _myplex_account: MyPlexAccount = None
175    _baseurl: str
176
177    @property
178    def instance_name_postfix(self) -> str | None:
179        """Return a postfix with the library name and type."""
180        library_name = extract_library_name(str(self.get_setup_value(CONF_LIBRARY_ID) or ""))
181        library_type = self._get_library_type()
182        if library_type in (LIBRARY_TYPE_AUDIOBOOKS, LIBRARY_TYPE_PODCASTS):
183            type_label = library_type.title()
184            # Avoid duplication when the library name already indicates its type
185            if library_name.lower() == type_label.lower():
186                return library_name
187            return f"{library_name} - {type_label}"
188        if library_name:
189            return library_name
190        return None
191
192    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
193        """
194        Return Config entries to configure this provider.
195
196        Server connection, authentication and library selection are handled by the setup flow
197        (see setup_flow.py); only the genuine options are configurable here.
198        """
199        entries: list[ConfigEntry] = []
200
201        # Collection import options (advanced settings)
202        entries.append(
203            ConfigEntry(
204                key=CONF_IMPORT_COLLECTIONS,
205                type=ConfigEntryType.BOOLEAN,
206                default_value=False,
207                advanced=True,
208            )
209        )
210        entries.append(
211            ConfigEntry(
212                key=CONF_COLLECTION_PREFIX,
213                type=ConfigEntryType.STRING,
214                default_value="Collection: ",
215                depends_on=CONF_IMPORT_COLLECTIONS,
216                advanced=True,
217            )
218        )
219
220        # rating/favorite sync configuration
221        entries.append(
222            ConfigEntry(
223                key=CONF_PLEX_LIKE_RATING,
224                type=ConfigEntryType.FLOAT,
225                default_value=10.0,
226                range=(0, 10),
227                category="sync_options",
228            )
229        )
230        entries.append(
231            ConfigEntry(
232                key=CONF_PLEX_FAVORITE_THRESHOLD,
233                type=ConfigEntryType.FLOAT,
234                default_value=10.0,
235                range=(0, 10),
236                category="sync_options",
237            )
238        )
239        entries.append(
240            ConfigEntry(
241                key=CONF_PLEX_UNLIKE_RATING,
242                type=ConfigEntryType.FLOAT,
243                default_value=0.0,
244                range=(0, 10),
245                category="sync_options",
246            )
247        )
248
249        # Recommendation settings (advanced)
250        entries.append(
251            ConfigEntry(
252                key=CONF_HUB_ITEMS_LIMIT,
253                type=ConfigEntryType.INTEGER,
254                default_value=10,
255                advanced=True,
256                range=(1, 100),
257            )
258        )
259        entries.append(
260            ConfigEntry(
261                key=CONF_EXTENDED_RECOMMENDATIONS,
262                type=ConfigEntryType.BOOLEAN,
263                default_value=True,
264                advanced=True,
265            )
266        )
267
268        # return all config entries
269        return tuple(entries)
270
271    async def handle_async_init(self) -> None:
272        """Set up the music provider by connecting to the server."""
273        # silence loggers
274        logging.getLogger("plexapi").setLevel(self.logger.level + 10)
275
276        library_name = extract_library_name(str(self.get_setup_value(CONF_LIBRARY_ID)))
277
278        def connect() -> PlexServer:
279            try:
280                session = requests.Session()
281                session.verify = (
282                    bool(self.get_setup_value(CONF_LOCAL_SERVER_VERIFY_CERT))
283                    if self.get_setup_value(CONF_LOCAL_SERVER_SSL)
284                    else False
285                )
286                # Add Music Assistant client identification headers
287                session.headers.update(
288                    {
289                        "X-Plex-Client-Identifier": self.instance_id,
290                        "X-Plex-Product": "Music Assistant",
291                        "X-Plex-Platform": "Music Assistant",
292                        "X-Plex-Version": self.mass.version,
293                    }
294                )
295                local_server_protocol = (
296                    "https" if self.get_setup_value(CONF_LOCAL_SERVER_SSL) else "http"
297                )
298                token = self.get_setup_value(CONF_AUTH_TOKEN)
299                plex_url = (
300                    f"{local_server_protocol}://{self.get_setup_value(CONF_LOCAL_SERVER_IP)}"
301                    f":{self.get_setup_value(CONF_LOCAL_SERVER_PORT)}"
302                )
303                # silence urllib3 InsecureRequestWarning from Plex connections
304                # using wildcard certificates that don't validate against LAN IPs
305                with warnings.catch_warnings():
306                    warnings.filterwarnings(
307                        "ignore",
308                        category=urllib3.exceptions.InsecureRequestWarning,
309                    )
310                    if token == AUTH_TOKEN_UNAUTH:
311                        # Doing local connection, not via plex.tv.
312                        plex_server = PlexServer(plex_url, session=session)
313                    else:
314                        plex_server = PlexServer(
315                            plex_url,
316                            token,
317                            session=session,
318                        )
319                # I don't think PlexAPI intends for this to be accessible, but we need it.
320                self._baseurl = plex_server._baseurl
321
322            except plexapi.exceptions.BadRequest as err:
323                if "Invalid token" in str(err):
324                    # the stored token is invalid; surface an auth failure so the user is
325                    # sent through the reconfigure (reauth) flow, which overwrites the token
326                    raise LoginFailed(ERR_AUTH_FAILED)
327                raise LoginFailed from err
328            return plex_server
329
330        self._myplex_account = await self.get_myplex_account_and_refresh_token(
331            str(self.get_setup_value(CONF_AUTH_TOKEN))
332        )
333        try:
334            self._plex_server = await self._run_async(connect)
335            self._plex_library = await self._run_async(
336                self._plex_server.library.section, library_name
337            )
338        except requests.exceptions.ConnectionError as err:
339            raise SetupFailedError from err
340        # the library type is collected by the setup flow (setup_data), so a change now
341        # arrives via a full reload rather than update_config; clean up any mappings left
342        # behind by a previous type on load (idempotent - a no-op once nothing is stale)
343        await self._cleanup_stale_library_mappings()
344
345    @property
346    def is_streaming_provider(self) -> bool:
347        """
348        Return True if the provider is a streaming provider.
349
350        This literally means that the catalog is not the same as the library contents.
351        For local based providers (files, plex), the catalog is the same as the library content.
352        It also means that data is if this provider is NOT a streaming provider,
353        data cross instances is unique, the catalog and library differs per instance.
354
355        Setting this to True will only query one instance of the provider for search and lookups.
356        Setting this to False will query all instances of this provider for search and lookups.
357        """
358        return False
359
360    @property
361    def supported_features(self) -> set[ProviderFeature]:
362        """Return the features supported by this Provider."""
363        library_type = self._get_library_type()
364        if library_type == LIBRARY_TYPE_AUDIOBOOKS:
365            return AUDIOBOOK_FEATURES.copy()
366        if library_type == LIBRARY_TYPE_PODCASTS:
367            return PODCAST_FEATURES.copy()
368        return self._supported_features.copy()
369
370    async def resolve_image(self, path: str) -> str | bytes:
371        """Return the full image URL including the auth token."""
372        return str(self._plex_server.url(path, True))
373
374    @use_cache(3600)  # Cache for 1 hour
375    async def search(
376        self,
377        search_query: str,
378        media_types: list[MediaType],
379        limit: int = 20,
380    ) -> SearchResults:
381        """
382        Perform search on the plex library.
383
384        :param search_query: Search query.
385        :param media_types: A list of media_types to include.
386        :param limit: Number of items to return in the search (per type).
387        """
388        artists = None
389        albums = None
390        tracks = None
391        playlists = None
392
393        async with TaskGroup() as tg:
394            if MediaType.ARTIST in media_types:
395                artists = tg.create_task(
396                    self._search_and_parse(
397                        self._search_artist(search_query, limit), self._parse_artist
398                    )
399                )
400
401            if MediaType.ALBUM in media_types:
402                albums = tg.create_task(
403                    self._search_and_parse(
404                        self._search_album(search_query, limit), self._parse_album
405                    )
406                )
407
408            if MediaType.TRACK in media_types:
409                tracks = tg.create_task(
410                    self._search_and_parse(
411                        self._search_track(search_query, limit), self._parse_track
412                    )
413                )
414
415            if MediaType.PLAYLIST in media_types:
416                playlists = tg.create_task(
417                    self._search_and_parse(
418                        self._search_playlist(search_query, limit),
419                        self._parse_playlist,
420                    )
421                )
422
423        search_results = SearchResults()
424
425        if artists:
426            search_results.artists = artists.result()
427
428        if albums:
429            search_results.albums = albums.result()
430
431        if tracks:
432            search_results.tracks = tracks.result()
433
434        if playlists:
435            search_results.playlists = playlists.result()
436
437        return search_results
438
439    async def get_library_artists(self) -> AsyncGenerator[Artist]:
440        """Retrieve all library artists from Plex Music."""
441        artists_obj = await self._run_async(self._plex_library.all)
442        for artist in artists_obj:
443            yield await self._parse_artist(artist)
444
445    async def get_library_albums(self) -> AsyncGenerator[Album]:
446        """Retrieve all library albums from Plex Music."""
447        albums_obj = await self._run_async(self._plex_library.albums)
448        for album in albums_obj:
449            yield await self._parse_album(album)
450
451    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
452        """Retrieve all library playlists from the provider."""
453        playlists_obj = await self._run_async(self._plex_library.playlists)
454        for playlist in playlists_obj:
455            yield await self._parse_playlist(playlist)
456
457        # Import collections as playlists if enabled
458        if self.config.get_value(CONF_IMPORT_COLLECTIONS):
459            collections_obj = await self._run_async(self._plex_library.collections)
460            for collection in collections_obj:
461                yield await self._parse_collection(collection)
462
463    async def get_library_tracks(self) -> AsyncGenerator[Track]:
464        """Retrieve library tracks from Plex Music."""
465        page_size = 500
466        offset = 0
467        while True:
468            # maxresults caps a single page; without it container_size is only the HTTP
469            # batch size and plexapi keeps fetching until the end of the library, so every
470            # iteration would return all remaining tracks (an O(n^2) re-scan of the library).
471            batch = cast(
472                "list[PlexTrack]",
473                await self._run_async(
474                    self._plex_library.searchTracks,
475                    title=None,
476                    maxresults=page_size,
477                    container_size=page_size,
478                    container_start=offset,
479                ),
480            )
481            if not batch:
482                break
483            for plex_track in batch:
484                yield await self._parse_track(plex_track)
485            offset += page_size
486
487    async def get_library_audiobooks(self) -> AsyncGenerator[Audiobook]:
488        """Retrieve all library audiobooks from the configured Plex audiobook section."""
489        if self._get_library_type() != LIBRARY_TYPE_AUDIOBOOKS:
490            return
491        try:
492            albums_obj = await self._run_async(self._plex_library.albums)
493        except Exception:
494            self.logger.exception("Failed to list albums from audiobook library")
495            return
496        self.logger.debug(
497            "Found %d albums in audiobook library '%s'",
498            len(albums_obj),
499            self._plex_library.title,
500        )
501        for album in albums_obj:
502            try:
503                yield await self._parse_audiobook(album, include_chapters=False)
504            except Exception:
505                self.logger.warning(
506                    "Failed to parse audiobook album '%s' (key=%s); skipping",
507                    getattr(album, "title", "[unknown]"),
508                    getattr(album, "key", "[no key]"),
509                    exc_info=True,
510                )
511
512    @use_cache(3600 * 3)  # Cache for 3 hours
513    async def get_audiobook(self, prov_audiobook_id: str) -> Audiobook:
514        """Get full audiobook details (including chapters) by id."""
515        if self._get_library_type() != LIBRARY_TYPE_AUDIOBOOKS:
516            msg = "Audiobook library not configured"
517            raise MediaNotFoundError(msg)
518        album_key = prov_audiobook_id.removeprefix(AUDIOBOOK_PREFIX)
519        try:
520            plex_album = cast(
521                "PlexAlbum",
522                await self._run_async(self._plex_library.fetchItem, album_key, PlexAlbum),
523            )
524        except plexapi.exceptions.NotFound:
525            msg = f"Audiobook {prov_audiobook_id} not found"
526            raise MediaNotFoundError(msg)
527        return await self._parse_audiobook(plex_album, include_chapters=True)
528
529    async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
530        """Retrieve all library podcasts from the configured Plex podcast section."""
531        if self._get_library_type() != LIBRARY_TYPE_PODCASTS:
532            return
533        try:
534            albums_obj = await self._run_async(self._plex_library.albums)
535        except Exception:
536            self.logger.exception("Failed to list albums from podcast library")
537            return
538        for album in albums_obj:
539            try:
540                yield await self._parse_podcast(album, include_episodes=False)
541            except Exception:
542                self.logger.warning(
543                    "Failed to parse podcast album '%s' (key=%s); skipping",
544                    getattr(album, "title", "[unknown]"),
545                    getattr(album, "key", "[no key]"),
546                    exc_info=True,
547                )
548
549    @use_cache(3600 * 3)  # Cache for 3 hours
550    async def get_podcast(self, prov_podcast_id: str) -> Podcast:
551        """Get full podcast details (including episodes) by id."""
552        if self._get_library_type() != LIBRARY_TYPE_PODCASTS:
553            msg = "Podcast library not configured"
554            raise MediaNotFoundError(msg)
555        album_key = prov_podcast_id.removeprefix(PODCAST_PREFIX)
556        try:
557            plex_album = cast(
558                "PlexAlbum",
559                await self._run_async(self._plex_library.fetchItem, album_key, PlexAlbum),
560            )
561        except plexapi.exceptions.NotFound:
562            msg = f"Podcast {prov_podcast_id} not found"
563            raise MediaNotFoundError(msg)
564        return await self._parse_podcast(plex_album, include_episodes=True)
565
566    async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
567        """Get all PodcastEpisodes for given podcast id."""
568        if self._get_library_type() != LIBRARY_TYPE_PODCASTS:
569            return
570        album_key = prov_podcast_id.removeprefix(PODCAST_PREFIX)
571        try:
572            plex_album = cast(
573                "PlexAlbum",
574                await self._run_async(self._plex_library.fetchItem, album_key, PlexAlbum),
575            )
576        except plexapi.exceptions.NotFound:
577            msg = f"Podcast {prov_podcast_id} not found"
578            raise MediaNotFoundError(msg)
579        for episode in await self._build_podcast_episodes(plex_album):
580            yield episode
581
582    @use_cache(3600 * 3)  # Cache for 3 hours
583    async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
584        """Get full podcast episode details by id."""
585        if self._get_library_type() != LIBRARY_TYPE_PODCASTS:
586            msg = "Podcast library not configured"
587            raise MediaNotFoundError(msg)
588        track_key = prov_episode_id.removeprefix(PODCAST_EPISODE_PREFIX)
589        try:
590            plex_track = cast(
591                "PlexTrack",
592                await self._run_async(self._plex_library.fetchItem, track_key, PlexTrack),
593            )
594        except plexapi.exceptions.NotFound:
595            msg = f"Podcast episode {prov_episode_id} not found"
596            raise MediaNotFoundError(msg)
597        return await self._parse_podcast_episode(plex_track)
598
599    async def get_resume_position(
600        self, item_id: str, media_type: MediaType
601    ) -> tuple[bool, int, datetime | None]:
602        """
603        Get progress (resume point) details for the given audiobook or podcast.
604
605        :param item_id: provider item id (e.g. "audiobook:<plex_key>").
606        :param media_type: the media type (AUDIOBOOK or PODCAST).
607        :return: (fully_played, position_ms, timestamp)
608        """
609        library_type = self._get_library_type()
610        if media_type == MediaType.AUDIOBOOK and library_type == LIBRARY_TYPE_AUDIOBOOKS:
611            album_key = item_id.removeprefix(AUDIOBOOK_PREFIX)
612        elif media_type == MediaType.PODCAST and library_type == LIBRARY_TYPE_PODCASTS:
613            album_key = item_id.removeprefix(PODCAST_PREFIX)
614        elif media_type == MediaType.PODCAST_EPISODE and library_type == LIBRARY_TYPE_PODCASTS:
615            episode_key = item_id.removeprefix(PODCAST_EPISODE_PREFIX)
616            try:
617                plex_track = cast(
618                    "PlexTrack",
619                    await self._run_async(self._plex_library.fetchItem, episode_key, PlexTrack),
620                )
621            except plexapi.exceptions.NotFound:
622                msg = f"Podcast episode {episode_key} not found"
623                raise MediaNotFoundError(msg)
624            # For podcast episodes, progress lives on each individual track.
625            # lastViewedAt may be on the parent album; fall back to the track.
626            fully_played = bool(getattr(plex_track, "viewCount", 0) > 0)
627            timestamp = getattr(plex_track, "lastViewedAt", None)
628            if timestamp is not None and timestamp.tzinfo is None:
629                timestamp = timestamp.replace(tzinfo=UTC)
630            resume_position_ms = getattr(plex_track, "viewOffset", 0) or 0
631            return fully_played, resume_position_ms, timestamp
632        else:
633            raise NotImplementedError
634        try:
635            plex_album = cast(
636                "PlexAlbum",
637                await self._run_async(self._plex_library.fetchItem, album_key, PlexAlbum),
638            )
639        except plexapi.exceptions.NotFound:
640            msg = f"Item {item_id} not found"
641            raise MediaNotFoundError(msg)
642
643        try:
644            await self._run_async(plex_album.reload)
645        except plexapi.exceptions.PlexApiException, requests.exceptions.RequestException:
646            self.logger.warning(
647                "Failed to reload metadata for position check (%s), using cached metadata",
648                item_id,
649            )
650
651        fully_played = bool(getattr(plex_album, "viewCount", 0) > 0)
652        timestamp = getattr(plex_album, "lastViewedAt", None)
653        if timestamp is not None and timestamp.tzinfo is None:
654            timestamp = timestamp.replace(tzinfo=UTC)
655
656        resume_position_ms = await self._calc_resume_position_ms(plex_album, fully_played)
657        return fully_played, resume_position_ms, timestamp
658
659    async def on_played(
660        self,
661        media_type: MediaType,
662        prov_item_id: str,
663        fully_played: bool,
664        position: int,
665        media_item: MediaItemType,
666        is_playing: bool = False,
667    ) -> None:
668        """
669        Handle callback when an audiobook or podcast has been played.
670
671        Syncs progress back to the Plex server using the timeline/progress API.
672
673        :param media_type: The media type (AUDIOBOOK or PODCAST).
674        :param prov_item_id: The provider-specific item id.
675        :param fully_played: True when the item has been played to the end.
676        :param position: Last known position in seconds.
677        :param media_item: The full media item details.
678        :param is_playing: True when currently playing.
679        """
680        library_type = self._get_library_type()
681        if media_type == MediaType.AUDIOBOOK and library_type == LIBRARY_TYPE_AUDIOBOOKS:
682            album_key = prov_item_id.removeprefix(AUDIOBOOK_PREFIX)
683        elif media_type == MediaType.PODCAST and library_type == LIBRARY_TYPE_PODCASTS:
684            album_key = prov_item_id.removeprefix(PODCAST_PREFIX)
685        elif media_type == MediaType.PODCAST_EPISODE and library_type == LIBRARY_TYPE_PODCASTS:
686            episode_key = prov_item_id.removeprefix(PODCAST_EPISODE_PREFIX)
687            plex_track = cast(
688                "PlexTrack",
689                await self._run_async(self._plex_library.fetchItem, episode_key, PlexTrack),
690            )
691            album_key = str(plex_track.parentKey)
692        else:
693            return
694
695        try:
696            plex_album = cast(
697                "PlexAlbum",
698                await self._run_async(self._plex_library.fetchItem, album_key, PlexAlbum),
699            )
700        except plexapi.exceptions.NotFound:
701            self.logger.warning(
702                "Failed to fetch %s %s for played sync", media_type.value, prov_item_id
703            )
704            return
705        except Exception:
706            self.logger.warning(
707                "Failed to fetch %s %s for played sync",
708                media_type.value,
709                prov_item_id,
710                exc_info=True,
711            )
712            return
713
714        if fully_played:
715            await self._run_async(plex_album.markPlayed)
716            self.logger.debug("Marked %s %s as played in Plex", media_type.value, prov_item_id)
717            return
718
719        if position <= 0:
720            await self._run_async(plex_album.markUnplayed)
721            self.logger.debug("Marked %s %s as unplayed in Plex", media_type.value, prov_item_id)
722            return
723
724        try:
725            target_track, target_offset_ms = await self._find_track_for_position(
726                plex_album, position
727            )
728            if target_track is None:
729                return
730
731            state = "playing" if is_playing else "paused"
732            # updateTimeline expects time in milliseconds (Plex native unit)
733            await self._run_async(
734                target_track.updateTimeline,
735                target_offset_ms,
736                state=state,
737                duration=getattr(target_track, "duration", None),
738            )
739            self.logger.debug(
740                "Synced %s %s progress to Plex: track %s at %dms (%s)",
741                media_type.value,
742                prov_item_id,
743                target_track.title,
744                target_offset_ms,
745                state,
746            )
747        except Exception:
748            self.logger.warning(
749                "Failed to sync %s %s progress to Plex",
750                media_type.value,
751                prov_item_id,
752                exc_info=True,
753            )
754
755    @use_cache(3600 * 3)  # Cache for 3 hours
756    async def get_album(self, prov_album_id: str) -> Album:
757        """Get full album details by id."""
758        plex_album = await self._get_data(prov_album_id, PlexAlbum)
759        return await self._parse_album(plex_album)
760
761    @use_cache(3600 * 3)  # Cache for 3 hours
762    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
763        """Get album tracks for given album id."""
764        plex_album: PlexAlbum = await self._get_data(prov_album_id, PlexAlbum)
765        tracks = []
766        for plex_track in await self._run_async(plex_album.tracks):
767            track = await self._parse_track(
768                plex_track,
769            )
770            tracks.append(track)
771        return tracks
772
773    @use_cache(3600 * 3)  # Cache for 3 hours
774    async def get_artist(self, prov_artist_id: str) -> Artist:
775        """Get full artist details by id."""
776        if prov_artist_id.startswith(FAKE_ARTIST_PREFIX):
777            # This artist does not exist in plex, so we can just load it from DB.
778
779            if db_artist := await self.mass.music.artists.get_library_item_by_prov_id(
780                prov_artist_id, self.instance_id
781            ):
782                return db_artist
783            raise MediaNotFoundError(ERR_ARTIST_NOT_FOUND.format(item_id=prov_artist_id))
784
785        plex_artist = await self._get_data(prov_artist_id, PlexArtist)
786        return await self._parse_artist(plex_artist)
787
788    @use_cache(3600 * 3)  # Cache for 3 hours
789    async def get_track(self, prov_track_id: str) -> Track:
790        """Get full track details by id."""
791        plex_track = await self._get_data(prov_track_id, PlexTrack)
792        track = await self._parse_track(plex_track)
793        await self._add_track_lyrics(plex_track, track)
794        return track
795
796    @use_cache(3600 * 3)  # Cache for 3 hours
797    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
798        """Get full playlist details by id."""
799        # Check if this is a collection (collections have the format "collection:<key>")
800        if prov_playlist_id.startswith(COLLECTION_ID_PREFIX):
801            collection_key = prov_playlist_id.removeprefix(COLLECTION_ID_PREFIX)
802            plex_collection: PlexObject = await self._get_data(collection_key)
803            return await self._parse_collection(plex_collection)
804
805        # "Mixes For You" items use a MIX_ITEM_PREFIX (see _build_mix_playlist).
806        if prov_playlist_id.startswith(MIX_ITEM_PREFIX):
807            mix_key = prov_playlist_id.removeprefix(MIX_ITEM_PREFIX)
808            fields = await self._find_mix_by_key(mix_key)
809            if fields is None:
810                msg = f"Mix {prov_playlist_id} not found"
811                raise MediaNotFoundError(msg)
812            _, title, thumb = fields
813            # Cache title/artwork on interaction so replay from recently-played
814            # still renders after Plex rotates the mix out of the hub.
815            if mix_key:
816                await self.mass.cache.set(
817                    key=mix_key,
818                    data={"title": title, "thumb": thumb},
819                    provider=self.instance_id,
820                    expiration=MIX_CACHE_EXPIRATION,
821                )
822            return self._build_mix_playlist(mix_key, title, thumb)
823
824        plex_playlist = await self._get_data(prov_playlist_id, PlexPlaylist)
825        return await self._parse_playlist(plex_playlist)
826
827    @use_cache(3600 * 3)  # Cache for 3 hours
828    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
829        """Get playlist tracks."""
830        result: list[Track] = []
831        if page > 0:
832            # paging not supported, we always return the whole list at once
833            return []
834
835        # Check if this is a collection (collections have the format "collection:<key>")
836        if prov_playlist_id.startswith(COLLECTION_ID_PREFIX):
837            collection_key = prov_playlist_id.removeprefix(COLLECTION_ID_PREFIX)
838            plex_collection: PlexObject = await self._get_data(collection_key)
839            if not (collection_items := await self._run_async(plex_collection.items)):
840                return result
841            # Collections can contain tracks, albums, or artists - we only want tracks
842            for item in collection_items:
843                if item.type == "track":
844                    if track := await self._parse_track(item):
845                        track.position = len(result) + 1
846                        result.append(track)
847                elif item.type == "album":
848                    # If the collection contains albums, get all tracks from each album
849                    album_tracks = await self.get_album_tracks(item.key)
850                    for album_track in album_tracks:
851                        album_track.position = len(result) + 1
852                        result.append(album_track)
853            return result
854
855        # "Mixes For You" items use a MIX_ITEM_PREFIX. Strip it to recover
856        # the Plex section-query key, append the track type filter to expand
857        # albums into tracks, then shuffle — Plexamp randomizes mix playback
858        # client-side.
859        if prov_playlist_id.startswith(MIX_ITEM_PREFIX):
860            mix_key = prov_playlist_id.removeprefix(MIX_ITEM_PREFIX)
861            tracks_key = f"{mix_key}&type={plexapi.utils.searchType('track')}"
862            plex_tracks = await self._run_async(self._plex_library.fetchItems, tracks_key)
863            random.shuffle(plex_tracks)
864            for index, plex_track in enumerate(plex_tracks, 1):
865                if track := await self._parse_track(plex_track):
866                    track.position = index
867                    result.append(track)
868            return result
869
870        plex_playlist: PlexPlaylist = await self._get_data(prov_playlist_id, PlexPlaylist)
871        if not (playlist_items := await self._run_async(plex_playlist.items)):
872            return result
873        for index, plex_track in enumerate(playlist_items, 1):
874            if track := await self._parse_track(plex_track):
875                track.position = index
876                result.append(track)
877        return result
878
879    @use_cache(3600 * 3)  # Cache for 3 hours
880    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
881        """Get a list of albums for the given artist."""
882        if not prov_artist_id.startswith(FAKE_ARTIST_PREFIX):
883            plex_artist = await self._get_data(prov_artist_id, PlexArtist)
884            try:
885                plex_albums = cast("list[PlexAlbum]", await self._run_async(plex_artist.albums))
886            except plexapi.exceptions.NotFound:
887                # PlexArtist.albums() relies on Plex's advanced filters API.
888                # Some Plex servers return no filtering metadata, making plexapi
889                # raise 'Unknown libtype "artist"'. Fall back to the artist's
890                # /children endpoint, which does not depend on the filters API.
891                albums_key = f"{plex_artist.key}/children"
892                plex_albums = cast(
893                    "list[PlexAlbum]",
894                    await self._run_async(plex_artist.fetchItems, albums_key, PlexAlbum),
895                )
896            if plex_albums:
897                albums = []
898                for album_obj in plex_albums:
899                    albums.append(await self._parse_album(album_obj))
900                return albums
901        return []
902
903    @use_cache(3600 * 3)  # Cache for 3 hours
904    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
905        """Get top tracks for the given artist."""
906        if prov_artist_id.startswith(FAKE_ARTIST_PREFIX):
907            return []
908        plex_artist = await self._get_data(prov_artist_id, PlexArtist)
909        try:
910            plex_tracks = cast("list[PlexTrack]", await self._run_async(plex_artist.popularTracks))
911        except plexapi.exceptions.NotFound:
912            # PlexArtist.popularTracks() relies on Plex's advanced filters API.
913            # Some Plex servers return no filtering metadata, making plexapi
914            # raise 'Unknown libtype "artist"'. Fall back to ranking the artist's
915            # own tracks, which does not depend on the filters API.
916            plex_tracks = await self._rank_artist_tracks(plex_artist)
917        return [await self._parse_track(plex_track) for plex_track in plex_tracks[:MAX_TOP_TRACKS]]
918
919    @use_cache(3600 * 3)  # Cache for 3 hours
920    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
921        """Get similar tracks using Plex's sonicallySimilar feature."""
922        try:
923            plex_track = await self._get_data(prov_track_id, PlexTrack)
924            # Get sonically similar tracks
925            similar_tracks = await self._run_async(plex_track.sonicallySimilar, limit=limit)
926            tracks = []
927            for similar_track in similar_tracks:
928                if track := await self._parse_track(similar_track):
929                    tracks.append(track)
930            self.logger.debug(
931                "Retrieved %d similar tracks for track %s", len(tracks), prov_track_id
932            )
933            return tracks
934        except Exception as err:
935            self.logger.warning("Error getting similar tracks for %s: %s", prov_track_id, err)
936        return []
937
938    async def get_recommendations(self) -> list[RecommendationFolder]:
939        """Get this provider's available recommendation rows, without items."""
940        return await self._recommendation_rows_from_payload()
941
942    async def get_recommendation_items(
943        self, item_id: str
944    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
945        """
946        Get the items for a single recommendation row.
947
948        :param item_id: The item_id of the row, as returned by get_recommendations.
949        """
950        return await self._recommendation_items_from_payload(item_id)
951
952    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
953        """Get streamdetails for a track/audiobook/podcast episode."""
954        if media_type == MediaType.AUDIOBOOK:
955            return await self._get_audiobook_stream_details(item_id)
956        if media_type == MediaType.PODCAST_EPISODE:
957            return await self._get_podcast_episode_stream_details(item_id)
958
959        plex_track = await self._get_data(item_id, PlexTrack)
960        if not plex_track.media:
961            raise MediaNotFoundError(ERR_TRACK_NOT_FOUND.format(item_id=item_id))
962
963        media: PlexMedia = plex_track.media[0]
964
965        content_type = (
966            ContentType.try_parse(media.container) if media.container else ContentType.UNKNOWN
967        )
968        media_part: PlexMediaPart = media.parts[0]
969        audio_streams = media_part.audioStreams()
970        audio_stream: PlexAudioStream | None = audio_streams[0] if audio_streams else None
971
972        stream_details = StreamDetails(
973            item_id=plex_track.key,
974            provider=self.instance_id,
975            audio_format=AudioFormat(
976                content_type=content_type,
977                channels=media.audioChannels,
978            ),
979            stream_type=StreamType.HTTP,
980            # plex reports duration in milliseconds, streamdetails expect seconds
981            duration=int(plex_track.duration / 1000) if plex_track.duration else None,
982            data=plex_track,
983            can_seek=True,
984            allow_seek=True,
985        )
986
987        download_url = self._plex_server.url(f"{media_part.key}?download=1", True)
988
989        if content_type != ContentType.M4A:
990            stream_details.path = download_url
991            if audio_stream and audio_stream.samplingRate:
992                stream_details.audio_format.sample_rate = audio_stream.samplingRate
993            if audio_stream and audio_stream.bitDepth:
994                stream_details.audio_format.bit_depth = audio_stream.bitDepth
995
996        else:
997            media_info = await async_parse_tags(download_url)
998            stream_details.path = download_url
999            stream_details.audio_format.channels = media_info.channels
1000            stream_details.audio_format.content_type = ContentType.try_parse(media_info.format)
1001            stream_details.audio_format.sample_rate = media_info.sample_rate
1002            stream_details.audio_format.bit_depth = media_info.bits_per_sample
1003
1004        return stream_details
1005
1006    async def get_myplex_account_and_refresh_token(self, auth_token: str) -> MyPlexAccount:
1007        """Get a MyPlexAccount object and refresh the token if needed."""
1008        if auth_token == AUTH_TOKEN_UNAUTH:
1009            return self._myplex_account
1010
1011        def _refresh_plex_token() -> MyPlexAccount:
1012            if self._myplex_account is None:
1013                myplex_account = MyPlexAccount(token=auth_token)
1014                self._myplex_account = myplex_account
1015            self._myplex_account.ping()
1016            return self._myplex_account
1017
1018        return await asyncio.to_thread(_refresh_plex_token)
1019
1020    async def set_favorite(self, prov_item_id: str, media_type: MediaType, favorite: bool) -> None:
1021        """Set favorite status by setting rating in Plex."""
1022        if favorite:
1023            # Set like rating
1024            rating = cast("float", self.config.get_value(CONF_PLEX_LIKE_RATING))
1025        else:
1026            # Set unlike rating
1027            rating = cast("float", self.config.get_value(CONF_PLEX_UNLIKE_RATING))
1028
1029        if media_type == MediaType.TRACK:
1030            plex_item: PlexTrack | PlexAlbum = await self._get_data(prov_item_id, PlexTrack)
1031        elif media_type == MediaType.ALBUM:
1032            plex_item = await self._get_data(prov_item_id, PlexAlbum)
1033        else:
1034            return
1035        await self._run_async(plex_item.rate, rating)
1036        self.logger.debug(
1037            "Set Plex rating to %s for %s with ID %s (ratingKey: %s)",
1038            rating,
1039            media_type.value,
1040            prov_item_id,
1041            plex_item.ratingKey,
1042        )
1043
1044    def _get_library_type(self) -> str:
1045        """Return the configured library type, defaulting to music."""
1046        return str(self.get_setup_value(CONF_LIBRARY_TYPE) or LIBRARY_TYPE_MUSIC)
1047
1048    async def _cleanup_stale_library_mappings(self) -> None:
1049        """Remove provider mappings that do not belong to the current library type."""
1050        if not self.mass.music.database:
1051            return
1052        valid_types = set(LIBRARY_TYPE_TO_MEDIA_TYPES.get(self._get_library_type(), ()))
1053        all_types = {t for types in LIBRARY_TYPE_TO_MEDIA_TYPES.values() for t in types}
1054        for media_type in all_types - valid_types:
1055            controller = self.mass.music.get_controller(media_type)
1056            query = (
1057                f"SELECT item_id FROM {DB_TABLE_PROVIDER_MAPPINGS} "
1058                f"WHERE media_type = '{media_type.value}' "
1059                f"AND provider_instance = '{self.instance_id}'"
1060            )
1061            rows = await self.mass.music.database.get_rows_from_query(query, limit=100000)
1062            if rows:
1063                self.logger.info(
1064                    "Cleaning up %d stale %s provider mapping(s)", len(rows), media_type.value
1065                )
1066            for db_row in rows:
1067                try:
1068                    await controller.remove_provider_mappings(db_row["item_id"], self.instance_id)
1069                except Exception as err:
1070                    self.logger.warning(
1071                        "Failed to remove stale %s provider mapping for %s: %s",
1072                        media_type.value,
1073                        db_row["item_id"],
1074                        err,
1075                    )
1076
1077    async def _rank_artist_tracks(self, plex_artist: PlexArtist) -> list[PlexTrack]:
1078        """
1079        Rank an artist's own tracks by popularity, keeping one version per title.
1080
1081        :param plex_artist: The Plex artist to rank the tracks of.
1082        """
1083        plex_tracks = cast("list[PlexTrack]", await self._run_async(plex_artist.tracks))
1084        best_per_title: dict[str, PlexTrack] = {}
1085        for plex_track in plex_tracks:
1086            if not plex_track.ratingCount:
1087                # ratingCount is the Last.fm scrobble count popularTracks() ranks on,
1088                # so a track without one has no rank. viewCount is local plays instead.
1089                continue
1090            title = (plex_track.title or "").casefold()
1091            best = best_per_title.get(title)
1092            if best is None or plex_track.ratingCount > best.ratingCount:
1093                best_per_title[title] = plex_track
1094        return sorted(best_per_title.values(), key=lambda track: track.ratingCount, reverse=True)
1095
1096    async def _run_async(
1097        self, call: Callable[Param, RetType], *args: Param.args, **kwargs: Param.kwargs
1098    ) -> RetType:
1099        await self.get_myplex_account_and_refresh_token(str(self.get_setup_value(CONF_AUTH_TOKEN)))
1100        return await asyncio.to_thread(call, *args, **kwargs)
1101
1102    async def _get_data(self, key: str, cls: type[PlexObjectT] | None = None) -> PlexObjectT:
1103        try:
1104            results = await self._run_async(self._plex_library.fetchItem, key, cls)
1105        except plexapi.exceptions.NotFound as err:
1106            raise MediaNotFoundError(ERR_ITEM_NOT_FOUND.format(item_id=key)) from err
1107        return cast("PlexObjectT", results)
1108
1109    def _get_item_mapping(self, media_type: MediaType, key: str, name: str) -> ItemMapping:
1110        """Get item mapping for a given media type, key, and name."""
1111        if not name:
1112            self.logger.info(
1113                "Received None or empty name for media item. Media type: %s, Key: %s",
1114                media_type,
1115                key,
1116            )
1117            name = UNKNOWN_NAME
1118
1119        mapped_name, mapped_version = parse_title_and_version(name)
1120
1121        if not mapped_name:
1122            self.logger.info(
1123                "Failed to map name for media item. Media type: %s, Key: %s, Original name: %s",
1124                media_type,
1125                key,
1126                name,
1127            )
1128            mapped_name = UNKNOWN_NAME
1129        if not mapped_version and media_type not in (MediaType.ALBUM, MediaType.TRACK):
1130            mapped_version = ""
1131
1132        return ItemMapping(
1133            media_type=media_type,
1134            item_id=key,
1135            provider=self.instance_id,
1136            name=mapped_name,
1137            version=mapped_version,
1138        )
1139
1140    async def _get_or_create_artist_by_name(self, artist_name: str) -> Artist | ItemMapping:
1141        if library_items := await self.mass.music.artists.get_library_items_by_query(
1142            search=artist_name, provider_filter=[self.instance_id]
1143        ):
1144            return ItemMapping.from_item(library_items[0])
1145
1146        artist_id = FAKE_ARTIST_PREFIX + artist_name
1147        return Artist(
1148            item_id=artist_id,
1149            name=artist_name or UNKNOWN_ARTIST,
1150            provider=self.instance_id,
1151            provider_mappings={
1152                ProviderMapping(
1153                    item_id=str(artist_id),
1154                    provider_domain=self.domain,
1155                    provider_instance=self.instance_id,
1156                )
1157            },
1158        )
1159
1160    async def _parse(self, plex_media: PlexObject) -> MediaItem | None:
1161        if plex_media.type == "artist":
1162            return await self._parse_artist(plex_media)
1163        if plex_media.type == "album":
1164            return await self._parse_album(plex_media)
1165        if plex_media.type == "track":
1166            return await self._parse_track(plex_media)
1167        if plex_media.type == "playlist":
1168            return await self._parse_playlist(plex_media)
1169        return None
1170
1171    async def _search_track(self, search_query: str, limit: int) -> list[PlexTrack]:
1172        return cast(
1173            "list[PlexTrack]",
1174            await self._run_async(self._plex_library.searchTracks, title=search_query, limit=limit),
1175        )
1176
1177    async def _search_album(self, search_query: str, limit: int) -> list[PlexAlbum]:
1178        return cast(
1179            "list[PlexAlbum]",
1180            await self._run_async(self._plex_library.searchAlbums, title=search_query, limit=limit),
1181        )
1182
1183    async def _search_artist(self, search_query: str, limit: int) -> list[PlexArtist]:
1184        return cast(
1185            "list[PlexArtist]",
1186            await self._run_async(
1187                self._plex_library.searchArtists, title=search_query, limit=limit
1188            ),
1189        )
1190
1191    async def _search_playlist(self, search_query: str, limit: int) -> list[PlexPlaylist]:
1192        return cast(
1193            "list[PlexPlaylist]",
1194            await self._run_async(self._plex_library.playlists, title=search_query, limit=limit),
1195        )
1196
1197    async def _search_and_parse(
1198        self,
1199        search_coro: Awaitable[list[PlexObjectT]],
1200        parse_coro: Callable[[PlexObjectT], Coroutine[Any, Any, MediaItemT]],
1201    ) -> list[MediaItemT]:
1202        task_results: list[Task[MediaItemT]] = []
1203        async with TaskGroup() as tg:
1204            for item in await search_coro:
1205                task_results.append(tg.create_task(parse_coro(item)))
1206
1207        results: list[MediaItemT] = []
1208        for task in task_results:
1209            results.append(task.result())
1210
1211        return results
1212
1213    async def _parse_album(self, plex_album: PlexAlbum) -> Album:
1214        """Parse a Plex Album response to an Album model object."""
1215        album_id = plex_album.key
1216        album = Album(
1217            item_id=album_id,
1218            provider=self.instance_id,
1219            name=plex_album.title or UNKNOWN_NAME,
1220            provider_mappings={
1221                ProviderMapping(
1222                    item_id=str(album_id),
1223                    provider_domain=self.domain,
1224                    provider_instance=self.instance_id,
1225                    url=plex_album.getWebURL(self._baseurl),
1226                )
1227            },
1228        )
1229        # Check if album rating meets the configured threshold for favorites
1230        favorite_threshold = cast("float", self.config.get_value(CONF_PLEX_FAVORITE_THRESHOLD))
1231        if (favorite := get_favorite_from_rating(plex_album, favorite_threshold)) is not None:
1232            album.favorite = favorite
1233
1234        if plex_album.year:
1235            album.year = plex_album.year
1236        if images := get_thumbnail_images(plex_album, self.instance_id):
1237            album.metadata.images = images
1238        if plex_album.summary:
1239            album.metadata.description = plex_album.summary
1240        if plex_album.genres:
1241            album.metadata.genres = {genre.tag for genre in plex_album.genres if genre.tag}
1242        if plex_album.moods:
1243            album.metadata.mood = next((mood.tag for mood in plex_album.moods if mood.tag), None)
1244        if plex_album.styles:
1245            album.metadata.style = next(
1246                (style.tag for style in plex_album.styles if style.tag), None
1247            )
1248        if plex_album.originallyAvailableAt:
1249            album.metadata.release_date = plex_album.originallyAvailableAt
1250        if (explicit := get_explicit(plex_album)) is not None:
1251            album.metadata.explicit = explicit
1252        if mbid := clean_mbid(
1253            get_musicbrainz_id(plex_album), f"album {plex_album.title}", self.logger
1254        ):
1255            album.mbid = mbid
1256
1257        album.artists.append(
1258            self._get_item_mapping(
1259                MediaType.ARTIST,
1260                plex_album.parentKey,
1261                plex_album.parentTitle or UNKNOWN_ARTIST,
1262            )
1263        )
1264        return album
1265
1266    async def _parse_artist(self, plex_artist: PlexArtist) -> Artist:
1267        """Parse a Plex Artist response to Artist model object."""
1268        artist_id = plex_artist.key
1269        if not artist_id:
1270            raise InvalidDataError(ERR_ARTIST_INVALID_ID)
1271        artist = Artist(
1272            item_id=artist_id,
1273            name=plex_artist.title or UNKNOWN_ARTIST,
1274            provider=self.instance_id,
1275            provider_mappings={
1276                ProviderMapping(
1277                    item_id=str(artist_id),
1278                    provider_domain=self.domain,
1279                    provider_instance=self.instance_id,
1280                    url=plex_artist.getWebURL(self._baseurl),
1281                )
1282            },
1283        )
1284        if plex_artist.summary:
1285            artist.metadata.description = plex_artist.summary
1286        if images := get_thumbnail_images(plex_artist, self.instance_id):
1287            artist.metadata.images = images
1288        if plex_artist.genres:
1289            artist.metadata.genres = {genre.tag for genre in plex_artist.genres if genre.tag}
1290        if plex_artist.moods:
1291            artist.metadata.mood = next((mood.tag for mood in plex_artist.moods if mood.tag), None)
1292        if plex_artist.styles:
1293            artist.metadata.style = next(
1294                (style.tag for style in plex_artist.styles if style.tag), None
1295            )
1296        if mbid := clean_mbid(
1297            get_musicbrainz_id(plex_artist), f"artist {plex_artist.title}", self.logger
1298        ):
1299            artist.mbid = mbid
1300        return artist
1301
1302    async def _parse_playlist(self, plex_playlist: PlexPlaylist) -> Playlist:
1303        """Parse a Plex Playlist response to a Playlist object."""
1304        playlist = Playlist(
1305            item_id=plex_playlist.key,
1306            provider=self.instance_id,
1307            name=plex_playlist.title or UNKNOWN_NAME,
1308            provider_mappings={
1309                ProviderMapping(
1310                    item_id=plex_playlist.key,
1311                    provider_domain=self.domain,
1312                    provider_instance=self.instance_id,
1313                    url=plex_playlist.getWebURL(self._baseurl),
1314                )
1315            },
1316        )
1317        if plex_playlist.summary:
1318            playlist.metadata.description = plex_playlist.summary
1319        if images := get_thumbnail_images(plex_playlist, self.instance_id):
1320            playlist.metadata.images = images
1321        playlist.is_editable = not plex_playlist.smart
1322        return playlist
1323
1324    async def _parse_collection(self, plex_collection: PlexCollection) -> Playlist:
1325        """Parse a Plex Collection response to a Playlist object."""
1326        # Get the configured collection prefix
1327        collection_prefix = str(self.config.get_value(CONF_COLLECTION_PREFIX) or "")
1328
1329        # Collections are imported as playlists with the configured prefix
1330        playlist = Playlist(
1331            item_id=f"{COLLECTION_ID_PREFIX}{plex_collection.key}",
1332            provider=self.instance_id,
1333            name=f"{collection_prefix}{plex_collection.title}",
1334            provider_mappings={
1335                ProviderMapping(
1336                    item_id=f"{COLLECTION_ID_PREFIX}{plex_collection.key}",
1337                    provider_domain=self.domain,
1338                    provider_instance=self.instance_id,
1339                )
1340            },
1341        )
1342        # Add collection poster/thumbnail if available
1343        if images := get_thumbnail_images(
1344            plex_collection, self.instance_id, ("thumb", "composite")
1345        ):
1346            playlist.metadata.images = images
1347        # Collections are not editable in Music Assistant
1348        playlist.is_editable = False
1349        return playlist
1350
1351    def _mix_playlist_fields(self, plex_mix: PlexPlaylist) -> tuple[str, str, str | None]:
1352        """
1353        Extract (smart-query key, title, centroid thumb) from a 'Mix For You' item.
1354
1355        :param plex_mix: A Plex Playlist parsed from the 'Mixes For You' hub.
1356        """
1357        # Read straight from the parsed XML element. These synthetic mix playlists
1358        # carry a centroid-derived ratingKey rather than their own, so touching any
1359        # attribute that triggers a reload (e.g. .thumb) re-fetches the wrong object
1360        # and corrupts it. The smart-query key, title, and centroid artist thumb are
1361        # all present on the partial element itself.
1362        data = plex_mix._data
1363        mix_key = data.get("key") or ""
1364        title = data.get("title") or "[Unknown Mix]"
1365        thumb = next(
1366            (child.get("thumb") for child in data if child.get("centroid") and child.get("thumb")),
1367            None,
1368        )
1369        return mix_key, title, thumb
1370
1371    def _build_mix_playlist(self, mix_key: str, title: str, thumb: str | None) -> Playlist:
1372        """
1373        Build a MA Playlist from a Plex 'Mix For You' hub item.
1374
1375        :param mix_key: The Plex smart-query key identifying the mix.
1376        :param title: The mix title.
1377        :param thumb: The centroid artist thumb path, if any.
1378        """
1379        item_id = f"{MIX_ITEM_PREFIX}{mix_key}"
1380        playlist = Playlist(
1381            item_id=item_id,
1382            provider=self.instance_id,
1383            name=title,
1384            provider_mappings={
1385                ProviderMapping(
1386                    item_id=item_id,
1387                    provider_domain=self.domain,
1388                    provider_instance=self.instance_id,
1389                )
1390            },
1391        )
1392        if thumb:
1393            playlist.metadata.images = UniqueList(
1394                [
1395                    MediaItemImage(
1396                        type=ImageType.THUMB,
1397                        path=thumb,
1398                        provider=self.instance_id,
1399                        remotely_accessible=False,
1400                    )
1401                ]
1402            )
1403        playlist.is_editable = False
1404        playlist.is_dynamic = True
1405        return playlist
1406
1407    async def _get_mix_playlists(self, count: int) -> list[PlexPlaylist]:
1408        """
1409        Fetch the 'Mixes For You' hub items as Plex Playlist objects.
1410
1411        :param count: Maximum number of items per hub.
1412        """
1413        key = f"/hubs/sections/{self._plex_library.key}?count={count}&{RECOMMENDATIONS_HUB_PARAMS}"
1414        hubs = await self._run_async(self._plex_library.fetchItems, key)
1415        for hub in hubs:
1416            if "music.mixes" in (hub.hubIdentifier or ""):
1417                return list(hub._partialItems)
1418        return []
1419
1420    async def _find_mix_by_key(self, mix_key: str) -> tuple[str, str, str | None] | None:
1421        """Find a 'Mix For You' by its smart-query key, falling back to cache."""
1422        limit_value = self.config.get_value(CONF_HUB_ITEMS_LIMIT)
1423        limit = int(limit_value) if isinstance(limit_value, (int, float, str)) else 10
1424        for plex_mix in await self._get_mix_playlists(limit):
1425            fields = self._mix_playlist_fields(plex_mix)
1426            if fields[0] == mix_key:
1427                return fields
1428        # Plex rotates mixes out of the hub, but the smart-query key remains a
1429        # valid section query, so replay from recently-played still works — we
1430        # only need the cache to restore the title and artwork.
1431        cached = await self.mass.cache.get(key=mix_key, provider=self.instance_id)
1432        if isinstance(cached, dict):
1433            return mix_key, cached.get("title") or "[Unknown Mix]", cached.get("thumb")
1434        return None
1435
1436    async def _parse_track(self, plex_track: PlexTrack) -> Track:
1437        """Parse a Plex Track response to a Track model object."""
1438        content = plex_track.media[0].container if plex_track.media else None
1439        track = Track(
1440            item_id=plex_track.key,
1441            provider=self.instance_id,
1442            name=plex_track.title or UNKNOWN_NAME,
1443            provider_mappings={
1444                ProviderMapping(
1445                    item_id=plex_track.key,
1446                    provider_domain=self.domain,
1447                    provider_instance=self.instance_id,
1448                    # For Plex (local library provider), assume tracks are available by default
1449                    # even if media attribute is not populated in the initial response.
1450                    # This prevents tracks from being skipped during library sync.
1451                    available=True,
1452                    audio_format=AudioFormat(
1453                        content_type=(
1454                            ContentType.try_parse(content) if content else ContentType.UNKNOWN
1455                        ),
1456                    ),
1457                    url=plex_track.getWebURL(self._baseurl),
1458                )
1459            },
1460            disc_number=plex_track.parentIndex or 0,
1461            track_number=plex_track.trackNumber or 0,
1462        )
1463        # Check if track rating meets the configured threshold for favorites
1464        favorite_threshold = cast("float", self.config.get_value(CONF_PLEX_FAVORITE_THRESHOLD))
1465        if (favorite := get_favorite_from_rating(plex_track, favorite_threshold)) is not None:
1466            track.favorite = favorite
1467
1468        if plex_track.originalTitle and plex_track.originalTitle != plex_track.grandparentTitle:
1469            # The artist of the track if different from the album's artist.
1470            # For this kind of artist, we just know the name, so we create a fake artist,
1471            # if it does not already exist.
1472            track.artists.append(
1473                await self._get_or_create_artist_by_name(plex_track.originalTitle or UNKNOWN_ARTIST)
1474            )
1475        elif plex_track.grandparentKey:
1476            track.artists.append(
1477                self._get_item_mapping(
1478                    MediaType.ARTIST,
1479                    plex_track.grandparentKey,
1480                    plex_track.grandparentTitle or UNKNOWN_ARTIST,
1481                )
1482            )
1483        else:
1484            raise InvalidDataError(ERR_NO_ARTIST_FOR_TRACK)
1485
1486        if images := get_thumbnail_images(plex_track, self.instance_id):
1487            track.metadata.images = images
1488        if plex_track.genres:
1489            track.metadata.genres = {genre.tag for genre in plex_track.genres if genre.tag}
1490        if plex_track.moods:
1491            track.metadata.mood = next((mood.tag for mood in plex_track.moods if mood.tag), None)
1492        if (explicit := get_explicit(plex_track)) is not None:
1493            track.metadata.explicit = explicit
1494        if mbid := clean_mbid(
1495            get_musicbrainz_id(plex_track), f"track {plex_track.title}", self.logger
1496        ):
1497            track.mbid = mbid
1498        if plex_track.parentKey:
1499            track.album = self._get_item_mapping(
1500                MediaType.ALBUM, plex_track.parentKey, plex_track.parentTitle
1501            )
1502        if plex_track.duration:
1503            track.duration = int(plex_track.duration / 1000)
1504
1505        return track
1506
1507    async def _add_track_lyrics(self, plex_track: PlexTrack, track: Track) -> None:
1508        """
1509        Fetch the track's lyric stream from Plex and attach it to the metadata.
1510
1511        :param plex_track: The fully loaded Plex track to read lyric streams from.
1512        :param track: The Music Assistant track to populate with lyrics.
1513        """
1514
1515        def _fetch() -> str | None:
1516            stream = next((stream for stream in plex_track.lyricStreams() if stream.key), None)
1517            if stream is None:
1518                return None
1519            url = plex_track._server.url(stream.key, includeToken=True)
1520            response: requests.Response = plex_track._server._session.get(
1521                url, headers={"Accept": "application/json"}, timeout=30
1522            )
1523            response.raise_for_status()
1524            # plexapi's untyped session makes the response Any for mypy; force str
1525            return str(response.text)
1526
1527        try:
1528            content = await self._run_async(_fetch)
1529        except (requests.RequestException, plexapi.exceptions.PlexApiException) as err:
1530            self.logger.debug("Failed to fetch lyrics for %s: %s", plex_track.key, err)
1531            return
1532        if not content or (parsed := parse_plex_lyrics_payload(content)) is None:
1533            return
1534        lyrics, synced = parsed
1535        if synced:
1536            track.metadata.lrc_lyrics = lyrics
1537        else:
1538            track.metadata.lyrics = lyrics
1539
1540    async def _parse_audiobook(
1541        self, plex_album: PlexAlbum, *, include_chapters: bool = False
1542    ) -> Audiobook:
1543        """Parse a Plex Album from the audiobook library into an Audiobook model."""
1544        audiobook_id = f"{AUDIOBOOK_PREFIX}{plex_album.key}"
1545        audiobook = Audiobook(
1546            item_id=audiobook_id,
1547            provider=self.instance_id,
1548            name=plex_album.title or UNKNOWN_NAME,
1549            provider_mappings={
1550                ProviderMapping(
1551                    item_id=audiobook_id,
1552                    provider_domain=self.domain,
1553                    provider_instance=self.instance_id,
1554                    url=plex_album.getWebURL(self._baseurl),
1555                )
1556            },
1557        )
1558        # Author: parentTitle is the album artist; grandparentTitle is the album
1559        # artist parent (for multi-level nesting in Plex). Some setups vary.
1560        if author_name := plex_album.parentTitle or plex_album.grandparentTitle:
1561            audiobook.authors = UniqueList([author_name])
1562        if plex_album.summary:
1563            audiobook.metadata.description = plex_album.summary
1564        if plex_album.year:
1565            audiobook.metadata.release_date = datetime(plex_album.year, 1, 1, tzinfo=UTC)
1566        if images := get_thumbnail_images(plex_album, self.instance_id):
1567            audiobook.metadata.images = images
1568        # minified path: use album-level duration if Plex exposes it
1569        if album_duration := getattr(plex_album, "duration", None):
1570            audiobook.duration = int(album_duration / 1000)
1571
1572        if include_chapters:
1573            chapters = await self._build_audiobook_chapters(plex_album)
1574            audiobook.metadata.chapters = chapters
1575            if chapters and chapters[-1].end is not None:
1576                audiobook.duration = int(chapters[-1].end)
1577
1578        return audiobook
1579
1580    async def _build_audiobook_chapters(self, plex_album: PlexAlbum) -> list[MediaItemChapter]:
1581        """Build chapter list from Plex tracks, skipping tracks without playable media."""
1582        plex_tracks = cast("list[PlexTrack]", await self._run_async(plex_album.tracks))
1583        plex_tracks.sort(key=lambda t: (t.parentIndex or 0, t.trackNumber or 0))
1584        chapters: list[MediaItemChapter] = []
1585        cumulative = 0.0
1586        chapter_num = 0
1587        for plex_track in plex_tracks:
1588            if not plex_track.media or not plex_track.media[0].parts:
1589                continue
1590            chapter_num += 1
1591            # plex_track.duration is in milliseconds (Plex native unit)
1592            duration_s = (plex_track.duration or 0) / 1000.0
1593            chapters.append(
1594                MediaItemChapter(
1595                    position=chapter_num,
1596                    name=plex_track.title or f"{CHAPTER_PREFIX} {chapter_num}",
1597                    start=cumulative,
1598                    end=cumulative + duration_s,
1599                )
1600            )
1601            cumulative += duration_s
1602        return chapters
1603
1604    async def _parse_podcast(
1605        self, plex_album: PlexAlbum, *, include_episodes: bool = False
1606    ) -> Podcast:
1607        """Parse a Plex Album from the podcast library into a Podcast model."""
1608        podcast_id = f"{PODCAST_PREFIX}{plex_album.key}"
1609        podcast = Podcast(
1610            item_id=podcast_id,
1611            provider=self.instance_id,
1612            name=plex_album.title or UNKNOWN_NAME,
1613            provider_mappings={
1614                ProviderMapping(
1615                    item_id=podcast_id,
1616                    provider_domain=self.domain,
1617                    provider_instance=self.instance_id,
1618                    url=plex_album.getWebURL(self._baseurl),
1619                )
1620            },
1621        )
1622        publisher = plex_album.studio or plex_album.parentTitle or plex_album.grandparentTitle
1623        if publisher:
1624            podcast.publisher = publisher
1625        if plex_album.summary:
1626            podcast.metadata.description = plex_album.summary
1627        if plex_album.year:
1628            podcast.metadata.release_date = datetime(plex_album.year, 1, 1, tzinfo=UTC)
1629        if images := get_thumbnail_images(plex_album, self.instance_id):
1630            podcast.metadata.images = images
1631        if include_episodes:
1632            podcast.total_episodes = await self._count_podcast_episodes(plex_album)
1633        return podcast
1634
1635    async def _count_podcast_episodes(self, plex_album: PlexAlbum) -> int:
1636        """Count playable tracks without building full PodcastEpisode objects."""
1637        plex_tracks = cast("list[PlexTrack]", await self._run_async(plex_album.tracks))
1638        return sum(1 for t in plex_tracks if t.media and t.media[0].parts)
1639
1640    async def _build_podcast_episodes(self, plex_album: PlexAlbum) -> list[PodcastEpisode]:
1641        """Build episode list from Plex tracks, skipping tracks without playable media."""
1642        plex_tracks = cast("list[PlexTrack]", await self._run_async(plex_album.tracks))
1643        plex_tracks.sort(key=lambda t: (t.parentIndex or 0, t.trackNumber or 0))
1644        episodes: list[PodcastEpisode] = []
1645        episode_num = 0
1646        for plex_track in plex_tracks:
1647            if not plex_track.media or not plex_track.media[0].parts:
1648                continue
1649            episode_num += 1
1650            duration_s = (plex_track.duration or 0) / 1000.0
1651            episode = PodcastEpisode(
1652                item_id=f"{PODCAST_EPISODE_PREFIX}{plex_track.key}",
1653                provider=self.instance_id,
1654                name=plex_track.title or f"{EPISODE_PREFIX} {episode_num}",
1655                position=episode_num,
1656                duration=int(duration_s),
1657                podcast=ItemMapping(
1658                    media_type=MediaType.PODCAST,
1659                    item_id=f"{PODCAST_PREFIX}{plex_album.key}",
1660                    provider=self.instance_id,
1661                    name=plex_album.title or UNKNOWN_NAME,
1662                ),
1663                provider_mappings={
1664                    ProviderMapping(
1665                        item_id=f"{PODCAST_EPISODE_PREFIX}{plex_track.key}",
1666                        provider_domain=self.domain,
1667                        provider_instance=self.instance_id,
1668                        url=plex_track.getWebURL(self._baseurl),
1669                        audio_format=AudioFormat(
1670                            content_type=(
1671                                ContentType.try_parse(plex_track.media[0].container)
1672                                if plex_track.media[0].container
1673                                else ContentType.UNKNOWN
1674                            )
1675                        ),
1676                    )
1677                },
1678            )
1679            if images := get_thumbnail_images(plex_track, self.instance_id):
1680                episode.metadata.images = images
1681            episodes.append(episode)
1682        return episodes
1683
1684    async def _parse_podcast_episode(self, plex_track: PlexTrack) -> PodcastEpisode:
1685        """Parse a Plex Track from the podcast library into a PodcastEpisode model."""
1686        duration_s = (plex_track.duration or 0) / 1000.0
1687        content_type = ContentType.UNKNOWN
1688        if plex_track.media and plex_track.media[0].container:
1689            content_type = ContentType.try_parse(plex_track.media[0].container)
1690        episode = PodcastEpisode(
1691            item_id=f"{PODCAST_EPISODE_PREFIX}{plex_track.key}",
1692            provider=self.instance_id,
1693            name=plex_track.title or UNKNOWN_NAME,
1694            position=plex_track.trackNumber or 0,
1695            duration=int(duration_s),
1696            podcast=ItemMapping(
1697                media_type=MediaType.PODCAST,
1698                item_id=f"{PODCAST_PREFIX}{plex_track.parentKey}",
1699                provider=self.instance_id,
1700                name=plex_track.parentTitle or UNKNOWN_NAME,
1701            ),
1702            provider_mappings={
1703                ProviderMapping(
1704                    item_id=f"{PODCAST_EPISODE_PREFIX}{plex_track.key}",
1705                    provider_domain=self.domain,
1706                    provider_instance=self.instance_id,
1707                    url=plex_track.getWebURL(self._baseurl),
1708                    audio_format=AudioFormat(content_type=content_type),
1709                )
1710            },
1711        )
1712        if images := get_thumbnail_images(plex_track, self.instance_id):
1713            episode.metadata.images = images
1714        return episode
1715
1716    async def _calc_resume_position_ms(self, plex_album: PlexAlbum, fully_played: bool) -> int:
1717        """Calculate resume position from per-track viewOffset values."""
1718        plex_tracks = cast("list[PlexTrack]", await self._run_async(plex_album.tracks))
1719        plex_tracks.sort(key=lambda t: (t.parentIndex or 0, t.trackNumber or 0))
1720
1721        # Per-track durations and viewOffset are in milliseconds (Plex native).
1722        resume_position_ms = 0
1723        cumulative_ms = 0
1724        for plex_track in plex_tracks:
1725            track_offset = getattr(plex_track, "viewOffset", 0) or 0
1726            if track_offset > 0:
1727                # Use the last non-zero offset — for sequential listening this
1728                # is the final playback position; it also handles non-linear
1729                # skipping better than first-match.
1730                resume_position_ms = cumulative_ms + track_offset
1731            cumulative_ms += getattr(plex_track, "duration", 0) or 0
1732
1733        if resume_position_ms == 0 and fully_played:
1734            album_duration = getattr(plex_album, "duration", 0) or 0
1735            resume_position_ms = int(album_duration)
1736
1737        return resume_position_ms
1738
1739    async def _find_track_for_position(
1740        self, plex_album: PlexAlbum, position: int
1741    ) -> tuple[PlexTrack | None, int]:
1742        """Find the track and offset (ms) corresponding to the given position (s)."""
1743        plex_tracks = cast("list[PlexTrack]", await self._run_async(plex_album.tracks))
1744        plex_tracks.sort(key=lambda t: (t.parentIndex or 0, t.trackNumber or 0))
1745
1746        position_ms = position * 1000
1747        cumulative_ms = 0
1748        for plex_track in plex_tracks:
1749            track_duration = getattr(plex_track, "duration", 0) or 0
1750            if cumulative_ms + track_duration > position_ms:
1751                return plex_track, position_ms - cumulative_ms
1752            cumulative_ms += track_duration
1753
1754        if plex_tracks:
1755            # Position is past all tracks — clamp to end of the last track.
1756            last_track = plex_tracks[-1]
1757            last_duration = getattr(last_track, "duration", 0) or 0
1758            return last_track, last_duration
1759
1760        return None, 0
1761
1762    async def _get_audiobook_stream_details(self, item_id: str) -> StreamDetails:
1763        """Build multi-part StreamDetails for an audiobook (one part per Plex track)."""
1764        if self._get_library_type() != LIBRARY_TYPE_AUDIOBOOKS:
1765            msg = "Library not configured for audiobooks"
1766            raise MediaNotFoundError(msg)
1767        album_key = item_id.removeprefix(AUDIOBOOK_PREFIX)
1768        try:
1769            plex_album = cast(
1770                "PlexAlbum",
1771                await self._run_async(self._plex_library.fetchItem, album_key, PlexAlbum),
1772            )
1773        except plexapi.exceptions.NotFound:
1774            msg = f"Audiobook {item_id} not found"
1775            raise MediaNotFoundError(msg)
1776
1777        plex_tracks = cast("list[PlexTrack]", await self._run_async(plex_album.tracks))
1778        plex_tracks.sort(key=lambda t: (t.parentIndex or 0, t.trackNumber or 0))
1779
1780        parts, total_duration, first_container = self._build_stream_parts(plex_tracks, item_id)
1781        if not parts:
1782            self.logger.error(
1783                "Audiobook %s (%s) has no playable parts (%d tracks checked)",
1784                item_id,
1785                plex_album.title,
1786                len(plex_tracks),
1787            )
1788            msg = f"Audiobook {item_id} has no playable parts"
1789            raise MediaNotFoundError(msg)
1790
1791        self.logger.debug(
1792            "Built StreamDetails for audiobook %s with %d parts, total_duration=%.1fs",
1793            item_id,
1794            len(parts),
1795            total_duration,
1796        )
1797
1798        content_type = (
1799            ContentType.try_parse(first_container) if first_container else ContentType.UNKNOWN
1800        )
1801
1802        return StreamDetails(
1803            provider=self.instance_id,
1804            item_id=item_id,
1805            media_type=MediaType.AUDIOBOOK,
1806            audio_format=AudioFormat(content_type=content_type),
1807            stream_type=StreamType.HTTP,
1808            duration=int(total_duration),
1809            path=parts[0].path if len(parts) == 1 else parts,
1810            can_seek=True,
1811            allow_seek=True,
1812        )
1813
1814    async def _fetch_recommendation_payload(self) -> list[RecommendationFolder]:
1815        """Fetch the full recommendations payload (folders with items) from the Plex hubs."""
1816        # Let fetch errors propagate: the payload mixin serves the last cached payload
1817        # on a failed refresh, and returning [] here would be cached as a valid empty
1818        # result for the full TTL.
1819        # Get the configured limit for items per hub
1820        limit_value = self.config.get_value(CONF_HUB_ITEMS_LIMIT)
1821        limit = int(limit_value) if isinstance(limit_value, (int, float, str)) else 10
1822
1823        # Build the hubs key manually because plexapi's hubs() method
1824        # doesn't accept a count parameter to limit items per hub.
1825        extended = self.config.get_value(CONF_EXTENDED_RECOMMENDATIONS)
1826        hub_params = RECOMMENDATIONS_HUB_PARAMS if extended else "includeStations=1"
1827        key = f"/hubs/sections/{self._plex_library.key}?count={limit}&{hub_params}"
1828        hubs = await self._run_async(self._plex_library.fetchItems, key)
1829
1830        if not hubs:
1831            self.logger.debug("No hubs available from Plex")
1832            return []
1833
1834        self.logger.debug(
1835            "Fetching %d hubs (limit: %d items per hub)",
1836            len(hubs),
1837            limit,
1838        )
1839
1840        folders = []
1841        for hub in hubs:
1842            # Create a recommendation folder for each hub
1843            folder = RecommendationFolder(
1844                name=hub.title,
1845                item_id=f"{self.instance_id}_{hub.hubIdentifier}",
1846                provider=self.instance_id,
1847                icon="mdi-music",
1848            )
1849
1850            # Mixes For You are synthetic smart playlists; build them from
1851            # their partial hub items (see _mix_playlist_fields).
1852            if "music.mixes" in (hub.hubIdentifier or ""):
1853                folder.items.extend(
1854                    self._build_mix_playlist(*self._mix_playlist_fields(plex_mix))
1855                    for plex_mix in hub._partialItems
1856                )
1857                if folder.items:
1858                    folders.append(folder)
1859                continue
1860
1861            # Parse each item based on its type (limit to configured max)
1862            # Use _partialItems to respect the count limit from the hubs() call
1863            # rather than hub.items() which fetches ALL items if more is True
1864            # _partialItems is a cached property that's already loaded, so no need for async
1865            hub_items = hub._partialItems
1866            self.logger.debug(
1867                "Processing hub '%s' (%s) with %d partial items",
1868                hub.title,
1869                hub.hubIdentifier,
1870                len(hub_items),
1871            )
1872            for item in hub_items:
1873                try:
1874                    # Skip items without type attribute
1875                    if not hasattr(item, "type"):
1876                        self.logger.debug(
1877                            "Skipping item in hub '%s': no type attribute",
1878                            hub.title,
1879                        )
1880                        continue
1881
1882                    if parsed_item := await self._parse(item):
1883                        folder.items.append(parsed_item)  # type: ignore[arg-type]
1884                    else:
1885                        self.logger.debug(
1886                            "Skipping unsupported item type '%s' in hub '%s'",
1887                            item.type,
1888                            hub.title,
1889                        )
1890                except Exception as err:
1891                    self.logger.debug(
1892                        "Failed to parse item (type: %s) in hub '%s': %s",
1893                        getattr(item, "type", "unknown"),
1894                        hub.title,
1895                        str(err),
1896                    )
1897                    continue
1898
1899            # Only add folder if it has items
1900            if folder.items:
1901                folders.append(folder)
1902                self.logger.debug(
1903                    "Added hub '%s' (%s) with %d items",
1904                    hub.title,
1905                    hub.hubIdentifier,
1906                    len(folder.items),
1907                )
1908            else:
1909                self.logger.debug(
1910                    "Skipping hub '%s' (%s): no items after parsing",
1911                    hub.title,
1912                    hub.hubIdentifier,
1913                )
1914
1915        self.logger.debug("Retrieved %d recommendation folders from Plex", len(folders))
1916        return folders
1917
1918    def _build_stream_parts(
1919        self, plex_tracks: list[PlexTrack], item_id: str
1920    ) -> tuple[list[MultiPartPath], float, str | None]:
1921        """Convert Plex tracks to MultiPartPath entries for streaming."""
1922        parts: list[MultiPartPath] = []
1923        total_duration = 0.0
1924        first_container: str | None = None
1925        for plex_track in plex_tracks:
1926            media = self._track_media_or_log(plex_track, item_id)
1927            if media is None:
1928                continue
1929            if first_container is None and media.container:
1930                first_container = media.container
1931            media_part: PlexMediaPart = media.parts[0]
1932            url = self._plex_server.url(f"{media_part.key}?download=1", True)
1933            duration_s = (plex_track.duration or 0) / 1000.0
1934            parts.append(MultiPartPath(path=url, duration=duration_s))
1935            total_duration += duration_s
1936            self.logger.debug(
1937                "Added audiobook part: track '%s' (%s) duration=%.1fs url=%s",
1938                plex_track.title,
1939                plex_track.key,
1940                duration_s,
1941                url,
1942            )
1943        return parts, total_duration, first_container
1944
1945    def _track_media_or_log(self, plex_track: PlexTrack, item_id: str) -> PlexMedia | None:
1946        """Return the first PlexMedia for a track, or log and return None if unavailable."""
1947        if not plex_track.media:
1948            self.logger.debug(
1949                "Skipping track '%s' (key=%s) in audiobook %s: no media",
1950                plex_track.title,
1951                plex_track.key,
1952                item_id,
1953            )
1954            return None
1955        media: PlexMedia = plex_track.media[0]
1956        if not media.parts:
1957            self.logger.debug(
1958                "Skipping track '%s' (key=%s) in audiobook %s: media has no parts",
1959                plex_track.title,
1960                plex_track.key,
1961                item_id,
1962            )
1963            return None
1964        return media
1965
1966    async def _get_podcast_episode_stream_details(self, item_id: str) -> StreamDetails:
1967        """Build streamdetails for a single podcast episode from a Plex track."""
1968        if self._get_library_type() != LIBRARY_TYPE_PODCASTS:
1969            msg = "Library not configured for podcasts"
1970            raise MediaNotFoundError(msg)
1971        track_key = item_id.removeprefix(PODCAST_EPISODE_PREFIX)
1972        try:
1973            plex_track = cast(
1974                "PlexTrack",
1975                await self._run_async(self._plex_library.fetchItem, track_key, PlexTrack),
1976            )
1977        except plexapi.exceptions.NotFound:
1978            msg = f"Podcast episode {item_id} not found"
1979            raise MediaNotFoundError(msg)
1980
1981        if not plex_track.media:
1982            msg = f"Podcast episode {item_id} has no media"
1983            raise MediaNotFoundError(msg)
1984
1985        media: PlexMedia = plex_track.media[0]
1986        if not media.parts:
1987            msg = f"Podcast episode {item_id} has no playable media parts"
1988            raise MediaNotFoundError(msg)
1989        content_type = (
1990            ContentType.try_parse(media.container) if media.container else ContentType.UNKNOWN
1991        )
1992        media_part: PlexMediaPart = media.parts[0]
1993        download_url = self._plex_server.url(f"{media_part.key}?download=1", True)
1994
1995        return StreamDetails(
1996            provider=self.instance_id,
1997            item_id=item_id,
1998            media_type=MediaType.PODCAST_EPISODE,
1999            audio_format=AudioFormat(content_type=content_type),
2000            stream_type=StreamType.HTTP,
2001            duration=plex_track.duration,
2002            path=download_url,
2003            can_seek=True,
2004            allow_seek=True,
2005        )
2006