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 MAXYEAR, MINYEAR, 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            parsed = await self._parse_or_skip(self._parse_artist, artist)
444            if parsed is not None:
445                yield parsed
446
447    async def get_library_albums(self) -> AsyncGenerator[Album]:
448        """Retrieve all library albums from Plex Music."""
449        albums_obj = await self._run_async(self._plex_library.albums)
450        for album in albums_obj:
451            parsed = await self._parse_or_skip(self._parse_album, album)
452            if parsed is not None:
453                yield parsed
454
455    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
456        """Retrieve all library playlists from the provider."""
457        playlists_obj = await self._run_async(self._plex_library.playlists)
458        for playlist in playlists_obj:
459            parsed = await self._parse_or_skip(self._parse_playlist, playlist)
460            if parsed is not None:
461                yield parsed
462
463        # Import collections as playlists if enabled
464        if self.config.get_value(CONF_IMPORT_COLLECTIONS):
465            collections_obj = await self._run_async(self._plex_library.collections)
466            for collection in collections_obj:
467                parsed = await self._parse_or_skip(self._parse_collection, collection)
468                if parsed is not None:
469                    yield parsed
470
471    async def get_library_tracks(self) -> AsyncGenerator[Track]:
472        """Retrieve library tracks from Plex Music."""
473        page_size = 500
474        offset = 0
475        while True:
476            # maxresults caps a single page; without it container_size is only the HTTP
477            # batch size and plexapi keeps fetching until the end of the library, so every
478            # iteration would return all remaining tracks (an O(n^2) re-scan of the library).
479            batch = cast(
480                "list[PlexTrack]",
481                await self._run_async(
482                    self._plex_library.searchTracks,
483                    title=None,
484                    maxresults=page_size,
485                    container_size=page_size,
486                    container_start=offset,
487                ),
488            )
489            if not batch:
490                break
491            for plex_track in batch:
492                parsed = await self._parse_or_skip(self._parse_track, plex_track)
493                if parsed is not None:
494                    yield parsed
495            offset += page_size
496
497    async def get_library_audiobooks(self) -> AsyncGenerator[Audiobook]:
498        """Retrieve all library audiobooks from the configured Plex audiobook section."""
499        if self._get_library_type() != LIBRARY_TYPE_AUDIOBOOKS:
500            return
501        albums_obj = await self._run_async(self._plex_library.albums)
502        self.logger.debug(
503            "Found %d albums in audiobook library '%s'",
504            len(albums_obj),
505            self._plex_library.title,
506        )
507        for album in albums_obj:
508            parsed = await self._parse_or_skip(self._parse_audiobook, album)
509            if parsed is not None:
510                yield parsed
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        albums_obj = await self._run_async(self._plex_library.albums)
534        for album in albums_obj:
535            parsed = await self._parse_or_skip(self._parse_podcast, album)
536            if parsed is not None:
537                yield parsed
538
539    @use_cache(3600 * 3)  # Cache for 3 hours
540    async def get_podcast(self, prov_podcast_id: str) -> Podcast:
541        """Get full podcast details (including episodes) by id."""
542        if self._get_library_type() != LIBRARY_TYPE_PODCASTS:
543            msg = "Podcast library not configured"
544            raise MediaNotFoundError(msg)
545        album_key = prov_podcast_id.removeprefix(PODCAST_PREFIX)
546        try:
547            plex_album = cast(
548                "PlexAlbum",
549                await self._run_async(self._plex_library.fetchItem, album_key, PlexAlbum),
550            )
551        except plexapi.exceptions.NotFound:
552            msg = f"Podcast {prov_podcast_id} not found"
553            raise MediaNotFoundError(msg)
554        return await self._parse_podcast(plex_album, include_episodes=True)
555
556    async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
557        """Get all PodcastEpisodes for given podcast id."""
558        if self._get_library_type() != LIBRARY_TYPE_PODCASTS:
559            return
560        album_key = prov_podcast_id.removeprefix(PODCAST_PREFIX)
561        try:
562            plex_album = cast(
563                "PlexAlbum",
564                await self._run_async(self._plex_library.fetchItem, album_key, PlexAlbum),
565            )
566        except plexapi.exceptions.NotFound:
567            msg = f"Podcast {prov_podcast_id} not found"
568            raise MediaNotFoundError(msg)
569        for episode in await self._build_podcast_episodes(plex_album):
570            yield episode
571
572    @use_cache(3600 * 3)  # Cache for 3 hours
573    async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
574        """Get full podcast episode details by id."""
575        if self._get_library_type() != LIBRARY_TYPE_PODCASTS:
576            msg = "Podcast library not configured"
577            raise MediaNotFoundError(msg)
578        track_key = prov_episode_id.removeprefix(PODCAST_EPISODE_PREFIX)
579        try:
580            plex_track = cast(
581                "PlexTrack",
582                await self._run_async(self._plex_library.fetchItem, track_key, PlexTrack),
583            )
584        except plexapi.exceptions.NotFound:
585            msg = f"Podcast episode {prov_episode_id} not found"
586            raise MediaNotFoundError(msg)
587        return await self._parse_podcast_episode(plex_track)
588
589    async def get_resume_position(
590        self, item_id: str, media_type: MediaType
591    ) -> tuple[bool, int, datetime | None]:
592        """
593        Get progress (resume point) details for the given audiobook or podcast.
594
595        :param item_id: provider item id (e.g. "audiobook:<plex_key>").
596        :param media_type: the media type (AUDIOBOOK or PODCAST).
597        :return: (fully_played, position_ms, timestamp)
598        """
599        library_type = self._get_library_type()
600        if media_type == MediaType.AUDIOBOOK and library_type == LIBRARY_TYPE_AUDIOBOOKS:
601            album_key = item_id.removeprefix(AUDIOBOOK_PREFIX)
602        elif media_type == MediaType.PODCAST and library_type == LIBRARY_TYPE_PODCASTS:
603            album_key = item_id.removeprefix(PODCAST_PREFIX)
604        elif media_type == MediaType.PODCAST_EPISODE and library_type == LIBRARY_TYPE_PODCASTS:
605            episode_key = item_id.removeprefix(PODCAST_EPISODE_PREFIX)
606            try:
607                plex_track = cast(
608                    "PlexTrack",
609                    await self._run_async(self._plex_library.fetchItem, episode_key, PlexTrack),
610                )
611            except plexapi.exceptions.NotFound:
612                msg = f"Podcast episode {episode_key} not found"
613                raise MediaNotFoundError(msg)
614            # For podcast episodes, progress lives on each individual track.
615            # lastViewedAt may be on the parent album; fall back to the track.
616            fully_played = bool(getattr(plex_track, "viewCount", 0) > 0)
617            timestamp = getattr(plex_track, "lastViewedAt", None)
618            if timestamp is not None and timestamp.tzinfo is None:
619                timestamp = timestamp.replace(tzinfo=UTC)
620            resume_position_ms = getattr(plex_track, "viewOffset", 0) or 0
621            return fully_played, resume_position_ms, timestamp
622        else:
623            raise NotImplementedError
624        try:
625            plex_album = cast(
626                "PlexAlbum",
627                await self._run_async(self._plex_library.fetchItem, album_key, PlexAlbum),
628            )
629        except plexapi.exceptions.NotFound:
630            msg = f"Item {item_id} not found"
631            raise MediaNotFoundError(msg)
632
633        try:
634            await self._run_async(plex_album.reload)
635        except plexapi.exceptions.PlexApiException, requests.exceptions.RequestException:
636            self.logger.warning(
637                "Failed to reload metadata for position check (%s), using cached metadata",
638                item_id,
639            )
640
641        fully_played = bool(getattr(plex_album, "viewCount", 0) > 0)
642        timestamp = getattr(plex_album, "lastViewedAt", None)
643        if timestamp is not None and timestamp.tzinfo is None:
644            timestamp = timestamp.replace(tzinfo=UTC)
645
646        resume_position_ms = await self._calc_resume_position_ms(plex_album, fully_played)
647        return fully_played, resume_position_ms, timestamp
648
649    async def on_played(
650        self,
651        media_type: MediaType,
652        prov_item_id: str,
653        fully_played: bool,
654        position: int,
655        media_item: MediaItemType,
656        is_playing: bool = False,
657    ) -> None:
658        """
659        Handle callback when an audiobook or podcast has been played.
660
661        Syncs progress back to the Plex server using the timeline/progress API.
662
663        :param media_type: The media type (AUDIOBOOK or PODCAST).
664        :param prov_item_id: The provider-specific item id.
665        :param fully_played: True when the item has been played to the end.
666        :param position: Last known position in seconds.
667        :param media_item: The full media item details.
668        :param is_playing: True when currently playing.
669        """
670        library_type = self._get_library_type()
671        if media_type == MediaType.AUDIOBOOK and library_type == LIBRARY_TYPE_AUDIOBOOKS:
672            album_key = prov_item_id.removeprefix(AUDIOBOOK_PREFIX)
673        elif media_type == MediaType.PODCAST and library_type == LIBRARY_TYPE_PODCASTS:
674            album_key = prov_item_id.removeprefix(PODCAST_PREFIX)
675        elif media_type == MediaType.PODCAST_EPISODE and library_type == LIBRARY_TYPE_PODCASTS:
676            episode_key = prov_item_id.removeprefix(PODCAST_EPISODE_PREFIX)
677            plex_track = cast(
678                "PlexTrack",
679                await self._run_async(self._plex_library.fetchItem, episode_key, PlexTrack),
680            )
681            album_key = str(plex_track.parentKey)
682        else:
683            return
684
685        try:
686            plex_album = cast(
687                "PlexAlbum",
688                await self._run_async(self._plex_library.fetchItem, album_key, PlexAlbum),
689            )
690        except plexapi.exceptions.NotFound:
691            self.logger.warning(
692                "Failed to fetch %s %s for played sync", media_type.value, prov_item_id
693            )
694            return
695        except Exception:
696            self.logger.warning(
697                "Failed to fetch %s %s for played sync",
698                media_type.value,
699                prov_item_id,
700                exc_info=True,
701            )
702            return
703
704        if fully_played:
705            await self._run_async(plex_album.markPlayed)
706            self.logger.debug("Marked %s %s as played in Plex", media_type.value, prov_item_id)
707            return
708
709        if position <= 0:
710            await self._run_async(plex_album.markUnplayed)
711            self.logger.debug("Marked %s %s as unplayed in Plex", media_type.value, prov_item_id)
712            return
713
714        try:
715            target_track, target_offset_ms = await self._find_track_for_position(
716                plex_album, position
717            )
718            if target_track is None:
719                return
720
721            state = "playing" if is_playing else "paused"
722            # updateTimeline expects time in milliseconds (Plex native unit)
723            await self._run_async(
724                target_track.updateTimeline,
725                target_offset_ms,
726                state=state,
727                duration=getattr(target_track, "duration", None),
728            )
729            self.logger.debug(
730                "Synced %s %s progress to Plex: track %s at %dms (%s)",
731                media_type.value,
732                prov_item_id,
733                target_track.title,
734                target_offset_ms,
735                state,
736            )
737        except Exception:
738            self.logger.warning(
739                "Failed to sync %s %s progress to Plex",
740                media_type.value,
741                prov_item_id,
742                exc_info=True,
743            )
744
745    @use_cache(3600 * 3)  # Cache for 3 hours
746    async def get_album(self, prov_album_id: str) -> Album:
747        """Get full album details by id."""
748        plex_album = await self._get_data(prov_album_id, PlexAlbum)
749        return await self._parse_album(plex_album)
750
751    @use_cache(3600 * 3)  # Cache for 3 hours
752    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
753        """Get album tracks for given album id."""
754        plex_album: PlexAlbum = await self._get_data(prov_album_id, PlexAlbum)
755        tracks = []
756        for plex_track in await self._run_async(plex_album.tracks):
757            if (track := await self._parse_or_skip(self._parse_track, plex_track)) is not None:
758                tracks.append(track)
759        return tracks
760
761    @use_cache(3600 * 3)  # Cache for 3 hours
762    async def get_artist(self, prov_artist_id: str) -> Artist:
763        """Get full artist details by id."""
764        if prov_artist_id.startswith(FAKE_ARTIST_PREFIX):
765            # This artist does not exist in plex, so we can just load it from DB.
766
767            if db_artist := await self.mass.music.artists.get_library_item_by_prov_id(
768                prov_artist_id, self.instance_id
769            ):
770                return db_artist
771            raise MediaNotFoundError(ERR_ARTIST_NOT_FOUND.format(item_id=prov_artist_id))
772
773        plex_artist = await self._get_data(prov_artist_id, PlexArtist)
774        return await self._parse_artist(plex_artist)
775
776    @use_cache(3600 * 3)  # Cache for 3 hours
777    async def get_track(self, prov_track_id: str) -> Track:
778        """Get full track details by id."""
779        plex_track = await self._get_data(prov_track_id, PlexTrack)
780        track = await self._parse_track(plex_track)
781        await self._add_track_lyrics(plex_track, track)
782        return track
783
784    @use_cache(3600 * 3)  # Cache for 3 hours
785    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
786        """Get full playlist details by id."""
787        # Check if this is a collection (collections have the format "collection:<key>")
788        if prov_playlist_id.startswith(COLLECTION_ID_PREFIX):
789            collection_key = prov_playlist_id.removeprefix(COLLECTION_ID_PREFIX)
790            plex_collection: PlexObject = await self._get_data(collection_key)
791            return await self._parse_collection(plex_collection)
792
793        # "Mixes For You" items use a MIX_ITEM_PREFIX (see _build_mix_playlist).
794        if prov_playlist_id.startswith(MIX_ITEM_PREFIX):
795            mix_key = prov_playlist_id.removeprefix(MIX_ITEM_PREFIX)
796            fields = await self._find_mix_by_key(mix_key)
797            if fields is None:
798                msg = f"Mix {prov_playlist_id} not found"
799                raise MediaNotFoundError(msg)
800            _, title, thumb = fields
801            # Cache title/artwork on interaction so replay from recently-played
802            # still renders after Plex rotates the mix out of the hub.
803            if mix_key:
804                await self.mass.cache.set(
805                    key=mix_key,
806                    data={"title": title, "thumb": thumb},
807                    provider=self.instance_id,
808                    expiration=MIX_CACHE_EXPIRATION,
809                )
810            return self._build_mix_playlist(mix_key, title, thumb)
811
812        plex_playlist = await self._get_data(prov_playlist_id, PlexPlaylist)
813        return await self._parse_playlist(plex_playlist)
814
815    @use_cache(3600 * 3)  # Cache for 3 hours
816    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
817        """Get playlist tracks."""
818        result: list[Track] = []
819        if page > 0:
820            # paging not supported, we always return the whole list at once
821            return []
822
823        # Check if this is a collection (collections have the format "collection:<key>")
824        if prov_playlist_id.startswith(COLLECTION_ID_PREFIX):
825            collection_key = prov_playlist_id.removeprefix(COLLECTION_ID_PREFIX)
826            plex_collection: PlexObject = await self._get_data(collection_key)
827            if not (collection_items := await self._run_async(plex_collection.items)):
828                return result
829            # Collections can contain tracks, albums, or artists - we only want tracks
830            for item in collection_items:
831                if item.type == "track":
832                    if (track := await self._parse_or_skip(self._parse_track, item)) is not None:
833                        track.position = len(result) + 1
834                        result.append(track)
835                elif item.type == "album":
836                    # If the collection contains albums, get all tracks from each album
837                    album_tracks = await self.get_album_tracks(item.key)
838                    for album_track in album_tracks:
839                        album_track.position = len(result) + 1
840                        result.append(album_track)
841            return result
842
843        # "Mixes For You" items use a MIX_ITEM_PREFIX. Strip it to recover
844        # the Plex section-query key, append the track type filter to expand
845        # albums into tracks, then shuffle — Plexamp randomizes mix playback
846        # client-side.
847        if prov_playlist_id.startswith(MIX_ITEM_PREFIX):
848            mix_key = prov_playlist_id.removeprefix(MIX_ITEM_PREFIX)
849            tracks_key = f"{mix_key}&type={plexapi.utils.searchType('track')}"
850            plex_tracks = await self._run_async(self._plex_library.fetchItems, tracks_key)
851            random.shuffle(plex_tracks)
852            for plex_track in plex_tracks:
853                if (track := await self._parse_or_skip(self._parse_track, plex_track)) is not None:
854                    track.position = len(result) + 1
855                    result.append(track)
856            return result
857
858        plex_playlist: PlexPlaylist = await self._get_data(prov_playlist_id, PlexPlaylist)
859        if not (playlist_items := await self._run_async(plex_playlist.items)):
860            return result
861        for plex_track in playlist_items:
862            if (track := await self._parse_or_skip(self._parse_track, plex_track)) is not None:
863                track.position = len(result) + 1
864                result.append(track)
865        return result
866
867    @use_cache(3600 * 3)  # Cache for 3 hours
868    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
869        """Get a list of albums for the given artist."""
870        if not prov_artist_id.startswith(FAKE_ARTIST_PREFIX):
871            plex_artist = await self._get_data(prov_artist_id, PlexArtist)
872            try:
873                plex_albums = cast("list[PlexAlbum]", await self._run_async(plex_artist.albums))
874            except plexapi.exceptions.NotFound:
875                # PlexArtist.albums() relies on Plex's advanced filters API.
876                # Some Plex servers return no filtering metadata, making plexapi
877                # raise 'Unknown libtype "artist"'. Fall back to the artist's
878                # /children endpoint, which does not depend on the filters API.
879                albums_key = f"{plex_artist.key}/children"
880                plex_albums = cast(
881                    "list[PlexAlbum]",
882                    await self._run_async(plex_artist.fetchItems, albums_key, PlexAlbum),
883                )
884            if plex_albums:
885                albums = []
886                for album_obj in plex_albums:
887                    albums.append(await self._parse_album(album_obj))
888                return albums
889        return []
890
891    @use_cache(3600 * 3)  # Cache for 3 hours
892    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
893        """Get top tracks for the given artist."""
894        if prov_artist_id.startswith(FAKE_ARTIST_PREFIX):
895            return []
896        plex_artist = await self._get_data(prov_artist_id, PlexArtist)
897        try:
898            plex_tracks = cast("list[PlexTrack]", await self._run_async(plex_artist.popularTracks))
899        except plexapi.exceptions.NotFound:
900            # PlexArtist.popularTracks() relies on Plex's advanced filters API.
901            # Some Plex servers return no filtering metadata, making plexapi
902            # raise 'Unknown libtype "artist"'. Fall back to ranking the artist's
903            # own tracks, which does not depend on the filters API.
904            plex_tracks = await self._rank_artist_tracks(plex_artist)
905        return [await self._parse_track(plex_track) for plex_track in plex_tracks[:MAX_TOP_TRACKS]]
906
907    @use_cache(3600 * 3)  # Cache for 3 hours
908    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
909        """Get similar tracks using Plex's sonicallySimilar feature."""
910        try:
911            plex_track = await self._get_data(prov_track_id, PlexTrack)
912            # Get sonically similar tracks
913            similar_tracks = await self._run_async(plex_track.sonicallySimilar, limit=limit)
914            tracks = []
915            for similar_track in similar_tracks:
916                if track := await self._parse_track(similar_track):
917                    tracks.append(track)
918            self.logger.debug(
919                "Retrieved %d similar tracks for track %s", len(tracks), prov_track_id
920            )
921            return tracks
922        except Exception as err:
923            self.logger.warning("Error getting similar tracks for %s: %s", prov_track_id, err)
924        return []
925
926    async def get_recommendations(self) -> list[RecommendationFolder]:
927        """Get this provider's available recommendation rows, without items."""
928        return await self._recommendation_rows_from_payload()
929
930    async def get_recommendation_items(
931        self, item_id: str
932    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
933        """
934        Get the items for a single recommendation row.
935
936        :param item_id: The item_id of the row, as returned by get_recommendations.
937        """
938        return await self._recommendation_items_from_payload(item_id)
939
940    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
941        """Get streamdetails for a track/audiobook/podcast episode."""
942        if media_type == MediaType.AUDIOBOOK:
943            return await self._get_audiobook_stream_details(item_id)
944        if media_type == MediaType.PODCAST_EPISODE:
945            return await self._get_podcast_episode_stream_details(item_id)
946
947        plex_track = await self._get_data(item_id, PlexTrack)
948        if not plex_track.media:
949            raise MediaNotFoundError(ERR_TRACK_NOT_FOUND.format(item_id=item_id))
950
951        media: PlexMedia = plex_track.media[0]
952
953        content_type = (
954            ContentType.try_parse(media.container) if media.container else ContentType.UNKNOWN
955        )
956        media_part: PlexMediaPart = media.parts[0]
957        audio_streams = media_part.audioStreams()
958        audio_stream: PlexAudioStream | None = audio_streams[0] if audio_streams else None
959
960        stream_details = StreamDetails(
961            item_id=plex_track.key,
962            provider=self.instance_id,
963            audio_format=AudioFormat(
964                content_type=content_type,
965                channels=media.audioChannels,
966            ),
967            stream_type=StreamType.HTTP,
968            # plex reports duration in milliseconds, streamdetails expect seconds
969            duration=int(plex_track.duration / 1000) if plex_track.duration else None,
970            data=plex_track,
971            can_seek=True,
972            allow_seek=True,
973        )
974
975        download_url = self._plex_server.url(f"{media_part.key}?download=1", True)
976
977        if content_type != ContentType.M4A:
978            stream_details.path = download_url
979            if audio_stream and audio_stream.samplingRate:
980                stream_details.audio_format.sample_rate = audio_stream.samplingRate
981            if audio_stream and audio_stream.bitDepth:
982                stream_details.audio_format.bit_depth = audio_stream.bitDepth
983
984        else:
985            media_info = await async_parse_tags(download_url)
986            stream_details.path = download_url
987            stream_details.audio_format.channels = media_info.channels
988            stream_details.audio_format.content_type = ContentType.try_parse(media_info.format)
989            stream_details.audio_format.sample_rate = media_info.sample_rate
990            stream_details.audio_format.bit_depth = media_info.bits_per_sample
991
992        return stream_details
993
994    async def get_myplex_account_and_refresh_token(self, auth_token: str) -> MyPlexAccount:
995        """Get a MyPlexAccount object and refresh the token if needed."""
996        if auth_token == AUTH_TOKEN_UNAUTH:
997            return self._myplex_account
998
999        def _refresh_plex_token() -> MyPlexAccount:
1000            if self._myplex_account is None:
1001                myplex_account = MyPlexAccount(token=auth_token)
1002                self._myplex_account = myplex_account
1003            self._myplex_account.ping()
1004            return self._myplex_account
1005
1006        return await asyncio.to_thread(_refresh_plex_token)
1007
1008    async def set_favorite(self, prov_item_id: str, media_type: MediaType, favorite: bool) -> None:
1009        """Set favorite status by setting rating in Plex."""
1010        if favorite:
1011            # Set like rating
1012            rating = cast("float", self.config.get_value(CONF_PLEX_LIKE_RATING))
1013        else:
1014            # Set unlike rating
1015            rating = cast("float", self.config.get_value(CONF_PLEX_UNLIKE_RATING))
1016
1017        if media_type == MediaType.TRACK:
1018            plex_item: PlexTrack | PlexAlbum = await self._get_data(prov_item_id, PlexTrack)
1019        elif media_type == MediaType.ALBUM:
1020            plex_item = await self._get_data(prov_item_id, PlexAlbum)
1021        else:
1022            return
1023        await self._run_async(plex_item.rate, rating)
1024        self.logger.debug(
1025            "Set Plex rating to %s for %s with ID %s (ratingKey: %s)",
1026            rating,
1027            media_type.value,
1028            prov_item_id,
1029            plex_item.ratingKey,
1030        )
1031
1032    def _get_library_type(self) -> str:
1033        """Return the configured library type, defaulting to music."""
1034        return str(self.get_setup_value(CONF_LIBRARY_TYPE) or LIBRARY_TYPE_MUSIC)
1035
1036    async def _cleanup_stale_library_mappings(self) -> None:
1037        """Remove provider mappings that do not belong to the current library type."""
1038        if not self.mass.music.database:
1039            return
1040        valid_types = set(LIBRARY_TYPE_TO_MEDIA_TYPES.get(self._get_library_type(), ()))
1041        all_types = {t for types in LIBRARY_TYPE_TO_MEDIA_TYPES.values() for t in types}
1042        for media_type in all_types - valid_types:
1043            controller = self.mass.music.get_controller(media_type)
1044            query = (
1045                f"SELECT item_id FROM {DB_TABLE_PROVIDER_MAPPINGS} "
1046                f"WHERE media_type = '{media_type.value}' "
1047                f"AND provider_instance = '{self.instance_id}'"
1048            )
1049            rows = await self.mass.music.database.get_rows_from_query(query, limit=100000)
1050            if rows:
1051                self.logger.info(
1052                    "Cleaning up %d stale %s provider mapping(s)", len(rows), media_type.value
1053                )
1054            for db_row in rows:
1055                try:
1056                    await controller.remove_provider_mappings(db_row["item_id"], self.instance_id)
1057                except Exception as err:
1058                    self.logger.warning(
1059                        "Failed to remove stale %s provider mapping for %s: %s",
1060                        media_type.value,
1061                        db_row["item_id"],
1062                        err,
1063                    )
1064
1065    async def _rank_artist_tracks(self, plex_artist: PlexArtist) -> list[PlexTrack]:
1066        """
1067        Rank an artist's own tracks by popularity, keeping one version per title.
1068
1069        :param plex_artist: The Plex artist to rank the tracks of.
1070        """
1071        plex_tracks = cast("list[PlexTrack]", await self._run_async(plex_artist.tracks))
1072        best_per_title: dict[str, PlexTrack] = {}
1073        for plex_track in plex_tracks:
1074            if not plex_track.ratingCount:
1075                # ratingCount is the Last.fm scrobble count popularTracks() ranks on,
1076                # so a track without one has no rank. viewCount is local plays instead.
1077                continue
1078            title = (plex_track.title or "").casefold()
1079            best = best_per_title.get(title)
1080            if best is None or plex_track.ratingCount > best.ratingCount:
1081                best_per_title[title] = plex_track
1082        return sorted(best_per_title.values(), key=lambda track: track.ratingCount, reverse=True)
1083
1084    async def _run_async(
1085        self, call: Callable[Param, RetType], *args: Param.args, **kwargs: Param.kwargs
1086    ) -> RetType:
1087        await self.get_myplex_account_and_refresh_token(str(self.get_setup_value(CONF_AUTH_TOKEN)))
1088        return await asyncio.to_thread(call, *args, **kwargs)
1089
1090    async def _get_data(self, key: str, cls: type[PlexObjectT] | None = None) -> PlexObjectT:
1091        try:
1092            results = await self._run_async(self._plex_library.fetchItem, key, cls)
1093        except plexapi.exceptions.NotFound as err:
1094            raise MediaNotFoundError(ERR_ITEM_NOT_FOUND.format(item_id=key)) from err
1095        return cast("PlexObjectT", results)
1096
1097    def _get_item_mapping(self, media_type: MediaType, key: str, name: str) -> ItemMapping:
1098        """Get item mapping for a given media type, key, and name."""
1099        if not name:
1100            self.logger.info(
1101                "Received None or empty name for media item. Media type: %s, Key: %s",
1102                media_type,
1103                key,
1104            )
1105            name = UNKNOWN_NAME
1106
1107        mapped_name, mapped_version = parse_title_and_version(name)
1108
1109        if not mapped_name:
1110            self.logger.info(
1111                "Failed to map name for media item. Media type: %s, Key: %s, Original name: %s",
1112                media_type,
1113                key,
1114                name,
1115            )
1116            mapped_name = UNKNOWN_NAME
1117        if not mapped_version and media_type not in (MediaType.ALBUM, MediaType.TRACK):
1118            mapped_version = ""
1119
1120        return ItemMapping(
1121            media_type=media_type,
1122            item_id=key,
1123            provider=self.instance_id,
1124            name=mapped_name,
1125            version=mapped_version,
1126        )
1127
1128    async def _get_or_create_artist_by_name(self, artist_name: str) -> Artist | ItemMapping:
1129        if library_items := await self.mass.music.artists.get_library_items_by_query(
1130            search=artist_name, provider_filter=[self.instance_id]
1131        ):
1132            return ItemMapping.from_item(library_items[0])
1133
1134        artist_id = FAKE_ARTIST_PREFIX + artist_name
1135        return Artist(
1136            item_id=artist_id,
1137            name=artist_name or UNKNOWN_ARTIST,
1138            provider=self.instance_id,
1139            provider_mappings={
1140                ProviderMapping(
1141                    item_id=str(artist_id),
1142                    provider_domain=self.domain,
1143                    provider_instance=self.instance_id,
1144                )
1145            },
1146        )
1147
1148    async def _parse(self, plex_media: PlexObject) -> MediaItem | None:
1149        if plex_media.type == "artist":
1150            return await self._parse_artist(plex_media)
1151        if plex_media.type == "album":
1152            return await self._parse_album(plex_media)
1153        if plex_media.type == "track":
1154            return await self._parse_track(plex_media)
1155        if plex_media.type == "playlist":
1156            return await self._parse_playlist(plex_media)
1157        return None
1158
1159    async def _search_track(self, search_query: str, limit: int) -> list[PlexTrack]:
1160        return cast(
1161            "list[PlexTrack]",
1162            await self._run_async(self._plex_library.searchTracks, title=search_query, limit=limit),
1163        )
1164
1165    async def _search_album(self, search_query: str, limit: int) -> list[PlexAlbum]:
1166        return cast(
1167            "list[PlexAlbum]",
1168            await self._run_async(self._plex_library.searchAlbums, title=search_query, limit=limit),
1169        )
1170
1171    async def _search_artist(self, search_query: str, limit: int) -> list[PlexArtist]:
1172        return cast(
1173            "list[PlexArtist]",
1174            await self._run_async(
1175                self._plex_library.searchArtists, title=search_query, limit=limit
1176            ),
1177        )
1178
1179    async def _search_playlist(self, search_query: str, limit: int) -> list[PlexPlaylist]:
1180        return cast(
1181            "list[PlexPlaylist]",
1182            await self._run_async(self._plex_library.playlists, title=search_query, limit=limit),
1183        )
1184
1185    async def _search_and_parse(
1186        self,
1187        search_coro: Awaitable[list[PlexObjectT]],
1188        parse_coro: Callable[[PlexObjectT], Coroutine[Any, Any, MediaItemT]],
1189    ) -> list[MediaItemT]:
1190        task_results: list[Task[MediaItemT]] = []
1191        async with TaskGroup() as tg:
1192            for item in await search_coro:
1193                task_results.append(tg.create_task(parse_coro(item)))
1194
1195        results: list[MediaItemT] = []
1196        for task in task_results:
1197            results.append(task.result())
1198
1199        return results
1200
1201    async def _parse_or_skip(
1202        self,
1203        parse_coro: Callable[[PlexObjectT], Coroutine[Any, Any, MediaItemT]],
1204        plex_item: PlexObjectT,
1205    ) -> MediaItemT | None:
1206        """
1207        Parse a plex object into a media item, or return None if the item must be skipped.
1208
1209        :param parse_coro: The parse method to apply to the given plex object.
1210        :param plex_item: The plex object to parse.
1211        """
1212        try:
1213            return await parse_coro(plex_item)
1214        except InvalidDataError as err:
1215            # only an item we can not build a media item from is skippable. anything else
1216            # may be a server or connection failure rather than a property of this item,
1217            # and we can not tell those apart here, so it has to abort the sync - that is
1218            # what holds back the deletion pass that would otherwise drop valid items.
1219            #
1220            # read the identifiers from the cached payload, so reporting a failed item
1221            # can not trigger a reload that fails all over again
1222            attrib = plex_item._data.attrib
1223            self.logger.warning(
1224                "Skipping Plex item '%s' (key=%s): %s",
1225                attrib.get("title", "[unknown]"),
1226                attrib.get("key", "[no key]"),
1227                err,
1228            )
1229            return None
1230
1231    async def _parse_album(self, plex_album: PlexAlbum) -> Album:
1232        """Parse a Plex Album response to an Album model object."""
1233        album_id = plex_album.key
1234        album = Album(
1235            item_id=album_id,
1236            provider=self.instance_id,
1237            name=plex_album.title or UNKNOWN_NAME,
1238            provider_mappings={
1239                ProviderMapping(
1240                    item_id=str(album_id),
1241                    provider_domain=self.domain,
1242                    provider_instance=self.instance_id,
1243                    url=plex_album.getWebURL(self._baseurl),
1244                )
1245            },
1246        )
1247        # Check if album rating meets the configured threshold for favorites
1248        favorite_threshold = cast("float", self.config.get_value(CONF_PLEX_FAVORITE_THRESHOLD))
1249        if (favorite := get_favorite_from_rating(plex_album, favorite_threshold)) is not None:
1250            album.favorite = favorite
1251
1252        if plex_album.year:
1253            album.year = plex_album.year
1254        if images := get_thumbnail_images(plex_album, self.instance_id):
1255            album.metadata.images = images
1256        if plex_album.summary:
1257            album.metadata.description = plex_album.summary
1258        if plex_album.genres:
1259            album.metadata.genres = {genre.tag for genre in plex_album.genres if genre.tag}
1260        if plex_album.moods:
1261            album.metadata.mood = next((mood.tag for mood in plex_album.moods if mood.tag), None)
1262        if plex_album.styles:
1263            album.metadata.style = next(
1264                (style.tag for style in plex_album.styles if style.tag), None
1265            )
1266        if plex_album.originallyAvailableAt:
1267            album.metadata.release_date = plex_album.originallyAvailableAt
1268        if (explicit := get_explicit(plex_album)) is not None:
1269            album.metadata.explicit = explicit
1270        if mbid := clean_mbid(
1271            get_musicbrainz_id(plex_album), f"album {plex_album.title}", self.logger
1272        ):
1273            album.mbid = mbid
1274
1275        album.artists.append(
1276            self._get_item_mapping(
1277                MediaType.ARTIST,
1278                plex_album.parentKey,
1279                plex_album.parentTitle or UNKNOWN_ARTIST,
1280            )
1281        )
1282        return album
1283
1284    async def _parse_artist(self, plex_artist: PlexArtist) -> Artist:
1285        """Parse a Plex Artist response to Artist model object."""
1286        artist_id = plex_artist.key
1287        if not artist_id:
1288            raise InvalidDataError(ERR_ARTIST_INVALID_ID)
1289        artist = Artist(
1290            item_id=artist_id,
1291            name=plex_artist.title or UNKNOWN_ARTIST,
1292            provider=self.instance_id,
1293            provider_mappings={
1294                ProviderMapping(
1295                    item_id=str(artist_id),
1296                    provider_domain=self.domain,
1297                    provider_instance=self.instance_id,
1298                    url=plex_artist.getWebURL(self._baseurl),
1299                )
1300            },
1301        )
1302        if plex_artist.summary:
1303            artist.metadata.description = plex_artist.summary
1304        if images := get_thumbnail_images(plex_artist, self.instance_id):
1305            artist.metadata.images = images
1306        if plex_artist.genres:
1307            artist.metadata.genres = {genre.tag for genre in plex_artist.genres if genre.tag}
1308        if plex_artist.moods:
1309            artist.metadata.mood = next((mood.tag for mood in plex_artist.moods if mood.tag), None)
1310        if plex_artist.styles:
1311            artist.metadata.style = next(
1312                (style.tag for style in plex_artist.styles if style.tag), None
1313            )
1314        if mbid := clean_mbid(
1315            get_musicbrainz_id(plex_artist), f"artist {plex_artist.title}", self.logger
1316        ):
1317            artist.mbid = mbid
1318        return artist
1319
1320    async def _parse_playlist(self, plex_playlist: PlexPlaylist) -> Playlist:
1321        """Parse a Plex Playlist response to a Playlist object."""
1322        playlist = Playlist(
1323            item_id=plex_playlist.key,
1324            provider=self.instance_id,
1325            name=plex_playlist.title or UNKNOWN_NAME,
1326            provider_mappings={
1327                ProviderMapping(
1328                    item_id=plex_playlist.key,
1329                    provider_domain=self.domain,
1330                    provider_instance=self.instance_id,
1331                    url=plex_playlist.getWebURL(self._baseurl),
1332                )
1333            },
1334        )
1335        if plex_playlist.summary:
1336            playlist.metadata.description = plex_playlist.summary
1337        if images := get_thumbnail_images(plex_playlist, self.instance_id):
1338            playlist.metadata.images = images
1339        playlist.is_editable = not plex_playlist.smart
1340        return playlist
1341
1342    async def _parse_collection(self, plex_collection: PlexCollection) -> Playlist:
1343        """Parse a Plex Collection response to a Playlist object."""
1344        # Get the configured collection prefix
1345        collection_prefix = str(self.config.get_value(CONF_COLLECTION_PREFIX) or "")
1346
1347        # Collections are imported as playlists with the configured prefix
1348        playlist = Playlist(
1349            item_id=f"{COLLECTION_ID_PREFIX}{plex_collection.key}",
1350            provider=self.instance_id,
1351            name=f"{collection_prefix}{plex_collection.title}",
1352            provider_mappings={
1353                ProviderMapping(
1354                    item_id=f"{COLLECTION_ID_PREFIX}{plex_collection.key}",
1355                    provider_domain=self.domain,
1356                    provider_instance=self.instance_id,
1357                )
1358            },
1359        )
1360        # Add collection poster/thumbnail if available
1361        if images := get_thumbnail_images(
1362            plex_collection, self.instance_id, ("thumb", "composite")
1363        ):
1364            playlist.metadata.images = images
1365        # Collections are not editable in Music Assistant
1366        playlist.is_editable = False
1367        return playlist
1368
1369    def _mix_playlist_fields(self, plex_mix: PlexPlaylist) -> tuple[str, str, str | None]:
1370        """
1371        Extract (smart-query key, title, centroid thumb) from a 'Mix For You' item.
1372
1373        :param plex_mix: A Plex Playlist parsed from the 'Mixes For You' hub.
1374        """
1375        # Read straight from the parsed XML element. These synthetic mix playlists
1376        # carry a centroid-derived ratingKey rather than their own, so touching any
1377        # attribute that triggers a reload (e.g. .thumb) re-fetches the wrong object
1378        # and corrupts it. The smart-query key, title, and centroid artist thumb are
1379        # all present on the partial element itself.
1380        data = plex_mix._data
1381        mix_key = data.get("key") or ""
1382        title = data.get("title") or "[Unknown Mix]"
1383        thumb = next(
1384            (child.get("thumb") for child in data if child.get("centroid") and child.get("thumb")),
1385            None,
1386        )
1387        return mix_key, title, thumb
1388
1389    def _build_mix_playlist(self, mix_key: str, title: str, thumb: str | None) -> Playlist:
1390        """
1391        Build a MA Playlist from a Plex 'Mix For You' hub item.
1392
1393        :param mix_key: The Plex smart-query key identifying the mix.
1394        :param title: The mix title.
1395        :param thumb: The centroid artist thumb path, if any.
1396        """
1397        item_id = f"{MIX_ITEM_PREFIX}{mix_key}"
1398        playlist = Playlist(
1399            item_id=item_id,
1400            provider=self.instance_id,
1401            name=title,
1402            provider_mappings={
1403                ProviderMapping(
1404                    item_id=item_id,
1405                    provider_domain=self.domain,
1406                    provider_instance=self.instance_id,
1407                )
1408            },
1409        )
1410        if thumb:
1411            playlist.metadata.images = UniqueList(
1412                [
1413                    MediaItemImage(
1414                        type=ImageType.THUMB,
1415                        path=thumb,
1416                        provider=self.instance_id,
1417                        remotely_accessible=False,
1418                    )
1419                ]
1420            )
1421        playlist.is_editable = False
1422        playlist.is_dynamic = True
1423        return playlist
1424
1425    async def _get_mix_playlists(self, count: int) -> list[PlexPlaylist]:
1426        """
1427        Fetch the 'Mixes For You' hub items as Plex Playlist objects.
1428
1429        :param count: Maximum number of items per hub.
1430        """
1431        key = f"/hubs/sections/{self._plex_library.key}?count={count}&{RECOMMENDATIONS_HUB_PARAMS}"
1432        hubs = await self._run_async(self._plex_library.fetchItems, key)
1433        for hub in hubs:
1434            if "music.mixes" in (hub.hubIdentifier or ""):
1435                return list(hub._partialItems)
1436        return []
1437
1438    async def _find_mix_by_key(self, mix_key: str) -> tuple[str, str, str | None] | None:
1439        """Find a 'Mix For You' by its smart-query key, falling back to cache."""
1440        limit_value = self.config.get_value(CONF_HUB_ITEMS_LIMIT)
1441        limit = int(limit_value) if isinstance(limit_value, (int, float, str)) else 10
1442        for plex_mix in await self._get_mix_playlists(limit):
1443            fields = self._mix_playlist_fields(plex_mix)
1444            if fields[0] == mix_key:
1445                return fields
1446        # Plex rotates mixes out of the hub, but the smart-query key remains a
1447        # valid section query, so replay from recently-played still works — we
1448        # only need the cache to restore the title and artwork.
1449        cached = await self.mass.cache.get(key=mix_key, provider=self.instance_id)
1450        if isinstance(cached, dict):
1451            return mix_key, cached.get("title") or "[Unknown Mix]", cached.get("thumb")
1452        return None
1453
1454    async def _parse_track(self, plex_track: PlexTrack) -> Track:
1455        """Parse a Plex Track response to a Track model object."""
1456        content = plex_track.media[0].container if plex_track.media else None
1457        track = Track(
1458            item_id=plex_track.key,
1459            provider=self.instance_id,
1460            name=plex_track.title or UNKNOWN_NAME,
1461            provider_mappings={
1462                ProviderMapping(
1463                    item_id=plex_track.key,
1464                    provider_domain=self.domain,
1465                    provider_instance=self.instance_id,
1466                    # For Plex (local library provider), assume tracks are available by default
1467                    # even if media attribute is not populated in the initial response.
1468                    # This prevents tracks from being skipped during library sync.
1469                    available=True,
1470                    audio_format=AudioFormat(
1471                        content_type=(
1472                            ContentType.try_parse(content) if content else ContentType.UNKNOWN
1473                        ),
1474                    ),
1475                    url=plex_track.getWebURL(self._baseurl),
1476                )
1477            },
1478            disc_number=plex_track.parentIndex or 0,
1479            track_number=plex_track.trackNumber or 0,
1480        )
1481        # Check if track rating meets the configured threshold for favorites
1482        favorite_threshold = cast("float", self.config.get_value(CONF_PLEX_FAVORITE_THRESHOLD))
1483        if (favorite := get_favorite_from_rating(plex_track, favorite_threshold)) is not None:
1484            track.favorite = favorite
1485
1486        if plex_track.originalTitle and plex_track.originalTitle != plex_track.grandparentTitle:
1487            # The artist of the track if different from the album's artist.
1488            # For this kind of artist, we just know the name, so we create a fake artist,
1489            # if it does not already exist.
1490            track.artists.append(
1491                await self._get_or_create_artist_by_name(plex_track.originalTitle or UNKNOWN_ARTIST)
1492            )
1493        elif plex_track.grandparentKey:
1494            track.artists.append(
1495                self._get_item_mapping(
1496                    MediaType.ARTIST,
1497                    plex_track.grandparentKey,
1498                    plex_track.grandparentTitle or UNKNOWN_ARTIST,
1499                )
1500            )
1501        else:
1502            raise InvalidDataError(ERR_NO_ARTIST_FOR_TRACK)
1503
1504        if images := get_thumbnail_images(plex_track, self.instance_id):
1505            track.metadata.images = images
1506        if plex_track.genres:
1507            track.metadata.genres = {genre.tag for genre in plex_track.genres if genre.tag}
1508        if plex_track.moods:
1509            track.metadata.mood = next((mood.tag for mood in plex_track.moods if mood.tag), None)
1510        if (explicit := get_explicit(plex_track)) is not None:
1511            track.metadata.explicit = explicit
1512        if mbid := clean_mbid(
1513            get_musicbrainz_id(plex_track), f"track {plex_track.title}", self.logger
1514        ):
1515            track.mbid = mbid
1516        if plex_track.parentKey:
1517            track.album = self._get_item_mapping(
1518                MediaType.ALBUM, plex_track.parentKey, plex_track.parentTitle
1519            )
1520        if plex_track.duration:
1521            track.duration = int(plex_track.duration / 1000)
1522
1523        return track
1524
1525    async def _add_track_lyrics(self, plex_track: PlexTrack, track: Track) -> None:
1526        """
1527        Fetch the track's lyric stream from Plex and attach it to the metadata.
1528
1529        :param plex_track: The fully loaded Plex track to read lyric streams from.
1530        :param track: The Music Assistant track to populate with lyrics.
1531        """
1532
1533        def _fetch() -> str | None:
1534            stream = next((stream for stream in plex_track.lyricStreams() if stream.key), None)
1535            if stream is None:
1536                return None
1537            url = plex_track._server.url(stream.key, includeToken=True)
1538            response: requests.Response = plex_track._server._session.get(
1539                url, headers={"Accept": "application/json"}, timeout=30
1540            )
1541            response.raise_for_status()
1542            # plexapi's untyped session makes the response Any for mypy; force str
1543            return str(response.text)
1544
1545        try:
1546            content = await self._run_async(_fetch)
1547        except (requests.RequestException, plexapi.exceptions.PlexApiException) as err:
1548            self.logger.debug("Failed to fetch lyrics for %s: %s", plex_track.key, err)
1549            return
1550        if not content or (parsed := parse_plex_lyrics_payload(content)) is None:
1551            return
1552        lyrics, synced = parsed
1553        if synced:
1554            track.metadata.lrc_lyrics = lyrics
1555        else:
1556            track.metadata.lyrics = lyrics
1557
1558    async def _parse_audiobook(
1559        self, plex_album: PlexAlbum, *, include_chapters: bool = False
1560    ) -> Audiobook:
1561        """Parse a Plex Album from the audiobook library into an Audiobook model."""
1562        audiobook_id = f"{AUDIOBOOK_PREFIX}{plex_album.key}"
1563        audiobook = Audiobook(
1564            item_id=audiobook_id,
1565            provider=self.instance_id,
1566            name=plex_album.title or UNKNOWN_NAME,
1567            provider_mappings={
1568                ProviderMapping(
1569                    item_id=audiobook_id,
1570                    provider_domain=self.domain,
1571                    provider_instance=self.instance_id,
1572                    url=plex_album.getWebURL(self._baseurl),
1573                )
1574            },
1575        )
1576        # Author: parentTitle is the album artist; grandparentTitle is the album
1577        # artist parent (for multi-level nesting in Plex). Some setups vary.
1578        if author_name := plex_album.parentTitle or plex_album.grandparentTitle:
1579            audiobook.authors = UniqueList([author_name])
1580        if plex_album.summary:
1581            audiobook.metadata.description = plex_album.summary
1582        if plex_album.year and MINYEAR <= plex_album.year <= MAXYEAR:
1583            audiobook.metadata.release_date = datetime(plex_album.year, 1, 1, tzinfo=UTC)
1584        if images := get_thumbnail_images(plex_album, self.instance_id):
1585            audiobook.metadata.images = images
1586        # minified path: use album-level duration if Plex exposes it
1587        if album_duration := getattr(plex_album, "duration", None):
1588            audiobook.duration = int(album_duration / 1000)
1589
1590        if include_chapters:
1591            chapters = await self._build_audiobook_chapters(plex_album)
1592            audiobook.metadata.chapters = chapters
1593            if chapters and chapters[-1].end is not None:
1594                audiobook.duration = int(chapters[-1].end)
1595
1596        return audiobook
1597
1598    async def _build_audiobook_chapters(self, plex_album: PlexAlbum) -> list[MediaItemChapter]:
1599        """Build chapter list from Plex tracks, skipping tracks without playable media."""
1600        plex_tracks = cast("list[PlexTrack]", await self._run_async(plex_album.tracks))
1601        plex_tracks.sort(key=lambda t: (t.parentIndex or 0, t.trackNumber or 0))
1602        chapters: list[MediaItemChapter] = []
1603        cumulative = 0.0
1604        chapter_num = 0
1605        for plex_track in plex_tracks:
1606            if not plex_track.media or not plex_track.media[0].parts:
1607                continue
1608            chapter_num += 1
1609            # plex_track.duration is in milliseconds (Plex native unit)
1610            duration_s = (plex_track.duration or 0) / 1000.0
1611            chapters.append(
1612                MediaItemChapter(
1613                    position=chapter_num,
1614                    name=plex_track.title or f"{CHAPTER_PREFIX} {chapter_num}",
1615                    start=cumulative,
1616                    end=cumulative + duration_s,
1617                )
1618            )
1619            cumulative += duration_s
1620        return chapters
1621
1622    async def _parse_podcast(
1623        self, plex_album: PlexAlbum, *, include_episodes: bool = False
1624    ) -> Podcast:
1625        """Parse a Plex Album from the podcast library into a Podcast model."""
1626        podcast_id = f"{PODCAST_PREFIX}{plex_album.key}"
1627        podcast = Podcast(
1628            item_id=podcast_id,
1629            provider=self.instance_id,
1630            name=plex_album.title or UNKNOWN_NAME,
1631            provider_mappings={
1632                ProviderMapping(
1633                    item_id=podcast_id,
1634                    provider_domain=self.domain,
1635                    provider_instance=self.instance_id,
1636                    url=plex_album.getWebURL(self._baseurl),
1637                )
1638            },
1639        )
1640        publisher = plex_album.studio or plex_album.parentTitle or plex_album.grandparentTitle
1641        if publisher:
1642            podcast.publisher = publisher
1643        if plex_album.summary:
1644            podcast.metadata.description = plex_album.summary
1645        if plex_album.year and MINYEAR <= plex_album.year <= MAXYEAR:
1646            podcast.metadata.release_date = datetime(plex_album.year, 1, 1, tzinfo=UTC)
1647        if images := get_thumbnail_images(plex_album, self.instance_id):
1648            podcast.metadata.images = images
1649        if include_episodes:
1650            podcast.total_episodes = await self._count_podcast_episodes(plex_album)
1651        return podcast
1652
1653    async def _count_podcast_episodes(self, plex_album: PlexAlbum) -> int:
1654        """Count playable tracks without building full PodcastEpisode objects."""
1655        plex_tracks = cast("list[PlexTrack]", await self._run_async(plex_album.tracks))
1656        return sum(1 for t in plex_tracks if t.media and t.media[0].parts)
1657
1658    async def _build_podcast_episodes(self, plex_album: PlexAlbum) -> list[PodcastEpisode]:
1659        """Build episode list from Plex tracks, skipping tracks without playable media."""
1660        plex_tracks = cast("list[PlexTrack]", await self._run_async(plex_album.tracks))
1661        plex_tracks.sort(key=lambda t: (t.parentIndex or 0, t.trackNumber or 0))
1662        episodes: list[PodcastEpisode] = []
1663        episode_num = 0
1664        for plex_track in plex_tracks:
1665            if not plex_track.media or not plex_track.media[0].parts:
1666                continue
1667            episode_num += 1
1668            duration_s = (plex_track.duration or 0) / 1000.0
1669            episode = PodcastEpisode(
1670                item_id=f"{PODCAST_EPISODE_PREFIX}{plex_track.key}",
1671                provider=self.instance_id,
1672                name=plex_track.title or f"{EPISODE_PREFIX} {episode_num}",
1673                position=episode_num,
1674                duration=int(duration_s),
1675                podcast=ItemMapping(
1676                    media_type=MediaType.PODCAST,
1677                    item_id=f"{PODCAST_PREFIX}{plex_album.key}",
1678                    provider=self.instance_id,
1679                    name=plex_album.title or UNKNOWN_NAME,
1680                ),
1681                provider_mappings={
1682                    ProviderMapping(
1683                        item_id=f"{PODCAST_EPISODE_PREFIX}{plex_track.key}",
1684                        provider_domain=self.domain,
1685                        provider_instance=self.instance_id,
1686                        url=plex_track.getWebURL(self._baseurl),
1687                        audio_format=AudioFormat(
1688                            content_type=(
1689                                ContentType.try_parse(plex_track.media[0].container)
1690                                if plex_track.media[0].container
1691                                else ContentType.UNKNOWN
1692                            )
1693                        ),
1694                    )
1695                },
1696            )
1697            if images := get_thumbnail_images(plex_track, self.instance_id):
1698                episode.metadata.images = images
1699            episodes.append(episode)
1700        return episodes
1701
1702    async def _parse_podcast_episode(self, plex_track: PlexTrack) -> PodcastEpisode:
1703        """Parse a Plex Track from the podcast library into a PodcastEpisode model."""
1704        duration_s = (plex_track.duration or 0) / 1000.0
1705        content_type = ContentType.UNKNOWN
1706        if plex_track.media and plex_track.media[0].container:
1707            content_type = ContentType.try_parse(plex_track.media[0].container)
1708        episode = PodcastEpisode(
1709            item_id=f"{PODCAST_EPISODE_PREFIX}{plex_track.key}",
1710            provider=self.instance_id,
1711            name=plex_track.title or UNKNOWN_NAME,
1712            position=plex_track.trackNumber or 0,
1713            duration=int(duration_s),
1714            podcast=ItemMapping(
1715                media_type=MediaType.PODCAST,
1716                item_id=f"{PODCAST_PREFIX}{plex_track.parentKey}",
1717                provider=self.instance_id,
1718                name=plex_track.parentTitle or UNKNOWN_NAME,
1719            ),
1720            provider_mappings={
1721                ProviderMapping(
1722                    item_id=f"{PODCAST_EPISODE_PREFIX}{plex_track.key}",
1723                    provider_domain=self.domain,
1724                    provider_instance=self.instance_id,
1725                    url=plex_track.getWebURL(self._baseurl),
1726                    audio_format=AudioFormat(content_type=content_type),
1727                )
1728            },
1729        )
1730        if images := get_thumbnail_images(plex_track, self.instance_id):
1731            episode.metadata.images = images
1732        return episode
1733
1734    async def _calc_resume_position_ms(self, plex_album: PlexAlbum, fully_played: bool) -> int:
1735        """Calculate resume position from per-track viewOffset values."""
1736        plex_tracks = cast("list[PlexTrack]", await self._run_async(plex_album.tracks))
1737        plex_tracks.sort(key=lambda t: (t.parentIndex or 0, t.trackNumber or 0))
1738
1739        # Per-track durations and viewOffset are in milliseconds (Plex native).
1740        resume_position_ms = 0
1741        cumulative_ms = 0
1742        for plex_track in plex_tracks:
1743            track_offset = getattr(plex_track, "viewOffset", 0) or 0
1744            if track_offset > 0:
1745                # Use the last non-zero offset — for sequential listening this
1746                # is the final playback position; it also handles non-linear
1747                # skipping better than first-match.
1748                resume_position_ms = cumulative_ms + track_offset
1749            cumulative_ms += getattr(plex_track, "duration", 0) or 0
1750
1751        if resume_position_ms == 0 and fully_played:
1752            album_duration = getattr(plex_album, "duration", 0) or 0
1753            resume_position_ms = int(album_duration)
1754
1755        return resume_position_ms
1756
1757    async def _find_track_for_position(
1758        self, plex_album: PlexAlbum, position: int
1759    ) -> tuple[PlexTrack | None, int]:
1760        """Find the track and offset (ms) corresponding to the given position (s)."""
1761        plex_tracks = cast("list[PlexTrack]", await self._run_async(plex_album.tracks))
1762        plex_tracks.sort(key=lambda t: (t.parentIndex or 0, t.trackNumber or 0))
1763
1764        position_ms = position * 1000
1765        cumulative_ms = 0
1766        for plex_track in plex_tracks:
1767            track_duration = getattr(plex_track, "duration", 0) or 0
1768            if cumulative_ms + track_duration > position_ms:
1769                return plex_track, position_ms - cumulative_ms
1770            cumulative_ms += track_duration
1771
1772        if plex_tracks:
1773            # Position is past all tracks — clamp to end of the last track.
1774            last_track = plex_tracks[-1]
1775            last_duration = getattr(last_track, "duration", 0) or 0
1776            return last_track, last_duration
1777
1778        return None, 0
1779
1780    async def _get_audiobook_stream_details(self, item_id: str) -> StreamDetails:
1781        """Build multi-part StreamDetails for an audiobook (one part per Plex track)."""
1782        if self._get_library_type() != LIBRARY_TYPE_AUDIOBOOKS:
1783            msg = "Library not configured for audiobooks"
1784            raise MediaNotFoundError(msg)
1785        album_key = item_id.removeprefix(AUDIOBOOK_PREFIX)
1786        try:
1787            plex_album = cast(
1788                "PlexAlbum",
1789                await self._run_async(self._plex_library.fetchItem, album_key, PlexAlbum),
1790            )
1791        except plexapi.exceptions.NotFound:
1792            msg = f"Audiobook {item_id} not found"
1793            raise MediaNotFoundError(msg)
1794
1795        plex_tracks = cast("list[PlexTrack]", await self._run_async(plex_album.tracks))
1796        plex_tracks.sort(key=lambda t: (t.parentIndex or 0, t.trackNumber or 0))
1797
1798        parts, total_duration, first_container = self._build_stream_parts(plex_tracks, item_id)
1799        if not parts:
1800            self.logger.error(
1801                "Audiobook %s (%s) has no playable parts (%d tracks checked)",
1802                item_id,
1803                plex_album.title,
1804                len(plex_tracks),
1805            )
1806            msg = f"Audiobook {item_id} has no playable parts"
1807            raise MediaNotFoundError(msg)
1808
1809        self.logger.debug(
1810            "Built StreamDetails for audiobook %s with %d parts, total_duration=%.1fs",
1811            item_id,
1812            len(parts),
1813            total_duration,
1814        )
1815
1816        content_type = (
1817            ContentType.try_parse(first_container) if first_container else ContentType.UNKNOWN
1818        )
1819
1820        return StreamDetails(
1821            provider=self.instance_id,
1822            item_id=item_id,
1823            media_type=MediaType.AUDIOBOOK,
1824            audio_format=AudioFormat(content_type=content_type),
1825            stream_type=StreamType.HTTP,
1826            duration=int(total_duration),
1827            path=parts[0].path if len(parts) == 1 else parts,
1828            can_seek=True,
1829            allow_seek=True,
1830        )
1831
1832    async def _fetch_recommendation_payload(self) -> list[RecommendationFolder]:
1833        """Fetch the full recommendations payload (folders with items) from the Plex hubs."""
1834        # Let fetch errors propagate: the payload mixin serves the last cached payload
1835        # on a failed refresh, and returning [] here would be cached as a valid empty
1836        # result for the full TTL.
1837        # Get the configured limit for items per hub
1838        limit_value = self.config.get_value(CONF_HUB_ITEMS_LIMIT)
1839        limit = int(limit_value) if isinstance(limit_value, (int, float, str)) else 10
1840
1841        # Build the hubs key manually because plexapi's hubs() method
1842        # doesn't accept a count parameter to limit items per hub.
1843        extended = self.config.get_value(CONF_EXTENDED_RECOMMENDATIONS)
1844        hub_params = RECOMMENDATIONS_HUB_PARAMS if extended else "includeStations=1"
1845        key = f"/hubs/sections/{self._plex_library.key}?count={limit}&{hub_params}"
1846        hubs = await self._run_async(self._plex_library.fetchItems, key)
1847
1848        if not hubs:
1849            self.logger.debug("No hubs available from Plex")
1850            return []
1851
1852        self.logger.debug(
1853            "Fetching %d hubs (limit: %d items per hub)",
1854            len(hubs),
1855            limit,
1856        )
1857
1858        folders = []
1859        for hub in hubs:
1860            # Create a recommendation folder for each hub
1861            folder = RecommendationFolder(
1862                name=hub.title,
1863                item_id=f"{self.instance_id}_{hub.hubIdentifier}",
1864                provider=self.instance_id,
1865                icon="mdi-music",
1866            )
1867
1868            # Mixes For You are synthetic smart playlists; build them from
1869            # their partial hub items (see _mix_playlist_fields).
1870            if "music.mixes" in (hub.hubIdentifier or ""):
1871                folder.items.extend(
1872                    self._build_mix_playlist(*self._mix_playlist_fields(plex_mix))
1873                    for plex_mix in hub._partialItems
1874                )
1875                if folder.items:
1876                    folders.append(folder)
1877                continue
1878
1879            # Parse each item based on its type (limit to configured max)
1880            # Use _partialItems to respect the count limit from the hubs() call
1881            # rather than hub.items() which fetches ALL items if more is True
1882            # _partialItems is a cached property that's already loaded, so no need for async
1883            hub_items = hub._partialItems
1884            self.logger.debug(
1885                "Processing hub '%s' (%s) with %d partial items",
1886                hub.title,
1887                hub.hubIdentifier,
1888                len(hub_items),
1889            )
1890            for item in hub_items:
1891                try:
1892                    # Skip items without type attribute
1893                    if not hasattr(item, "type"):
1894                        self.logger.debug(
1895                            "Skipping item in hub '%s': no type attribute",
1896                            hub.title,
1897                        )
1898                        continue
1899
1900                    if parsed_item := await self._parse(item):
1901                        folder.items.append(parsed_item)  # type: ignore[arg-type]
1902                    else:
1903                        self.logger.debug(
1904                            "Skipping unsupported item type '%s' in hub '%s'",
1905                            item.type,
1906                            hub.title,
1907                        )
1908                except Exception as err:
1909                    self.logger.debug(
1910                        "Failed to parse item (type: %s) in hub '%s': %s",
1911                        getattr(item, "type", "unknown"),
1912                        hub.title,
1913                        str(err),
1914                    )
1915                    continue
1916
1917            # Only add folder if it has items
1918            if folder.items:
1919                folders.append(folder)
1920                self.logger.debug(
1921                    "Added hub '%s' (%s) with %d items",
1922                    hub.title,
1923                    hub.hubIdentifier,
1924                    len(folder.items),
1925                )
1926            else:
1927                self.logger.debug(
1928                    "Skipping hub '%s' (%s): no items after parsing",
1929                    hub.title,
1930                    hub.hubIdentifier,
1931                )
1932
1933        self.logger.debug("Retrieved %d recommendation folders from Plex", len(folders))
1934        return folders
1935
1936    def _build_stream_parts(
1937        self, plex_tracks: list[PlexTrack], item_id: str
1938    ) -> tuple[list[MultiPartPath], float, str | None]:
1939        """Convert Plex tracks to MultiPartPath entries for streaming."""
1940        parts: list[MultiPartPath] = []
1941        total_duration = 0.0
1942        first_container: str | None = None
1943        for plex_track in plex_tracks:
1944            media = self._track_media_or_log(plex_track, item_id)
1945            if media is None:
1946                continue
1947            if first_container is None and media.container:
1948                first_container = media.container
1949            media_part: PlexMediaPart = media.parts[0]
1950            url = self._plex_server.url(f"{media_part.key}?download=1", True)
1951            duration_s = (plex_track.duration or 0) / 1000.0
1952            parts.append(MultiPartPath(path=url, duration=duration_s))
1953            total_duration += duration_s
1954            self.logger.debug(
1955                "Added audiobook part: track '%s' (%s) duration=%.1fs url=%s",
1956                plex_track.title,
1957                plex_track.key,
1958                duration_s,
1959                url,
1960            )
1961        return parts, total_duration, first_container
1962
1963    def _track_media_or_log(self, plex_track: PlexTrack, item_id: str) -> PlexMedia | None:
1964        """Return the first PlexMedia for a track, or log and return None if unavailable."""
1965        if not plex_track.media:
1966            self.logger.debug(
1967                "Skipping track '%s' (key=%s) in audiobook %s: no media",
1968                plex_track.title,
1969                plex_track.key,
1970                item_id,
1971            )
1972            return None
1973        media: PlexMedia = plex_track.media[0]
1974        if not media.parts:
1975            self.logger.debug(
1976                "Skipping track '%s' (key=%s) in audiobook %s: media has no parts",
1977                plex_track.title,
1978                plex_track.key,
1979                item_id,
1980            )
1981            return None
1982        return media
1983
1984    async def _get_podcast_episode_stream_details(self, item_id: str) -> StreamDetails:
1985        """Build streamdetails for a single podcast episode from a Plex track."""
1986        if self._get_library_type() != LIBRARY_TYPE_PODCASTS:
1987            msg = "Library not configured for podcasts"
1988            raise MediaNotFoundError(msg)
1989        track_key = item_id.removeprefix(PODCAST_EPISODE_PREFIX)
1990        try:
1991            plex_track = cast(
1992                "PlexTrack",
1993                await self._run_async(self._plex_library.fetchItem, track_key, PlexTrack),
1994            )
1995        except plexapi.exceptions.NotFound:
1996            msg = f"Podcast episode {item_id} not found"
1997            raise MediaNotFoundError(msg)
1998
1999        if not plex_track.media:
2000            msg = f"Podcast episode {item_id} has no media"
2001            raise MediaNotFoundError(msg)
2002
2003        media: PlexMedia = plex_track.media[0]
2004        if not media.parts:
2005            msg = f"Podcast episode {item_id} has no playable media parts"
2006            raise MediaNotFoundError(msg)
2007        content_type = (
2008            ContentType.try_parse(media.container) if media.container else ContentType.UNKNOWN
2009        )
2010        media_part: PlexMediaPart = media.parts[0]
2011        download_url = self._plex_server.url(f"{media_part.key}?download=1", True)
2012
2013        return StreamDetails(
2014            provider=self.instance_id,
2015            item_id=item_id,
2016            media_type=MediaType.PODCAST_EPISODE,
2017            audio_format=AudioFormat(content_type=content_type),
2018            stream_type=StreamType.HTTP,
2019            duration=plex_track.duration,
2020            path=download_url,
2021            can_seek=True,
2022            allow_seek=True,
2023        )
2024