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