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