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