music-assistant-server

1"""
2Media resolution for the Player Queues controller.
3
4Resolves source media items (artist, album, genre, playlist, audiobook, podcast, browse folder)
5into the concrete tracks / playable items that enqueueing them produces, honoring the user's
6per-type selection preferences. Pure media->tracks logic dispatched out of the controller; it reads
7config and the music controller via its owning controller, and holds no per-queue state.
8"""
9
10from __future__ import annotations
11
12import asyncio
13import random
14from types import NoneType
15from typing import TYPE_CHECKING, Any, cast
16
17from music_assistant_models.enums import ArtistType, MediaType
18from music_assistant_models.errors import InvalidDataError, MediaNotFoundError
19from music_assistant_models.media_items import (
20    Album,
21    Artist,
22    Audiobook,
23    BrowseFolder,
24    Genre,
25    ItemMapping,
26    MediaCollection,
27    MediaItemType,
28    Playlist,
29    Podcast,
30    PodcastEpisode,
31    Track,
32    UniqueList,
33)
34
35from music_assistant.constants import PlaylistPlayableItem
36from music_assistant.controllers.player_queues.constants import (
37    CONF_DEFAULT_ENQUEUE_SELECT_ALBUM,
38    CONF_DEFAULT_ENQUEUE_SELECT_ARTIST,
39    ENQUEUE_SELECT_ALBUM_DEFAULT_VALUE,
40    ENQUEUE_SELECT_ARTIST_DEFAULT_VALUE,
41)
42from music_assistant.controllers.player_queues.helpers import sort_tracks
43from music_assistant.controllers.webserver.helpers.auth_middleware import ImpersonatedUser
44from music_assistant.helpers.collections import (
45    get_collection_item_id,
46    get_collection_item_media_type_from_item_id,
47)
48
49if TYPE_CHECKING:
50    from collections.abc import Sequence
51
52    from music_assistant.controllers.player_queues.controller import PlayerQueuesController
53
54_LATEST_EPISODE_KEYWORDS = frozenset({"latest", "newest"})
55_START_ITEM_SUBSTRING_MIN_LEN = 3
56
57
58def _start_item_matches(start_item: str, item: Any) -> bool:
59    """
60    Return whether `item` satisfies a `start_item` directive.
61
62    :param start_item: Exact `item_id` / `uri`, or a case-insensitive
63        substring of the item's name.
64    :param item: Candidate media item.
65    """
66    if start_item in (getattr(item, "item_id", None), getattr(item, "uri", None)):
67        return True
68    if len(start_item) < _START_ITEM_SUBSTRING_MIN_LEN:
69        return False
70    name = getattr(item, "name", None)
71    return bool(name and start_item.lower() in name.lower())
72
73
74class MediaResolver:
75    """Resolve source media items into the concrete tracks/playable items to enqueue."""
76
77    def __init__(self, queues: PlayerQueuesController) -> None:
78        """
79        Initialize the media resolver.
80
81        :param queues: The owning player queues controller.
82        """
83        self.queues = queues
84        self.mass = queues.mass
85        self.logger = queues.logger.getChild("media_resolver")
86
87    async def get_tracks_for_playback(self, media_item: MediaItemType) -> list[Track]:
88        """
89        Return the playable tracks for a media item, honoring the user's selection preferences.
90
91        Resolves an umbrella media item (artist, album, genre, playlist) to the tracks that
92        playing it would enqueue; a track resolves to itself, other types to an empty list.
93
94        :param media_item: The media item to resolve to playable tracks.
95        """
96        if media_item.media_type == MediaType.TRACK:
97            return [cast("Track", media_item)]
98        if media_item.media_type == MediaType.ALBUM:
99            return await self.get_album_tracks(cast("Album", media_item), None)
100        if media_item.media_type == MediaType.ARTIST:
101            return await self.get_artist_tracks(cast("Artist", media_item))
102        if media_item.media_type == MediaType.GENRE:
103            return await self.get_genre_tracks(cast("Genre", media_item), None)
104        if media_item.media_type == MediaType.PLAYLIST:
105            return [
106                track
107                for track in await self.get_playlist_tracks(cast("Playlist", media_item), None)
108                if isinstance(track, Track)
109            ]
110        return []
111
112    async def get_artist_tracks(self, artist: Artist) -> list[Track]:
113        """Return the tracks to play for the given artist, based on user preference."""
114        artist_items_conf = self.mass.config.get_raw_core_config_value(
115            self.queues.domain,
116            CONF_DEFAULT_ENQUEUE_SELECT_ARTIST,
117            ENQUEUE_SELECT_ARTIST_DEFAULT_VALUE,
118        )
119        self.logger.info(
120            "Fetching tracks to play for artist %s (selection: %s)", artist.name, artist_items_conf
121        )
122        if artist_items_conf == "top_tracks":
123            tracks = await self.mass.music.artists.top_tracks(artist.item_id, artist.provider)
124            random.shuffle(tracks)
125            return tracks
126        # legacy "library_album_tracks" also resolves to the in-library tracks
127        if artist_items_conf in ("library_tracks", "library_album_tracks"):
128            tracks = await self._library_artist_tracks(artist)
129            random.shuffle(tracks)
130            return tracks
131        if artist_items_conf == "prefer_library":
132            tracks = await self._library_artist_tracks(artist)
133            if not tracks:
134                tracks = await self.mass.music.artists.top_tracks(artist.item_id, artist.provider)
135            random.shuffle(tracks)
136            return tracks
137        result: list[Track] = []
138        seen: set[str] = set()
139        sources = await asyncio.gather(
140            self._library_artist_tracks(artist),
141            self._provider_artist_tracks(artist),
142            return_exceptions=True,
143        )
144        for source in sources:
145            if isinstance(source, BaseException):
146                self.logger.warning(
147                    "Error resolving some tracks for artist %s", artist.name, exc_info=source
148                )
149                continue
150            for track in source:
151                unique_id = f"{track.name}.{track.version}"
152                if unique_id in seen:
153                    continue
154                seen.add(unique_id)
155                result.append(track)
156        random.shuffle(result)
157        return result
158
159    async def get_album_tracks(
160        self,
161        album: Album,
162        start_item: str | None,
163        sort_by: str | None = None,
164        keep_preceding_items: bool = False,
165    ) -> list[Track]:
166        """
167        Return tracks for given album, based on user preference.
168
169        :param album: The album to fetch the tracks for.
170        :param start_item: Optional item_id/uri of the track to start from.
171        :param sort_by: Optional sort key to order the tracks by before applying start_item.
172        :param keep_preceding_items: Move the tracks before start_item behind the rest instead
173            of dropping them, so the full album is returned with start_item first.
174        """
175        album_items_conf = self.mass.config.get_raw_core_config_value(
176            self.queues.domain,
177            CONF_DEFAULT_ENQUEUE_SELECT_ALBUM,
178            ENQUEUE_SELECT_ALBUM_DEFAULT_VALUE,
179        )
180        result: list[Track] = []
181        self.logger.info(
182            "Fetching tracks to play for album %s",
183            album.name,
184        )
185        for album_track in await self.mass.music.albums.tracks(
186            item_id=album.item_id,
187            provider_instance_id_or_domain=album.provider,
188            in_library_only=album_items_conf == "library_tracks",
189        ):
190            if not album_track.available:
191                continue
192            result.append(album_track)
193        if sort_by and sort_by != "track_number":
194            result = sort_tracks(result, sort_by)
195        if start_item is not None:
196            for idx, track in enumerate(result):
197                if start_item in (track.item_id, track.uri):
198                    return result[idx:] + (result[:idx] if keep_preceding_items else [])
199            return []
200        return result
201
202    async def get_genre_tracks(self, genre: Genre, start_item: str | None) -> list[Track]:
203        """
204        Return tracks for given genre, based on alias mappings.
205
206        Limits results to avoid loading thousands of tracks for broad genres.
207        Directly mapped tracks are fetched with random ordering, then supplemented
208        with tracks from a limited set of mapped albums and artists.
209        """
210        result: list[Track] = []
211        start_item_found = False
212        self.logger.info(
213            "Fetching tracks to play for genre %s",
214            genre.name,
215        )
216        tracks, albums, artists = await self.mass.music.genres.mapped_media(
217            genre,
218            track_limit=25,
219            album_limit=5,
220            artist_limit=5,
221            order_by="random",
222        )
223
224        for genre_track in tracks:
225            if not genre_track.available:
226                continue
227            if start_item in (genre_track.item_id, genre_track.uri):
228                start_item_found = True
229            if start_item is not None and not start_item_found:
230                continue
231            result.append(genre_track)
232
233        for album in albums:
234            album_tracks = await self.get_album_tracks(album, None)
235            result.extend(album_tracks[:5])
236
237        for artist in artists:
238            artist_tracks = await self.get_artist_tracks(artist)
239            result.extend(artist_tracks[:5])
240        return result
241
242    async def get_playlist_tracks(
243        self,
244        playlist: Playlist,
245        start_item: str | None,
246        sort_by: str | None = None,
247        keep_preceding_items: bool = False,
248    ) -> list[PlaylistPlayableItem]:
249        """
250        Return tracks for given playlist, based on user preference.
251
252        :param playlist: The playlist to fetch the tracks for.
253        :param start_item: Optional item_id/uri/name of the track to start from.
254        :param sort_by: Optional sort key to order the tracks by before applying start_item.
255        :param keep_preceding_items: Move the tracks before start_item behind the rest instead
256            of dropping them, so the full playlist is returned with start_item first.
257        """
258        result: list[PlaylistPlayableItem] = []
259        self.logger.info(
260            "Fetching tracks to play for playlist %s",
261            playlist.name,
262        )
263        force_refresh = playlist.is_dynamic
264        needs_sort = sort_by is not None and sort_by != "position"
265        # Fast path: no re-sort needed and the preceding tracks are dropped anyway, so
266        # skip-until-found in a single pass and never materialize huge playlists when
267        # starting near the end.
268        if not needs_sort and not keep_preceding_items:
269            start_item_found = False
270            async for playlist_track in self.mass.music.playlists.tracks(
271                playlist.item_id,
272                playlist.provider,
273                force_refresh=force_refresh,
274                allow_dynamic_tracks=playlist.is_dynamic,
275            ):
276                if not playlist_track.available:
277                    continue
278                if start_item is not None and _start_item_matches(start_item, playlist_track):
279                    start_item_found = True
280                if start_item is not None and not start_item_found:
281                    continue
282                result.append(playlist_track)
283            return result
284        # Sort/rotate path: must materialize all tracks before sorting or rotating, then slice.
285        async for playlist_track in self.mass.music.playlists.tracks(
286            playlist.item_id,
287            playlist.provider,
288            force_refresh=force_refresh,
289            allow_dynamic_tracks=playlist.is_dynamic,
290        ):
291            if not playlist_track.available:
292                continue
293            result.append(playlist_track)
294        if needs_sort:
295            result = sort_tracks(result, cast("str", sort_by))
296        if start_item is not None:
297            for idx, track in enumerate(result):
298                if _start_item_matches(start_item, track):
299                    return result[idx:] + (result[:idx] if keep_preceding_items else [])
300            return []
301        return result
302
303    async def get_audiobook_resume_point(
304        self, audio_book: Audiobook, chapter: str | int | None = None, userid: str | None = None
305    ) -> int:
306        """Return resume point (in milliseconds) for given audio book."""
307        self.logger.debug(
308            "Fetching resume point to play for audio book %s",
309            audio_book.name,
310        )
311        if chapter is not None:
312            # user explicitly selected a chapter to play
313            start_chapter = int(chapter) if isinstance(chapter, str) else chapter
314            if chapters := audio_book.metadata.chapters:
315                if _chapter := next((x for x in chapters if x.position == start_chapter), None):
316                    return int(_chapter.start * 1000)
317            raise InvalidDataError(
318                f"Unable to resolve chapter to play for Audiobook {audio_book.name}"
319            )
320        full_played, resume_position_ms = await self.mass.music.get_resume_position(
321            audio_book, userid=userid
322        )
323        return 0 if full_played else resume_position_ms
324
325    async def get_next_podcast_episodes(
326        self,
327        podcast: Podcast | None,
328        episode: PodcastEpisode | str | None,
329        userid: str | None = None,
330        start_from_beginning: bool = False,
331    ) -> UniqueList[PodcastEpisode]:
332        """
333        Return the next episode(s) and resume point for the given podcast.
334
335        :param podcast: Podcast to enqueue, or `None` if `episode` is a
336            concrete `PodcastEpisode`.
337        :param episode: A concrete `PodcastEpisode`, an `item_id` / `uri`,
338            a case-insensitive substring of an episode name, or one of the
339            reserved lowercase keywords `"latest"` / `"newest"`.
340        :param userid: User whose resume position should be applied.
341        :param start_from_beginning: When True, the resolved episode starts at position 0,
342            ignoring any saved resume position. The stored progress itself is left untouched.
343        """
344        if podcast is None and isinstance(episode, str | NoneType):
345            raise InvalidDataError("Either podcast or episode must be provided")
346        if podcast is None:
347            # single podcast episode requested
348            assert isinstance(episode, PodcastEpisode)  # checked above
349            self.logger.debug(
350                "Fetching resume point to play for Podcast episode %s",
351                episode.name,
352            )
353            await self._set_episode_resume_point(episode, userid, start_from_beginning)
354            return UniqueList([episode])
355        # podcast with optional start episode requested
356        self.logger.debug(
357            "Fetching episode(s) and resume point to play for Podcast %s",
358            podcast.name,
359        )
360        # Require exact case and keyword match to minimise false positives.
361        if isinstance(episode, str) and episode in _LATEST_EPISODE_KEYWORDS:
362            # provider yields newest-first, so only pull the first episode here and skip
363            # materialising the rest, which avoids a per-episode resume lookup on each one
364            latest = await anext(
365                self.mass.music.podcasts.episodes(podcast.item_id, podcast.provider), None
366            )
367            if latest is None:
368                raise InvalidDataError(
369                    f"Unable to resolve episode to play for Podcast {podcast.name}"
370                )
371            await self._set_episode_resume_point(latest, userid, start_from_beginning)
372            return UniqueList([latest])
373        all_episodes = [
374            x async for x in self.mass.music.podcasts.episodes(podcast.item_id, podcast.provider)
375        ]
376        all_episodes.sort(key=lambda x: x.position)
377        # if a episode was provided, a user explicitly selected a episode to play
378        # so we need to find the index of the episode in the list
379        resolved_episode: PodcastEpisode | None = None
380        if isinstance(episode, PodcastEpisode):
381            resolved_episode = next((x for x in all_episodes if x.uri == episode.uri), None)
382            if resolved_episode:
383                # ensure we have accurate resume info
384                (
385                    fully_played,
386                    resume_position_ms,
387                ) = await self.mass.music.get_resume_position(resolved_episode, userid=userid)
388                resolved_episode.resume_position_ms = 0 if fully_played else resume_position_ms
389        elif isinstance(episode, str):
390            resolved_episode = next(
391                (x for x in all_episodes if _start_item_matches(episode, x)), None
392            )
393            if resolved_episode:
394                # ensure we have accurate resume info
395                (
396                    fully_played,
397                    resume_position_ms,
398                ) = await self.mass.music.get_resume_position(resolved_episode, userid=userid)
399                resolved_episode.resume_position_ms = 0 if fully_played else resume_position_ms
400        else:
401            # get first episode that is not fully played
402            for ep in all_episodes:
403                if ep.fully_played:
404                    continue
405                # ensure we have accurate resume info
406                (
407                    fully_played,
408                    resume_position_ms,
409                ) = await self.mass.music.get_resume_position(ep, userid=userid)
410                if fully_played:
411                    continue
412                ep.resume_position_ms = resume_position_ms
413                resolved_episode = ep
414                break
415            else:
416                # no episodes found that are not fully played, so we start at the beginning
417                resolved_episode = next((x for x in all_episodes), None)
418        if resolved_episode is None:
419            raise InvalidDataError(f"Unable to resolve episode to play for Podcast {podcast.name}")
420        if start_from_beginning:
421            # play the resolved episode from position 0 without touching stored progress
422            resolved_episode.fully_played = False
423            resolved_episode.resume_position_ms = 0
424        # get the index of the episode
425        episode_index = all_episodes.index(resolved_episode)
426        # return the (remaining) episode(s) to play
427        return UniqueList(all_episodes[episode_index:])
428
429    async def get_next_podcast_episode(
430        self, episode: PodcastEpisode, userid: str | None = None
431    ) -> PodcastEpisode | None:
432        """
433        Return the episode to play after the given one, or None if there is none left.
434
435        Episodes are walked in the same order a full podcast enqueue produces, skipping the
436        ones that were already fully played.
437
438        :param episode: The episode that is being continued.
439        :param userid: User whose resume position should be applied.
440        """
441        podcast = episode.podcast
442        all_episodes = [
443            x async for x in self.mass.music.podcasts.episodes(podcast.item_id, podcast.provider)
444        ]
445        all_episodes.sort(key=lambda x: x.position)
446        current_index = next(
447            (idx for idx, x in enumerate(all_episodes) if x.uri == episode.uri), None
448        )
449        if current_index is None:
450            # the episode is no longer part of the feed, so we have nothing to continue from
451            return None
452        for candidate in all_episodes[current_index + 1 :]:
453            if candidate.fully_played:
454                continue
455            # ensure we have accurate resume info
456            fully_played, resume_position_ms = await self.mass.music.get_resume_position(
457                candidate, userid=userid
458            )
459            if fully_played:
460                continue
461            candidate.resume_position_ms = resume_position_ms
462            return candidate
463        return None
464
465    async def get_next_audiobook(
466        self, audiobook: Audiobook, userid: str | None = None
467    ) -> Audiobook | None:
468        """
469        Return the next book in the collection(s) the given book belongs to, if there is one.
470
471        Returns None for a standalone book and for a book whose collection has no not-fully-played
472        book left after it.
473
474        :param audiobook: The audiobook that is being continued.
475        :param userid: User whose resume position should be applied.
476        """
477        # collections are built from the library metadata, so a book that is not in the
478        # library has no series to continue with
479        library_item = (
480            audiobook
481            if audiobook.provider == "library"
482            else await self.mass.music.audiobooks.get_library_item_by_prov_id(
483                audiobook.item_id, audiobook.provider
484            )
485        )
486        if library_item is None:
487            return None
488        for collection in library_item.metadata.collections or []:
489            try:
490                media_collection = await self.mass.music.audiobooks.get_collection(
491                    get_collection_item_id(collection.title, MediaType.AUDIOBOOK)
492                )
493            except MediaNotFoundError:
494                continue
495            if next_book := await self._next_unplayed_book(media_collection, library_item, userid):
496                return next_book
497        return None
498
499    async def get_author_narrator_audiobooks(
500        self, author_narrator: Artist, userid: str | None
501    ) -> list[Audiobook]:
502        """
503        Return audiobooks to play of a given artist.
504
505        If all books are played, enqueue all of them. If not, enqueue books in a collection's order
506        if they are part of a collection.
507        """
508        audiobooks: UniqueList[Audiobook] = UniqueList([])
509        async with ImpersonatedUser(self.mass, user=userid):
510            # ensure we get the position status on the current user
511            all_audiobooks = await self.mass.music.artists.audiobooks(
512                author_narrator.item_id, author_narrator.provider, author_narrator.artist_type
513            )
514        for book in all_audiobooks:
515            # do not use get_resume_position here, as an artist may potentially have a lot of audiobooks,
516            # resulting in many API calls.
517            if book.fully_played:
518                continue
519            audiobooks.append(book)
520        if len(audiobooks) == 0:
521            audiobooks = UniqueList(all_audiobooks)
522
523        # treat books part of a collection separately by keeping the collections order
524        collections: list[MediaCollection[Audiobook]] = []
525        collection_item_ids: list[str] = []
526
527        books_with_collection: dict[str, set[str]] = {}  # book_item_id: {collection_ids}
528        for book in audiobooks:
529            for media_item_collection in book.metadata.collections or []:
530                collection_item_id = get_collection_item_id(
531                    media_item_collection.title, MediaType.AUDIOBOOK
532                )
533                if collection_item_id not in collection_item_ids:
534                    collection_item_ids.append(collection_item_id)
535                entry = books_with_collection.get(book.item_id, set())
536                entry.add(collection_item_id)
537                books_with_collection[book.item_id] = entry
538        async with ImpersonatedUser(self.mass, user=userid):
539            for collection_item_id in collection_item_ids:
540                try:
541                    collection = await self.mass.music.audiobooks.get_collection(collection_item_id)
542                    collections.append(collection)
543                except MediaNotFoundError:
544                    # Remove invalid collection everywhere
545                    for book_collections in books_with_collection.values():
546                        book_collections.discard(collection_item_id)
547                    continue
548        # ensure, that books with collection only holds books which have a verified collection
549        books_with_collection = {
550            book_item_id: collection_ids
551            for book_item_id, collection_ids in books_with_collection.items()
552            if collection_ids
553        }
554
555        # remove books which are part of a collection
556        audiobooks = UniqueList(
557            [book for book in audiobooks if book.item_id not in books_with_collection]
558        )
559        # enqueue books which are part of a collection in the collection's order, however, as a collection
560        # may have books of different artists, only enqueue the books which belong to the artist.
561        # if a book happens to be part of multiple collections, only enqueue once
562        books_with_collection_sorted: list[Audiobook] = []
563        for collection in collections:
564            for book in collection.items:
565                if (
566                    book.item_id in books_with_collection
567                    and book not in books_with_collection_sorted
568                ):
569                    books_with_collection_sorted.append(book)
570
571        return list(audiobooks) + books_with_collection_sorted
572
573    async def _set_episode_resume_point(
574        self, episode: PodcastEpisode, userid: str | None, start_from_beginning: bool
575    ) -> None:
576        """
577        Apply the resume point to a resolved podcast episode.
578
579        When start_from_beginning is set the episode starts at position 0 and the resume
580        lookup is skipped; the stored progress itself is left untouched.
581        """
582        if start_from_beginning:
583            episode.fully_played = False
584            episode.resume_position_ms = 0
585            return
586        fully_played, resume_position_ms = await self.mass.music.get_resume_position(
587            episode, userid=userid
588        )
589        episode.fully_played = fully_played
590        episode.resume_position_ms = 0 if fully_played else resume_position_ms
591
592    async def _next_unplayed_book(
593        self,
594        collection: MediaCollection[Audiobook],
595        current: Audiobook,
596        userid: str | None,
597    ) -> Audiobook | None:
598        """Return the first not fully played book after `current` in the given collection."""
599        books = [x for x in collection.items if isinstance(x, Audiobook)]
600        current_index = next(
601            (idx for idx, x in enumerate(books) if x.item_id == current.item_id), None
602        )
603        if current_index is None:
604            return None
605        for candidate in books[current_index + 1 :]:
606            fully_played, resume_position_ms = await self.mass.music.get_resume_position(
607                candidate, userid=userid
608            )
609            if fully_played:
610                continue
611            candidate.resume_position_ms = resume_position_ms
612            return candidate
613        return None
614
615    async def _resolve_library_artist(self, artist: Artist) -> Artist | None:
616        """
617        Resolve the in-library artist for the given (possibly provider) artist item.
618
619        :param artist: The artist item, which may be a library or a provider item.
620        """
621        if artist.provider == "library":
622            return artist
623        return await self.mass.music.artists.get_library_item_by_prov_id(
624            artist.item_id, artist.provider
625        )
626
627    async def _library_artist_tracks(self, artist: Artist) -> list[Track]:
628        """
629        Return the in-library tracks for the given artist (empty if it is not saved).
630
631        :param artist: The artist to resolve in-library tracks for.
632        """
633        if (library_artist := await self._resolve_library_artist(artist)) is None:
634            return []
635        return await self.mass.music.artists.tracks(library_artist.item_id, "library")
636
637    async def _provider_artist_tracks(self, artist: Artist) -> list[Track]:
638        """
639        Return all of the artist's tracks across its (streaming) providers.
640
641        :param artist: The artist to resolve provider tracks for.
642        """
643        unique_providers = self.mass.music.get_unique_providers()
644        tracks: list[Track] = []
645        for mapping in artist.provider_mappings:
646            if mapping.provider_instance not in unique_providers:
647                continue
648            tracks.extend(
649                await self.mass.music.artists.tracks(mapping.item_id, mapping.provider_instance)
650            )
651        return tracks
652
653    async def _resolve_media_items(
654        self,
655        media_item: MediaItemType | ItemMapping | BrowseFolder,
656        start_item: str | None = None,
657        userid: str | None = None,
658        queue_id: str | None = None,
659        sort_by: str | None = None,
660        start_from_beginning: bool = False,
661        keep_preceding_items: bool = False,
662    ) -> list[MediaItemType]:
663        """
664        Resolve/unwrap media items to enqueue.
665
666        :param media_item: The media item to resolve into playable items.
667        :param start_item: Optional item to start a playlist/album/genre from, or the chapter
668            to start an audiobook/podcast episode at.
669        :param userid: Optional user the playback is attributed to.
670        :param queue_id: Optional queue the playback is requested for.
671        :param sort_by: Optional sort key to order tracks by before applying start_item.
672        :param start_from_beginning: Ignore any saved resume position for a podcast episode.
673        :param keep_preceding_items: For a playlist/album, move the tracks before start_item
674            behind the rest instead of dropping them, so the full item is returned with
675            start_item first.
676        """
677        # resolve Itemmapping to full media item
678        if isinstance(media_item, ItemMapping):
679            if media_item.uri is None:
680                raise InvalidDataError("ItemMapping has no URI")
681            media_item = await self.mass.music.get_item_by_uri(media_item.uri)
682        if media_item.media_type == MediaType.PLAYLIST:
683            media_item = cast("Playlist", media_item)
684            playlist_tracks = await self.get_playlist_tracks(
685                media_item,
686                start_item,
687                sort_by=sort_by,
688                keep_preceding_items=keep_preceding_items,
689            )
690            self._mark_container_played(media_item, playlist_tracks, userid, queue_id)
691            return list(playlist_tracks)
692        if media_item.media_type == MediaType.ARTIST:
693            media_item = cast("Artist", media_item)
694            artist_items: list[Audiobook] | list[Track]
695            if media_item.artist_type in [ArtistType.AUTHOR, ArtistType.NARRATOR]:
696                artist_items = await self.get_author_narrator_audiobooks(media_item, userid)
697            else:
698                artist_items = await self.get_artist_tracks(media_item)
699            self._mark_container_played(media_item, artist_items, userid, queue_id)
700            return list(artist_items)
701        if media_item.media_type == MediaType.ALBUM:
702            media_item = cast("Album", media_item)
703            return list(
704                await self.get_album_tracks(
705                    media_item,
706                    start_item,
707                    sort_by=sort_by,
708                    keep_preceding_items=keep_preceding_items,
709                )
710            )
711        if media_item.media_type == MediaType.GENRE:
712            media_item = cast("Genre", media_item)
713            genre_tracks = await self.get_genre_tracks(media_item, start_item)
714            self._mark_container_played(media_item, genre_tracks, userid, queue_id)
715            return list(genre_tracks)
716        if media_item.media_type == MediaType.AUDIOBOOK:
717            media_item = cast("Audiobook", media_item)
718            # ensure we grab the correct/latest resume point info
719            media_item.resume_position_ms = await self.get_audiobook_resume_point(
720                media_item, start_item, userid=userid
721            )
722            return [media_item]
723        if media_item.media_type == MediaType.COLLECTION:
724            collection_item_media_type = get_collection_item_media_type_from_item_id(
725                media_item.item_id
726            )
727            if collection_item_media_type != MediaType.AUDIOBOOK:
728                self.logger.error("Collections are only available for audiobooks.")
729                return []
730            if TYPE_CHECKING:
731                assert isinstance(media_item, MediaCollection)
732            book: Audiobook | None = None
733            for item in media_item.items:
734                if TYPE_CHECKING:
735                    assert isinstance(item, Audiobook)
736                # enqueue the first not fully finished audiobook
737                fully_played, resume_position_ms = await self.mass.music.get_resume_position(
738                    item, userid=userid
739                )
740                if not fully_played:
741                    item.resume_position_ms = resume_position_ms
742                    book = item
743                    break
744            if book is None:
745                if len(media_item.items) > 0:
746                    return [media_item.items[0]]
747                return []
748            return [book]
749
750        if media_item.media_type == MediaType.PODCAST:
751            media_item = cast("Podcast", media_item)
752            episodes = await self.get_next_podcast_episodes(
753                media_item, start_item, userid=userid, start_from_beginning=start_from_beginning
754            )
755            self._mark_container_played(media_item, episodes, userid, queue_id)
756            return list(episodes)
757        if media_item.media_type == MediaType.PODCAST_EPISODE:
758            media_item = cast("PodcastEpisode", media_item)
759            return list(
760                await self.get_next_podcast_episodes(
761                    None, media_item, userid=userid, start_from_beginning=start_from_beginning
762                )
763            )
764        if media_item.media_type == MediaType.FOLDER:
765            media_item = cast("BrowseFolder", media_item)
766            return list(await self._get_folder_tracks(media_item))
767        # all other: single track or radio item
768        return [cast("MediaItemType", media_item)]
769
770    async def _get_folder_tracks(self, folder: BrowseFolder) -> list[Track]:
771        """Fetch (playable) tracks for given browse folder."""
772        self.logger.info(
773            "Fetching tracks to play for folder %s",
774            folder.name,
775        )
776        try:
777            folder_items = await self.mass.music.browse(folder.path)
778        except OSError as err:
779            # e.g. the (top-level) folder URI points at a path that no longer exists
780            raise MediaNotFoundError(f"Folder '{folder.path}' could not be found") from err
781        tracks: list[Track] = []
782        for item in folder_items:
783            if not item.is_playable:
784                continue
785            try:
786                # recursively fetch tracks from all media types
787                resolved = await self._resolve_media_items(item)
788            except MediaNotFoundError:
789                # best-effort: skip child items/subfolders that are empty or unreachable
790                # so a single bad entry does not abort playback of the whole folder
791                continue
792            tracks += [x for x in resolved if isinstance(x, Track)]
793
794        return tracks
795
796    def _mark_container_played(
797        self,
798        container: MediaItemType,
799        resolved_items: Sequence[MediaItemType],
800        userid: str | None,
801        queue_id: str | None,
802    ) -> None:
803        """
804        Credit a container the user asked to play with an explicit play.
805
806        Only credits when the container actually resolved to something, so an empty
807        playlist/artist/genre/podcast never lands in the play history.
808
809        :param container: The playlist, artist, genre or podcast that was asked for.
810        :param resolved_items: The items the container resolved to.
811        :param userid: Optional user the playback is attributed to.
812        :param queue_id: Optional queue the playback is requested for.
813        """
814        if not resolved_items:
815            return
816        self.mass.create_task(
817            self.mass.music.mark_item_played(
818                container, userid=userid, queue_id=queue_id, user_initiated=True
819            )
820        )
821