music-assistant-server

1"""Youtube Music support for MusicAssistant."""
2
3from __future__ import annotations
4
5import asyncio
6import importlib
7import logging
8import time
9from collections.abc import AsyncGenerator
10from contextlib import suppress
11from datetime import datetime
12from io import StringIO
13from typing import TYPE_CHECKING, Any, cast
14from urllib.parse import parse_qs, unquote, urlparse
15
16from aiohttp import ClientError
17from duration_parser import parse as parse_str_duration
18from music_assistant_models.enums import (
19    AlbumType,
20    ContentType,
21    ImageType,
22    MediaType,
23    ProviderFeature,
24    StreamType,
25)
26from music_assistant_models.errors import (
27    InvalidDataError,
28    LoginFailed,
29    MediaNotFoundError,
30    SetupFailedError,
31    UnplayableMediaError,
32)
33from music_assistant_models.media_items import (
34    Album,
35    Artist,
36    AudioFormat,
37    BrowseFolder,
38    ItemMapping,
39    MediaItemImage,
40    MediaItemType,
41    Playlist,
42    Podcast,
43    PodcastEpisode,
44    ProviderMapping,
45    RecommendationFolder,
46    SearchResults,
47    Track,
48    UniqueList,
49)
50from music_assistant_models.streamdetails import StreamDetails
51from ytmusicapi.constants import SUPPORTED_LANGUAGES
52from ytmusicapi.exceptions import YTMusicServerError
53from ytmusicapi.helpers import get_authorization, sapisid_from_cookie
54from ytmusicapi.parsers.podcasts import Description
55
56from music_assistant.constants import (
57    CONF_ENTRY_UNOFFICIAL_PROVIDER,
58    CONF_USERNAME,
59    VERBOSE_LOG_LEVEL,
60)
61from music_assistant.controllers.cache import use_cache
62from music_assistant.helpers.util import (
63    import_module_in_thread,
64    infer_album_type,
65    install_package,
66    parse_title_and_version,
67)
68from music_assistant.models.music_provider import MusicProvider
69from music_assistant.models.recommendation_payload import RecommendationPayloadMixin
70
71from .helpers import (
72    YTMSearchFilter,
73    add_remove_playlist_tracks,
74    convert_to_netscape,
75    determine_recommendation_icon,
76    get_album,
77    get_artist,
78    get_home,
79    get_library_albums,
80    get_library_artists,
81    get_library_playlists,
82    get_library_podcasts,
83    get_library_tracks,
84    get_playlist,
85    get_podcast,
86    get_podcast_episode,
87    get_song_radio_tracks,
88    get_track,
89    is_brand_account,
90    library_add_remove_album,
91    library_add_remove_artist,
92    library_add_remove_playlist,
93    search,
94)
95
96if TYPE_CHECKING:
97    from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
98    from music_assistant_models.provider import ProviderManifest
99
100    from music_assistant import MusicAssistant
101    from music_assistant.models import ProviderInstanceType
102
103
104CONF_COOKIE = "cookie"
105CONF_PO_TOKEN_SERVER_URL = "po_token_server_url"
106DEFAULT_PO_TOKEN_SERVER_URL = "http://127.0.0.1:4416"
107
108YTM_DOMAIN = "https://music.youtube.com"
109YTM_COOKIE_DOMAIN = ".youtube.com"
110YTM_BASE_URL = f"{YTM_DOMAIN}/youtubei/v1/"
111VARIOUS_ARTISTS_YTM_ID = "UCUTXlgdcKU5vfzFqHOWIvkA"
112# Playlist ID's are not unique across instances for lists like 'Liked videos', 'SuperMix' etc.
113# So we need to add a delimiter to make them unique
114YT_PLAYLIST_ID_DELIMITER = "🎵"
115PODCAST_EPISODE_SPLITTER = "|"
116YT_LIKED_SONGS_PLAYLIST_ID = "LM"
117YT_PERSONAL_PODCAST_PLAYLISTS = (
118    "SE",  # Episodes for later
119    "RDPN",  # New episodes
120)
121YT_PERSONAL_PLAYLISTS = (
122    YT_LIKED_SONGS_PLAYLIST_ID,  # Liked songs
123    "RDTMAK5uy_kset8DisdE7LSD4TNjEVvrKRTmG7a56sY",  # SuperMix
124    "RDTMAK5uy_nGQKSMIkpr4o9VI_2i56pkGliD6FQRo50",  # My Mix 1
125    "RDTMAK5uy_lz2owBgwWf1mjzyn_NbxzMViQzIg8IAIg",  # My Mix 2
126    "RDTMAK5uy_k5UUl0lmrrfrjMpsT0CoMpdcBz1ruAO1k",  # My Mix 3
127    "RDTMAK5uy_nTsa0Irmcu2li2-qHBoZxtrpG9HuC3k_Q",  # My Mix 4
128    "RDTMAK5uy_lfZhS7zmIcmUhsKtkWylKzc0EN0LW90-s",  # My Mix 5
129    "RDTMAK5uy_k78ni6Y4fyyl0r2eiKkBEICh9Q5wJdfXk",  # My Mix 6
130    "RDTMAK5uy_lfhhWWw9v71CPrR7MRMHgZzbH6Vku9iJc",  # My Mix 7
131    "RDTMAK5uy_n_5IN6hzAOwdCnM8D8rzrs3vDl12UcZpA",  # Discover Mix
132    "RDTMAK5uy_lr0LWzGrq6FU9GIxWvFHTRPQD2LHMqlFA",  # New Release Mix
133    "RDTMAK5uy_nilrsVWxrKskY0ZUpVZ3zpB0u4LwWTVJ4",  # Replay Mix
134    "RDTMAK5uy_mZtXeU08kxXJOUhL0ETdAuZTh1z7aAFAo",  # Archive Mix
135)
136DYNAMIC_PLAYLIST_TRACK_LIMIT = 300
137YTM_PREMIUM_CHECK_TRACK_ID = "dQw4w9WgXcQ"
138PACKAGES_TO_INSTALL = ("yt-dlp[default]", "bgutil-ytdlp-pot-provider")
139DEFAULT_STREAM_URL_EXPIRATION = 3600  # 1 hour
140
141SUPPORTED_FEATURES = {
142    ProviderFeature.LIBRARY_ARTISTS,
143    ProviderFeature.LIBRARY_ALBUMS,
144    ProviderFeature.LIBRARY_TRACKS,
145    ProviderFeature.LIBRARY_PLAYLISTS,
146    ProviderFeature.BROWSE,
147    ProviderFeature.SEARCH,
148    ProviderFeature.ARTIST_ALBUMS,
149    ProviderFeature.ARTIST_TOPTRACKS,
150    ProviderFeature.SIMILAR_TRACKS,
151    ProviderFeature.LIBRARY_PODCASTS,
152    ProviderFeature.RECOMMENDATIONS,
153}
154
155
156# TODO: fix disabled tests
157# ruff: noqa: PLW2901
158
159
160async def setup(
161    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
162) -> ProviderInstanceType:
163    """Initialize provider(instance) with given configuration."""
164    return YoutubeMusicProvider(mass, manifest, config, SUPPORTED_FEATURES)
165
166
167class YoutubeMusicProvider(RecommendationPayloadMixin, MusicProvider):
168    """Provider for Youtube Music."""
169
170    _headers: dict[str, str]
171    _context: dict[str, Any]
172    _cookies: dict[str, str]
173    _cipher = None
174    _yt_user: str | None
175    _cookie: str
176    _yt_dlp_module = None
177
178    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
179        """Return Config entries to configure this provider."""
180        return (CONF_ENTRY_UNOFFICIAL_PROVIDER,)
181
182    async def handle_async_init(self) -> None:
183        """Set up the YTMusic provider."""
184        logging.getLogger("yt_dlp").setLevel(self.logger.level + 10)
185        await self._install_packages()
186        self._cookie = str(self.get_setup_value(CONF_COOKIE))
187        self._po_token_server_url = (
188            self.get_setup_value(CONF_PO_TOKEN_SERVER_URL) or DEFAULT_PO_TOKEN_SERVER_URL
189        )
190        if not await self._verify_po_token_url():
191            raise LoginFailed(
192                "PO Token server URL is not reachable. "
193                "Make sure you have installed the YT Music PO Token Generator "
194                "and that it is running."
195            )
196        yt_username = str(self.get_setup_value(CONF_USERNAME))
197        self._yt_user = yt_username if is_brand_account(yt_username) else None
198        # yt-dlp needs a netscape formatted cookie
199        self._netscape_cookie = convert_to_netscape(self._cookie, YTM_COOKIE_DOMAIN)
200        self._initialize_headers()
201        self._initialize_context()
202        self._cookies = {"CONSENT": "YES+1"}
203        # get default language (that is supported by YTM)
204        mass_locale = self.mass.metadata.locale
205        for lang_code in SUPPORTED_LANGUAGES:
206            if lang_code in (mass_locale, mass_locale.split("_")[0]):
207                self.language = lang_code
208                break
209        else:
210            self.language = "en"
211        if not await self._user_has_ytm_premium():
212            raise LoginFailed("User does not have Youtube Music Premium")
213
214    @use_cache(3600 * 24 * 7)  # Cache for 7 days
215    async def search(
216        self, search_query: str, media_types: list[MediaType], limit: int = 5
217    ) -> SearchResults:
218        """
219        Perform search on musicprovider.
220
221        :param search_query: Search query.
222        :param media_types: A list of media_types to include. All types if None.
223        :param limit: Number of items to return in the search (per type).
224        """
225        parsed_results = SearchResults()
226        ytm_filter: YTMSearchFilter | None = None
227        if len(media_types) == 1:
228            # YTM does not support multiple searchtypes, falls back to all if no type given
229            if media_types[0] == MediaType.ARTIST:
230                ytm_filter = "artists"
231            if media_types[0] == MediaType.ALBUM:
232                ytm_filter = "albums"
233            if media_types[0] == MediaType.TRACK:
234                ytm_filter = "songs"
235            if media_types[0] == MediaType.PLAYLIST:
236                ytm_filter = "playlists"
237            if media_types[0] == MediaType.PODCAST:
238                ytm_filter = "podcasts"
239            if media_types[0] == MediaType.RADIO:
240                # bit of an edge case but still good to handle
241                return parsed_results
242        results = await search(
243            query=search_query, ytm_filter=ytm_filter, limit=limit, language=self.language
244        )
245        parsed_results = SearchResults()
246        artists: list[Artist | ItemMapping] = []
247        albums: list[Album | ItemMapping] = []
248        playlists: list[Playlist | ItemMapping] = []
249        tracks: list[Track | ItemMapping] = []
250        podcasts: list[Podcast | ItemMapping] = []
251        for result in results:
252            try:
253                if result["resultType"] == "artist" and MediaType.ARTIST in media_types:
254                    artists.append(self._parse_artist(result))
255                elif result["resultType"] == "album" and MediaType.ALBUM in media_types:
256                    albums.append(self._parse_album(result))
257                elif result["resultType"] == "playlist" and MediaType.PLAYLIST in media_types:
258                    playlists.append(self._parse_playlist(result))
259                elif result["resultType"] == "podcast" and MediaType.PODCAST in media_types:
260                    if podcast := self._parse_browse_podcast(result):
261                        podcasts.append(podcast)
262                elif (
263                    result["resultType"] in ("song", "video")
264                    and MediaType.TRACK in media_types
265                    and (track := self._parse_track(result))
266                ):
267                    tracks.append(track)
268            except InvalidDataError:
269                pass  # ignore invalid item
270        parsed_results.artists = artists
271        parsed_results.albums = albums
272        parsed_results.playlists = playlists
273        parsed_results.tracks = tracks
274        parsed_results.podcasts = podcasts
275        return parsed_results
276
277    async def get_library_artists(self) -> AsyncGenerator[Artist]:
278        """Retrieve all library artists from Youtube Music."""
279        artists_obj = await get_library_artists(
280            headers=self._headers, language=self.language, user=self._yt_user
281        )
282        for artist in artists_obj:
283            yield self._parse_artist(artist)
284
285    async def get_library_albums(self) -> AsyncGenerator[Album]:
286        """Retrieve all library albums from Youtube Music."""
287        albums_obj = await get_library_albums(
288            headers=self._headers, language=self.language, user=self._yt_user
289        )
290        for album in albums_obj:
291            yield self._parse_album(album, album["browseId"])
292
293    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
294        """Retrieve all library playlists from the provider."""
295        playlists_obj = await get_library_playlists(
296            headers=self._headers, language=self.language, user=self._yt_user
297        )
298        for playlist in playlists_obj:
299            playlist_id = playlist["id"]
300            if playlist_id in YT_PERSONAL_PODCAST_PLAYLISTS:
301                continue
302            yield self._parse_playlist(playlist)
303
304    async def get_library_tracks(self) -> AsyncGenerator[Track]:
305        """Retrieve library tracks from Youtube Music."""
306        tracks_obj = await get_library_tracks(
307            headers=self._headers, language=self.language, user=self._yt_user
308        )
309        for track in tracks_obj:
310            # Library tracks sometimes do not have a valid artist id
311            # In that case, call the API for track details based on track id
312            try:
313                yield self._parse_track(track)
314            except InvalidDataError:
315                full_track = await self.get_track(track["videoId"])
316                yield full_track
317
318    async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
319        """Retrieve the library podcasts from Youtube Music."""
320        podcasts_obj = await get_library_podcasts(
321            headers=self._headers, language=self.language, user=self._yt_user
322        )
323        for podcast in podcasts_obj:
324            podcast_id = podcast.get("podcastId")
325            if podcast_id in YT_PERSONAL_PODCAST_PLAYLISTS:
326                continue
327            yield self._parse_podcast(podcast)
328
329    @use_cache(3600 * 24 * 30)  # Cache for 30 days
330    async def get_album(self, prov_album_id: str) -> Album:
331        """Get full album details by id."""
332        if album_obj := await get_album(
333            headers=self._headers,
334            prov_album_id=prov_album_id,
335            language=self.language,
336            user=self._yt_user,
337        ):
338            return self._parse_album(album_obj=album_obj, album_id=prov_album_id)
339        msg = f"Item {prov_album_id} not found"
340        raise MediaNotFoundError(msg)
341
342    @use_cache(3600 * 24 * 30, allow_expired_cache=True)  # Cache for 30 days
343    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
344        """Get album tracks for given album id."""
345        album_obj = await get_album(
346            headers=self._headers,
347            prov_album_id=prov_album_id,
348            language=self.language,
349            user=self._yt_user,
350        )
351        if not album_obj.get("tracks"):
352            return []
353        tracks = []
354        for track_number, track_obj in enumerate(album_obj["tracks"], 1):
355            try:
356                track = self._parse_track(track_obj=track_obj, track_number=track_number)
357            except InvalidDataError:
358                continue
359            tracks.append(track)
360        return tracks
361
362    @use_cache(3600 * 24 * 30)  # Cache for 30 days
363    async def get_artist(self, prov_artist_id: str) -> Artist:
364        """Get full artist details by id."""
365        if artist_obj := await get_artist(
366            prov_artist_id=prov_artist_id, headers=self._headers, language=self.language
367        ):
368            return self._parse_artist(artist_obj=artist_obj)
369        msg = f"Item {prov_artist_id} not found"
370        raise MediaNotFoundError(msg)
371
372    @use_cache(3600 * 24 * 30)  # Cache for 30 days
373    async def get_track(self, prov_track_id: str) -> Track:
374        """Get full track details by id."""
375        if track_obj := await get_track(
376            prov_track_id=prov_track_id,
377            headers=self._headers,
378            language=self.language,
379        ):
380            return self._parse_track(track_obj)
381        msg = f"Item {prov_track_id} not found"
382        raise MediaNotFoundError(msg)
383
384    @use_cache(3600 * 24 * 7)  # Cache for 7 days
385    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
386        """Get full playlist details by id."""
387        # Grab the full playlist by default
388        limit = None
389        # Grab the playlist id from the full url in case of personal playlists
390        if YT_PLAYLIST_ID_DELIMITER in prov_playlist_id:
391            prov_playlist_id = prov_playlist_id.split(YT_PLAYLIST_ID_DELIMITER, maxsplit=1)[0]
392        if (
393            prov_playlist_id in YT_PERSONAL_PLAYLISTS
394            and prov_playlist_id != YT_LIKED_SONGS_PLAYLIST_ID
395        ):
396            # Personal playlists are dynamic and can result in endless tracks
397            # limit to avoid memory issues
398            limit = DYNAMIC_PLAYLIST_TRACK_LIMIT
399        if playlist_obj := await get_playlist(
400            prov_playlist_id=prov_playlist_id,
401            headers=self._headers,
402            language=self.language,
403            user=self._yt_user,
404            limit=limit,
405        ):
406            return self._parse_playlist(playlist_obj)
407        msg = f"Item {prov_playlist_id} not found"
408        raise MediaNotFoundError(msg)
409
410    @use_cache(3600 * 3, allow_expired_cache=True)  # Cache for 3 hours
411    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
412        """Return playlist tracks for the given provider playlist id."""
413        if page > 0:
414            # paging not supported, we always return the whole list at once
415            return []
416        # Grab the full playlist by default
417        limit = None
418        # Grab the playlist id from the full url in case of personal playlists
419        if YT_PLAYLIST_ID_DELIMITER in prov_playlist_id:
420            prov_playlist_id = prov_playlist_id.split(YT_PLAYLIST_ID_DELIMITER, maxsplit=1)[0]
421        if (
422            prov_playlist_id in YT_PERSONAL_PLAYLISTS
423            and prov_playlist_id != YT_LIKED_SONGS_PLAYLIST_ID
424        ):
425            # Personal playlists are dynamic and can result in endless tracks
426            # limit to avoid memory issues
427            limit = DYNAMIC_PLAYLIST_TRACK_LIMIT
428        # Add a try to prevent MA from stopping syncing whenever we fail a single playlist
429        try:
430            playlist_obj = await get_playlist(
431                prov_playlist_id=prov_playlist_id,
432                headers=self._headers,
433                user=self._yt_user,
434                limit=limit,
435            )
436        except KeyError as ke:
437            self.logger.warning("Could not load playlist: %s: %s", prov_playlist_id, ke)
438            return []
439        if "tracks" not in playlist_obj:
440            return []
441        result = []
442        # TODO: figure out how to handle paging in YTM
443        for index, track_obj in enumerate(playlist_obj["tracks"], 1):
444            if track_obj["isAvailable"]:
445                # Playlist tracks sometimes do not have a valid artist id
446                # In that case, call the API for track details based on track id
447                try:
448                    if track := self._parse_track(track_obj):
449                        track.position = index
450                        result.append(track)
451                except InvalidDataError:
452                    if track := await self.get_track(track_obj["videoId"]):
453                        track.position = index
454                        result.append(track)
455        # YTM doesn't seem to support paging so we ignore offset and limit
456        return result
457
458    @use_cache(3600 * 24 * 7, allow_expired_cache=True)  # Cache for 7 days
459    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
460        """Get a list of albums for the given artist."""
461        artist_obj = await get_artist(prov_artist_id=prov_artist_id, headers=self._headers)
462        if "albums" in artist_obj and "results" in artist_obj["albums"]:
463            albums = []
464            for album_obj in artist_obj["albums"]["results"]:
465                if "artists" not in album_obj:
466                    album_obj["artists"] = [
467                        {"id": artist_obj["channelId"], "name": artist_obj["name"]}
468                    ]
469                albums.append(self._parse_album(album_obj, album_obj["browseId"]))
470            return albums
471        return []
472
473    @use_cache(3600 * 24 * 7, allow_expired_cache=True)  # Cache for 7 days
474    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
475        """Get a list of 25 most popular tracks for the given artist."""
476        artist_obj = await get_artist(prov_artist_id=prov_artist_id, headers=self._headers)
477        if artist_obj.get("songs") and artist_obj["songs"].get("browseId"):
478            prov_playlist_id = artist_obj["songs"]["browseId"]
479            playlist_tracks = await self.get_playlist_tracks(prov_playlist_id)
480            return playlist_tracks[:25]
481        return []
482
483    @use_cache(3600 * 24 * 14)  # Cache for 14 days
484    async def get_podcast(self, prov_podcast_id: str) -> Podcast:
485        """Get the full details of a Podcast."""
486        podcast_obj = await get_podcast(prov_podcast_id, headers=self._headers)
487        return self._parse_podcast(podcast_obj)
488
489    async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
490        """Get all episodes from a podcast."""
491        podcast_obj = await get_podcast(prov_podcast_id, headers=self._headers)
492        podcast_obj["podcastId"] = prov_podcast_id
493        podcast = self._parse_podcast(podcast_obj)
494        for index, episode_obj in enumerate(podcast_obj.get("episodes", []), start=1):
495            episode = self._parse_podcast_episode(episode_obj, podcast)
496            ep_index = episode_obj.get("index") or index
497            episode.position = ep_index
498            yield episode
499
500    @use_cache(3600 * 3)  # Cache for 3 hours
501    async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
502        """Get a single Podcast Episode."""
503        podcast_id, episode_id = prov_episode_id.split(PODCAST_EPISODE_SPLITTER)
504        podcast = await self.get_podcast(podcast_id)
505        episode_obj = await get_podcast_episode(episode_id, headers=self._headers)
506        episode = self._parse_podcast_episode(episode_obj, podcast)
507        episode.position = 0
508        return episode
509
510    async def library_add(self, item: MediaItemType) -> bool:
511        """Add an item to the library."""
512        result = False
513        if item.media_type == MediaType.ARTIST:
514            result = await library_add_remove_artist(
515                headers=self._headers, prov_artist_id=item.item_id, add=True, user=self._yt_user
516            )
517        elif item.media_type == MediaType.ALBUM:
518            result = bool(
519                await library_add_remove_album(
520                    headers=self._headers, prov_item_id=item.item_id, add=True, user=self._yt_user
521                )
522            )
523        elif item.media_type == MediaType.PLAYLIST:
524            result = await library_add_remove_playlist(
525                headers=self._headers, prov_item_id=item.item_id, add=True, user=self._yt_user
526            )
527        elif item.media_type == MediaType.TRACK:
528            raise NotImplementedError
529        return result
530
531    async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
532        """Remove an item from the library."""
533        result = False
534        try:
535            if media_type == MediaType.ARTIST:
536                result = await library_add_remove_artist(
537                    headers=self._headers,
538                    prov_artist_id=prov_item_id,
539                    add=False,
540                    user=self._yt_user,
541                )
542            elif media_type == MediaType.ALBUM:
543                result = bool(
544                    await library_add_remove_album(
545                        headers=self._headers,
546                        prov_item_id=prov_item_id,
547                        add=False,
548                        user=self._yt_user,
549                    )
550                )
551            elif media_type == MediaType.PLAYLIST:
552                result = await library_add_remove_playlist(
553                    headers=self._headers, prov_item_id=prov_item_id, add=False, user=self._yt_user
554                )
555            elif media_type == MediaType.TRACK:
556                raise NotImplementedError
557        except YTMusicServerError as err:
558            # YTM raises if trying to remove an item that is not in the library
559            raise NotImplementedError(err) from err
560        return result
561
562    async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
563        """Add track(s) to playlist."""
564        # Grab the playlist id from the full url in case of personal playlists
565        if YT_PLAYLIST_ID_DELIMITER in prov_playlist_id:
566            prov_playlist_id = prov_playlist_id.split(YT_PLAYLIST_ID_DELIMITER, maxsplit=1)[0]
567        await add_remove_playlist_tracks(
568            headers=self._headers,
569            prov_playlist_id=prov_playlist_id,
570            prov_track_ids=prov_track_ids,
571            add=True,
572            user=self._yt_user,
573        )
574
575    async def remove_playlist_tracks(
576        self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
577    ) -> None:
578        """Remove track(s) from playlist."""
579        # Grab the full playlist by default
580        limit = None
581        # Grab the playlist id from the full url in case of personal playlists
582        if YT_PLAYLIST_ID_DELIMITER in prov_playlist_id:
583            prov_playlist_id = prov_playlist_id.split(YT_PLAYLIST_ID_DELIMITER, maxsplit=1)[0]
584        if (
585            prov_playlist_id in YT_PERSONAL_PLAYLISTS
586            and prov_playlist_id != YT_LIKED_SONGS_PLAYLIST_ID
587        ):
588            # Personal playlists are dynamic and can result in endless tracks
589            # limit to avoid memory issues
590            limit = DYNAMIC_PLAYLIST_TRACK_LIMIT
591        playlist_obj = await get_playlist(
592            prov_playlist_id=prov_playlist_id, headers=self._headers, limit=limit
593        )
594        if "tracks" not in playlist_obj:
595            return
596        tracks_to_delete = []
597        for index, track in enumerate(playlist_obj["tracks"]):
598            if index in positions_to_remove:
599                # YT needs both the videoId and the setVideoId in order to remove
600                # the track. Thus, we need to obtain the playlist details and
601                # grab the info from there.
602                tracks_to_delete.append(
603                    {"videoId": track["videoId"], "setVideoId": track["setVideoId"]}
604                )
605
606        await add_remove_playlist_tracks(
607            headers=self._headers,
608            prov_playlist_id=prov_playlist_id,
609            prov_track_ids=tracks_to_delete,
610            add=False,
611            user=self._yt_user,
612        )
613
614    @use_cache(3600 * 24, allow_expired_cache=True)  # Cache for 1 day
615    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
616        """Retrieve a dynamic list of tracks based on the provided item."""
617        result = await get_song_radio_tracks(
618            headers=self._headers, prov_item_id=prov_track_id, limit=limit, user=self._yt_user
619        )
620        if "tracks" in result:
621            tracks = []
622            for track in result["tracks"]:
623                # Playlist tracks sometimes do not have a valid artist id
624                # In that case, call the API for track details based on track id
625                try:
626                    track = self._parse_track(track)
627                    if track:
628                        tracks.append(track)
629                except InvalidDataError:
630                    if track := await self.get_track(track["videoId"]):
631                        tracks.append(track)
632            return tracks
633        return []
634
635    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
636        """Return the content details for the given track when it will be streamed."""
637        if media_type == MediaType.PODCAST_EPISODE:
638            item_id = item_id.split(PODCAST_EPISODE_SPLITTER)[1]
639        stream_format = await self._get_stream_format(item_id=item_id)
640        self.logger.debug("Found stream_format: %s for song %s", stream_format["format"], item_id)
641        url = stream_format["url"]
642        expiration = DEFAULT_STREAM_URL_EXPIRATION
643        if parsed := parse_qs(urlparse(url).query):
644            if expire_ts := parsed.get("expire", [None])[0]:
645                expiration = int(expire_ts) - int(time.time())
646        stream_details = StreamDetails(
647            provider=self.instance_id,
648            item_id=item_id,
649            audio_format=AudioFormat(
650                content_type=ContentType.try_parse(stream_format["audio_ext"]),
651            ),
652            stream_type=StreamType.HTTP,
653            path=url,
654            can_seek=True,
655            allow_seek=True,
656            expiration=expiration,
657        )
658        if (audio_channels := stream_format.get("audio_channels")) and str(
659            audio_channels
660        ).isdigit():
661            stream_details.audio_format.channels = int(audio_channels)
662        if asr := stream_format.get("asr"):
663            stream_details.audio_format.sample_rate = int(asr)
664        return stream_details
665
666    async def get_recommendations(self) -> list[RecommendationFolder]:
667        """Get this provider's available recommendation rows, without items."""
668        rows = await self._recommendation_rows_from_payload()
669        rows.append(
670            RecommendationFolder(
671                name="Mixed for you",
672                translation_key="mixed_for_you",
673                item_id=f"{self.instance_id}_mixed_for_you",
674                provider=self.instance_id,
675                icon="mdi:shuffle-variant",
676            )
677        )
678        return rows
679
680    async def get_recommendation_items(
681        self, item_id: str
682    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
683        """
684        Get the items for a single recommendation row.
685
686        :param item_id: The item_id of the row, as returned by get_recommendations.
687        """
688        if item_id == f"{self.instance_id}_mixed_for_you":
689            return (await self._get_mixed_for_you_folder()).items
690        return await self._recommendation_items_from_payload(item_id)
691
692    async def _fetch_recommendation_payload(self) -> list[RecommendationFolder]:
693        """Fetch the home feed and parse its sections into recommendation folders with items."""
694        recommendations = await get_home(self._headers, self.language, user=self._yt_user)
695
696        def _parse_sections() -> list[RecommendationFolder]:
697            # building model objects from the raw payload is CPU-bound; run off the event loop
698            folders: list[RecommendationFolder] = []
699            for section in recommendations:
700                folder = RecommendationFolder(
701                    name=section["title"],
702                    item_id=f"{self.instance_id}_{section['title']}",
703                    provider=self.instance_id,
704                    icon=determine_recommendation_icon(section["title"]),
705                )
706                for recommended_item in section.get("contents", []):
707                    if not recommended_item:
708                        continue  # yeah this seems to happen sometimes ?!
709                    if recommended_item.get("videoId"):
710                        # Probably a track
711                        try:
712                            track = self._parse_track(recommended_item)
713                            folder.items.append(track)
714                        except InvalidDataError:
715                            self.logger.debug(
716                                "Invalid track in recommendations: %s", recommended_item
717                            )
718                            continue
719                    elif recommended_item.get("playlistId"):
720                        # Probably a playlist
721                        recommended_item["id"] = recommended_item["playlistId"]
722                        del recommended_item["playlistId"]
723                        folder.items.append(self._parse_playlist(recommended_item))
724                    elif recommended_item.get("browseId"):
725                        if podcast := self._parse_browse_podcast(recommended_item):
726                            folder.items.append(podcast)
727                        else:
728                            # Probably an album
729                            folder.items.append(self._parse_album(recommended_item))
730                    elif recommended_item.get("subscribers"):
731                        # Probably artist
732                        folder.items.append(self._parse_album(recommended_item))
733                    elif recommended_item.get("videoType") == "MUSIC_VIDEO_TYPE_PODCAST_EPISODE":
734                        # Podcast episodes show up here without a videoId/browseId,
735                        # so there is no playable item to build from them
736                        self.logger.debug(
737                            "Skipping podcast episode in recommendation folder: %s",
738                            recommended_item.get("title"),
739                        )
740                        continue
741                    else:
742                        self.logger.warning(
743                            "Unknown item type in recommendation folder: %s",
744                            recommended_item,
745                        )
746                        continue
747                folders.append(folder)
748            return folders
749
750        return await asyncio.to_thread(_parse_sections)
751
752    @use_cache(3600 * 24, allow_expired_cache=True)  # Cache for 24 hours
753    async def _get_mixed_for_you_folder(self) -> RecommendationFolder:
754        """
755        Build the "Mixed for you" recommendation folder from the user's personal mixes.
756
757        :return: The folder, which has no items when no personal mixes are available.
758        """
759        mixed_for_you_folder = RecommendationFolder(
760            name="Mixed for you",
761            translation_key="mixed_for_you",
762            item_id=f"{self.instance_id}_mixed_for_you",
763            provider=self.instance_id,
764            icon="mdi:shuffle-variant",
765        )
766
767        async def _get_personal_playlist_preview(playlist_id: str) -> Playlist | None:
768            if playlist_id == YT_LIKED_SONGS_PLAYLIST_ID:
769                return None
770            try:
771                playlist_obj = await get_playlist(
772                    prov_playlist_id=playlist_id,
773                    headers=self._headers,
774                    language=self.language,
775                    user=self._yt_user,
776                    limit=5,  # limit fetched tracks here to reduce payload size
777                )
778                return self._parse_playlist(playlist_obj)
779            except MediaNotFoundError:
780                # personal playlist might not be available for all users, ignore if not found
781                return None
782            except Exception as err:
783                self.logger.debug("Failed to fetch personal mix %s: %s", playlist_id, err)
784                return None
785
786        playlist_previews = await asyncio.gather(
787            *(_get_personal_playlist_preview(playlist_id) for playlist_id in YT_PERSONAL_PLAYLISTS)
788        )
789        mixed_for_you_folder.items.extend(
790            preview for preview in playlist_previews if preview is not None
791        )
792        return mixed_for_you_folder
793
794    async def _post_data(self, endpoint: str, data: dict[str, str], **kwargs: Any) -> Any:
795        """Post data to the given endpoint."""
796        url = f"{YTM_BASE_URL}{endpoint}"
797        data.update(self._context)
798        async with self.mass.http_session.post(
799            url,
800            headers=self._headers,
801            json=data,
802            ssl=False,
803            cookies=self._cookies,
804        ) as response:
805            return await response.json()
806
807    async def _get_data(self, url: str, params: dict[str, Any] | None = None) -> str:
808        """Get data from the given URL."""
809        async with self.mass.http_session.get(
810            url, headers=self._headers, params=params, cookies=self._cookies
811        ) as response:
812            return await response.text()
813
814    def _initialize_headers(self) -> None:
815        """Initialize the headers to include in the requests."""
816        headers = {
817            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:72.0) Gecko/20100101 Firefox/72.0",
818            "Accept": "*/*",
819            "Accept-Language": "en-US,en;q=0.5",
820            "Content-Type": "application/json",
821            "X-Goog-AuthUser": "0",
822            "x-origin": YTM_DOMAIN,
823            "Cookie": self._cookie,
824        }
825        if "__Secure-3PAPISID" not in self._cookie:
826            raise LoginFailed(
827                "Invalid Cookie detected. Cookie is missing the __Secure-3PAPISID field. "
828                "Please ensure you are passing the correct cookie. "
829                "You can verify this by checking if the string "
830                "'__Secure-3PAPISID' is present in the cookie string."
831            )
832        sapisid = sapisid_from_cookie(self._cookie)
833        headers["Authorization"] = get_authorization(sapisid + " " + YTM_DOMAIN)
834        self._headers = headers
835
836    def _initialize_context(self) -> None:
837        """Initialize the context to use in requests."""
838        self._context = {
839            "context": {
840                "client": {"clientName": "WEB_REMIX", "clientVersion": "0.1"},
841                "user": {},
842            }
843        }
844
845    def _parse_album(self, album_obj: dict[str, Any], album_id: str | None = None) -> Album:
846        """Parse a YT Album response to an Album model object."""
847        album_id = album_id or album_obj.get("id") or album_obj.get("browseId")
848
849        if not album_id:
850            raise InvalidDataError("Album ID is required but not found")
851
852        if "title" in album_obj:
853            name, version = parse_title_and_version(album_obj["title"])
854        elif "name" in album_obj:
855            name, version = parse_title_and_version(album_obj["name"])
856        else:
857            name, version = "", ""
858        album = Album(
859            item_id=album_id,
860            name=name,
861            version=version,
862            provider=self.instance_id,
863            provider_mappings={
864                ProviderMapping(
865                    item_id=str(album_id),
866                    provider_domain=self.domain,
867                    provider_instance=self.instance_id,
868                    url=f"{YTM_DOMAIN}/playlist?list={album_obj.get('audioPlaylistId')}",
869                )
870            },
871            favorite=album_obj.get("likeStatus", "INDIFFERENT") == "LIKE",
872        )
873        if album_obj.get("year") and album_obj["year"].isdigit():
874            album.year = album_obj["year"]
875        if "thumbnails" in album_obj:
876            album.metadata.images = UniqueList(self._parse_thumbnails(album_obj["thumbnails"]))
877        if description := album_obj.get("description"):
878            album.metadata.description = unquote(description)
879        if "isExplicit" in album_obj:
880            album.metadata.explicit = album_obj["isExplicit"]
881        if "artists" in album_obj:
882            album.artists = UniqueList(
883                [
884                    self._get_artist_item_mapping(artist)
885                    for artist in album_obj["artists"]
886                    if artist.get("id")
887                    or artist.get("channelId")
888                    or artist.get("name") == "Various Artists"
889                ]
890            )
891        if "type" in album_obj:
892            if album_obj["type"] == "Single":
893                album_type = AlbumType.SINGLE
894            elif album_obj["type"] == "EP":
895                album_type = AlbumType.EP
896            elif album_obj["type"] == "Album":
897                album_type = AlbumType.ALBUM
898            else:
899                album_type = AlbumType.UNKNOWN
900            album.album_type = album_type
901
902        # Try inference - override if it finds something more specific
903        inferred_type = infer_album_type(name, version)
904        if inferred_type in (AlbumType.SOUNDTRACK, AlbumType.LIVE):
905            album.album_type = inferred_type
906
907        return album
908
909    def _parse_artist(self, artist_obj: dict[str, Any]) -> Artist:
910        """Parse a YT Artist response to Artist model object."""
911        artist_id = None
912        if "channelId" in artist_obj:
913            artist_id = artist_obj["channelId"]
914        elif artist_obj.get("id"):
915            artist_id = artist_obj["id"]
916        elif artist_obj["name"] == "Various Artists":
917            artist_id = VARIOUS_ARTISTS_YTM_ID
918        if not artist_id:
919            msg = "Artist does not have a valid ID"
920            raise InvalidDataError(msg)
921        artist = Artist(
922            item_id=artist_id,
923            name=artist_obj["name"],
924            provider=self.instance_id,
925            provider_mappings={
926                ProviderMapping(
927                    item_id=str(artist_id),
928                    provider_domain=self.domain,
929                    provider_instance=self.instance_id,
930                    url=f"{YTM_DOMAIN}/channel/{artist_id}",
931                )
932            },
933            favorite=artist_obj.get("likeStatus", "INDIFFERENT") == "LIKE",
934        )
935        if "description" in artist_obj:
936            artist.metadata.description = artist_obj["description"]
937        if artist_obj.get("thumbnails"):
938            artist.metadata.images = self._parse_thumbnails(artist_obj["thumbnails"])
939        return artist
940
941    def _parse_playlist(self, playlist_obj: dict[str, Any]) -> Playlist:
942        """Parse a YT Playlist response to a Playlist object."""
943        raw_playlist_id = playlist_obj["id"]
944        playlist_id = raw_playlist_id
945        playlist_name = playlist_obj["title"]
946        is_editable = playlist_obj.get("privacy", "") == "PRIVATE"
947        # Playlist ID's are not unique across instances for lists like 'Likes', 'Supermix', etc.
948        # So suffix with the instance id to make them unique
949        if playlist_id in YT_PERSONAL_PLAYLISTS:
950            playlist_id = f"{playlist_id}{YT_PLAYLIST_ID_DELIMITER}{self.instance_id}"
951            playlist_name = f"{playlist_name} ({self.name})"
952        playlist = Playlist(
953            item_id=playlist_id,
954            provider=self.instance_id,
955            name=playlist_name,
956            provider_mappings={
957                ProviderMapping(
958                    item_id=playlist_id,
959                    provider_domain=self.domain,
960                    provider_instance=self.instance_id,
961                    url=f"{YTM_DOMAIN}/playlist?list={raw_playlist_id}",
962                    is_unique=is_editable,  # user-owned playlists are unique
963                )
964            },
965            is_editable=is_editable,
966            favorite=playlist_obj.get("likeStatus", "INDIFFERENT") == "LIKE",
967        )
968        if "description" in playlist_obj:
969            playlist.metadata.description = playlist_obj["description"]
970        if playlist_obj.get("thumbnails"):
971            playlist.metadata.images = self._parse_thumbnails(playlist_obj["thumbnails"])
972
973        if authors := playlist_obj.get("author"):
974            if isinstance(authors, str):
975                playlist.owner = authors
976            elif isinstance(authors, list):
977                playlist.owner = authors[0]["name"]
978            else:
979                playlist.owner = authors["name"]
980        else:
981            playlist.owner = self.name
982        return playlist
983
984    def _parse_track(self, track_obj: dict[str, Any], track_number: int = 0) -> Track:
985        """Parse a YT Track response to a Track model object."""
986        if not track_obj.get("videoId"):
987            msg = "Track is missing videoId"
988            raise InvalidDataError(msg)
989        track_id = str(track_obj["videoId"])
990        name, version = parse_title_and_version(track_obj["title"])
991        track = Track(
992            item_id=track_id,
993            provider=self.instance_id,
994            name=name,
995            version=version,
996            provider_mappings={
997                ProviderMapping(
998                    item_id=track_id,
999                    provider_domain=self.domain,
1000                    provider_instance=self.instance_id,
1001                    available=track_obj.get("isAvailable", True),
1002                    url=f"{YTM_DOMAIN}/watch?v={track_id}",
1003                    audio_format=AudioFormat(
1004                        content_type=ContentType.M4A,
1005                    ),
1006                )
1007            },
1008            favorite=track_obj.get("likeStatus", "INDIFFERENT") == "LIKE",
1009            # Disc info is not available in YTM, assume a single disc
1010            disc_number=1,
1011            # Track number is "sometimes" available in the track object, otherwise approach
1012            # by counting album tracks when fetching full album details
1013            track_number=track_obj.get("trackNumber") or track_number or 0,
1014        )
1015
1016        if track_obj.get("artists"):
1017            track.artists = UniqueList(
1018                self._get_artist_item_mapping(artist)
1019                for artist in track_obj["artists"]
1020                if artist.get("id")
1021                or artist.get("channelId")
1022                or artist.get("name") == "Various Artists"
1023            )
1024        # guard that track has valid artists
1025        if not track.artists:
1026            msg = "Track is missing artists"
1027            raise InvalidDataError(msg)
1028        if track_obj.get("thumbnails"):
1029            track.metadata.images = self._parse_thumbnails(track_obj["thumbnails"])
1030        if (
1031            track_obj.get("album")
1032            and isinstance(track_obj.get("album"), dict)
1033            and track_obj["album"].get("id")
1034        ):
1035            album = track_obj["album"]
1036            track.album = self._get_item_mapping(MediaType.ALBUM, album["id"], album["name"])
1037        if "isExplicit" in track_obj:
1038            track.metadata.explicit = track_obj["isExplicit"]
1039        if "duration" in track_obj and str(track_obj["duration"]).isdigit():
1040            track.duration = int(track_obj["duration"])
1041        elif "duration_seconds" in track_obj and str(track_obj["duration_seconds"]).isdigit():
1042            track.duration = int(track_obj["duration_seconds"])
1043        return track
1044
1045    def _parse_podcast(self, podcast_obj: dict[str, Any]) -> Podcast:
1046        """Parse a YTM Podcast into a MA Podcast."""
1047        podcast = Podcast(
1048            item_id=podcast_obj["podcastId"],
1049            name=podcast_obj["title"],
1050            provider=self.instance_id,
1051            provider_mappings={
1052                ProviderMapping(
1053                    item_id=podcast_obj["podcastId"],
1054                    provider_domain=self.domain,
1055                    provider_instance=self.instance_id,
1056                )
1057            },
1058        )
1059        if description := podcast_obj.get("description"):
1060            podcast.metadata.description = description
1061        if author := podcast_obj.get("author") or podcast_obj.get("channel"):
1062            podcast.publisher = author["name"]
1063        if thumbnails := podcast_obj.get("thumbnails"):
1064            podcast.metadata.images = self._parse_thumbnails(thumbnails)
1065        return podcast
1066
1067    def _parse_browse_podcast(self, item_obj: dict[str, Any]) -> Podcast | None:
1068        """Parse a podcast from a browse or search item."""
1069        browse_id: str = item_obj.get("browseId") or ""
1070        if not browse_id.startswith("MPSP"):
1071            return None
1072        podcast_obj = dict(item_obj)
1073        # Match the unprefixed IDs used by library podcasts.
1074        podcast_obj["podcastId"] = item_obj.get("podcastId") or browse_id.removeprefix("MPSP")
1075        return self._parse_podcast(podcast_obj)
1076
1077    def _parse_podcast_episode(
1078        self, episode_obj: dict[str, Any], podcast: Podcast
1079    ) -> PodcastEpisode:
1080        """Parse a raw episode into a PodcastEpisode."""
1081        episode_id = episode_obj.get("videoId")
1082        if not episode_id:
1083            msg = "Podcast episode is missing videoId"
1084            raise InvalidDataError(msg)
1085        item_id = f"{podcast.item_id}{PODCAST_EPISODE_SPLITTER}{episode_id}"
1086        episode = PodcastEpisode(
1087            item_id=item_id,
1088            provider=self.instance_id,
1089            name=episode_obj.get("title", ""),
1090            position=0,
1091            podcast=podcast,
1092            provider_mappings={
1093                ProviderMapping(
1094                    item_id=item_id,
1095                    provider_domain=self.domain,
1096                    provider_instance=self.instance_id,
1097                    audio_format=AudioFormat(
1098                        content_type=ContentType.M4A,
1099                    ),
1100                    url=f"{YTM_DOMAIN}/watch?v={episode_id}",
1101                )
1102            },
1103        )
1104        if duration := episode_obj.get("duration"):
1105            duration_sec = parse_str_duration(duration)
1106            episode.duration = int(duration_sec)
1107        if description := episode_obj.get("description"):
1108            # the single episode endpoint returns a Description object instead of a string
1109            episode.metadata.description = (
1110                description.text if isinstance(description, Description) else description
1111            )
1112        if thumbnails := episode_obj.get("thumbnails"):
1113            episode.metadata.images = self._parse_thumbnails(thumbnails)
1114        if release_date := episode_obj.get("date"):
1115            with suppress(ValueError):
1116                episode.metadata.release_date = datetime.fromisoformat(release_date)
1117        return episode
1118
1119    async def _get_stream_format(self, item_id: str) -> dict[str, Any]:
1120        """Figure out the stream URL to use and return the highest quality."""
1121
1122        def _extract_best_stream_url_format() -> dict[str, Any]:
1123            if self._yt_dlp_module is None:
1124                self._yt_dlp_module = importlib.import_module("yt_dlp")
1125            yt_dlp = self._yt_dlp_module
1126            url = f"{YTM_DOMAIN}/watch?v={item_id}"
1127            ydl_opts = {
1128                "quiet": self.logger.level > logging.DEBUG,
1129                "verbose": self.logger.level == VERBOSE_LOG_LEVEL,
1130                "cookiefile": StringIO(self._netscape_cookie),
1131                # This enforces a player client and skips unnecessary scraping to increase speed
1132                "extractor_args": {
1133                    "youtubepot-bgutilhttp": {
1134                        "base_url": [self._po_token_server_url],
1135                    },
1136                    "youtube": {
1137                        "skip": ["translated_subs", "dash"],
1138                        "player_client": ["web_music"],
1139                    },
1140                },
1141            }
1142            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
1143                try:
1144                    info = ydl.extract_info(url, download=False)
1145                except yt_dlp.utils.DownloadError as err:
1146                    raise UnplayableMediaError(err) from err
1147                format_selector = ydl.build_format_selector("m4a/bestaudio")
1148                stream_format: dict[str, Any] | None = next(
1149                    format_selector({"formats": info["formats"]}), None
1150                )
1151                if not stream_format:
1152                    raise UnplayableMediaError("No stream formats found")
1153                return stream_format
1154
1155        return await asyncio.to_thread(_extract_best_stream_url_format)
1156
1157    def _get_item_mapping(self, media_type: MediaType, key: str, name: str) -> ItemMapping:
1158        return ItemMapping(
1159            media_type=media_type,
1160            item_id=key,
1161            provider=self.instance_id,
1162            name=name,
1163        )
1164
1165    def _get_artist_item_mapping(self, artist_obj: dict[str, Any]) -> ItemMapping:
1166        artist_id = artist_obj.get("id") or artist_obj.get("channelId")
1167        if not artist_id and artist_obj["name"] == "Various Artists":
1168            artist_id = VARIOUS_ARTISTS_YTM_ID
1169        # callers only pass artists with a valid id
1170        return self._get_item_mapping(
1171            MediaType.ARTIST, cast("str", artist_id), artist_obj.get("name", "")
1172        )
1173
1174    async def _verify_po_token_url(self) -> bool:
1175        """Ping the PO Token server and verify the response."""
1176        url = f"{self._po_token_server_url}/ping"
1177        try:
1178            async with self.mass.http_session.get(url) as response:
1179                response.raise_for_status()
1180                self.logger.debug("PO Token server responded with %s", response.status)
1181                return response.status == 200
1182        except (ClientError, TimeoutError) as err:
1183            self.logger.debug("PO Token server ping failed: %s", err)
1184            return False
1185
1186    async def _user_has_ytm_premium(self) -> bool:
1187        """Check if the user has Youtube Music Premium."""
1188        stream_format = await self._get_stream_format(YTM_PREMIUM_CHECK_TRACK_ID)
1189        # Only premium users can stream the HQ stream of this song
1190        format_id: str = stream_format["format_id"]
1191        return format_id == "141"
1192
1193    def _parse_thumbnails(self, thumbnails_obj: list[dict[str, Any]]) -> UniqueList[MediaItemImage]:
1194        """Parse and YTM thumbnails to MediaItemImage."""
1195        result: UniqueList[MediaItemImage] = UniqueList()
1196        processed_images = set()
1197        for img in sorted(thumbnails_obj, key=lambda w: w.get("width", 0), reverse=True):
1198            url: str = img["url"]
1199            url_base = url.split("=w", maxsplit=1)[0]
1200            width: int = img["width"]
1201            height: int = img["height"]
1202            image_ratio: float = width / height
1203            image_type = (
1204                ImageType.LANDSCAPE
1205                if "maxresdefault" in url or image_ratio > 2.0
1206                else ImageType.THUMB
1207            )
1208            if "=w" not in url and width < 500:
1209                continue
1210            # if the size is in the url, we can actually request a higher thumb
1211            if "=w" in url and width < 600:
1212                url = f"{url_base}=w600-h600-p"
1213                image_type = ImageType.THUMB
1214            if (url_base, image_type) in processed_images:
1215                continue
1216            processed_images.add((url_base, image_type))
1217            result.append(
1218                MediaItemImage(
1219                    type=image_type,
1220                    path=url,
1221                    provider=self.instance_id,
1222                    remotely_accessible=True,
1223                )
1224            )
1225        return result
1226
1227    async def _install_packages(self) -> None:
1228        """Install frequently changing packages dynamically."""
1229        # NOTE: Google breaks things quite often which requires us to update
1230        # some packages very frequently. Installing them dynamically prevents
1231        # us from having to update MA to ensure this provider works.
1232        for package_name in PACKAGES_TO_INSTALL:
1233            await install_package(package_name)
1234        # verify if the yt_dlp package is usable
1235        try:
1236            await import_module_in_thread("yt_dlp")
1237        except ImportError:
1238            raise SetupFailedError("Package yt_dlp failed to install")
1239