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