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        for index, episode_obj in enumerate(podcast_obj.get("episodes", []), start=1):
517            episode = self._parse_podcast_episode(episode_obj, podcast)
518            ep_index = episode_obj.get("index") or index
519            episode.position = ep_index
520            yield episode
521
522    @use_cache(3600 * 3)  # Cache for 3 hours
523    async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
524        """Get a single Podcast Episode."""
525        podcast_id, episode_id = prov_episode_id.split(PODCAST_EPISODE_SPLITTER)
526        podcast = await self.get_podcast(podcast_id)
527        episode_obj = await get_podcast_episode(episode_id, headers=self._headers)
528        episode = self._parse_podcast_episode(episode_obj, podcast)
529        episode.position = 0
530        return episode
531
532    async def library_add(self, item: MediaItemType) -> bool:
533        """Add an item to the library."""
534        result = False
535        if item.media_type == MediaType.ARTIST:
536            result = await library_add_remove_artist(
537                headers=self._headers, prov_artist_id=item.item_id, add=True, user=self._yt_user
538            )
539        elif item.media_type == MediaType.ALBUM:
540            result = bool(
541                await library_add_remove_album(
542                    headers=self._headers, prov_item_id=item.item_id, add=True, user=self._yt_user
543                )
544            )
545        elif item.media_type == MediaType.PLAYLIST:
546            result = await library_add_remove_playlist(
547                headers=self._headers, prov_item_id=item.item_id, add=True, user=self._yt_user
548            )
549        elif item.media_type == MediaType.TRACK:
550            raise NotImplementedError
551        return result
552
553    async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
554        """Remove an item from the library."""
555        result = False
556        try:
557            if media_type == MediaType.ARTIST:
558                result = await library_add_remove_artist(
559                    headers=self._headers,
560                    prov_artist_id=prov_item_id,
561                    add=False,
562                    user=self._yt_user,
563                )
564            elif media_type == MediaType.ALBUM:
565                result = bool(
566                    await library_add_remove_album(
567                        headers=self._headers,
568                        prov_item_id=prov_item_id,
569                        add=False,
570                        user=self._yt_user,
571                    )
572                )
573            elif media_type == MediaType.PLAYLIST:
574                result = await library_add_remove_playlist(
575                    headers=self._headers, prov_item_id=prov_item_id, add=False, user=self._yt_user
576                )
577            elif media_type == MediaType.TRACK:
578                raise NotImplementedError
579        except YTMusicServerError as err:
580            # YTM raises if trying to remove an item that is not in the library
581            raise NotImplementedError(err) from err
582        return result
583
584    async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
585        """Add track(s) to playlist."""
586        # Grab the playlist id from the full url in case of personal playlists
587        if YT_PLAYLIST_ID_DELIMITER in prov_playlist_id:
588            prov_playlist_id = prov_playlist_id.split(YT_PLAYLIST_ID_DELIMITER, maxsplit=1)[0]
589        await add_remove_playlist_tracks(
590            headers=self._headers,
591            prov_playlist_id=prov_playlist_id,
592            prov_track_ids=prov_track_ids,
593            add=True,
594            user=self._yt_user,
595        )
596
597    async def remove_playlist_tracks(
598        self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
599    ) -> None:
600        """Remove track(s) from playlist."""
601        # Grab the full playlist by default
602        limit = None
603        # Grab the playlist id from the full url in case of personal playlists
604        if YT_PLAYLIST_ID_DELIMITER in prov_playlist_id:
605            prov_playlist_id = prov_playlist_id.split(YT_PLAYLIST_ID_DELIMITER, maxsplit=1)[0]
606        if (
607            prov_playlist_id in YT_PERSONAL_PLAYLISTS
608            and prov_playlist_id != YT_LIKED_SONGS_PLAYLIST_ID
609        ):
610            # Personal playlists are dynamic and can result in endless tracks
611            # limit to avoid memory issues
612            limit = DYNAMIC_PLAYLIST_TRACK_LIMIT
613        playlist_obj = await get_playlist(
614            prov_playlist_id=prov_playlist_id, headers=self._headers, limit=limit
615        )
616        if "tracks" not in playlist_obj:
617            return
618        tracks_to_delete = []
619        for index, track in enumerate(playlist_obj["tracks"]):
620            if index in positions_to_remove:
621                # YT needs both the videoId and the setVideoId in order to remove
622                # the track. Thus, we need to obtain the playlist details and
623                # grab the info from there.
624                tracks_to_delete.append(
625                    {"videoId": track["videoId"], "setVideoId": track["setVideoId"]}
626                )
627
628        await add_remove_playlist_tracks(
629            headers=self._headers,
630            prov_playlist_id=prov_playlist_id,
631            prov_track_ids=tracks_to_delete,
632            add=False,
633            user=self._yt_user,
634        )
635
636    @use_cache(3600 * 24, allow_expired_cache=True)  # Cache for 1 day
637    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
638        """Retrieve a dynamic list of tracks based on the provided item."""
639        result = await get_song_radio_tracks(
640            headers=self._headers, prov_item_id=prov_track_id, limit=limit, user=self._yt_user
641        )
642        if "tracks" in result:
643            tracks = []
644            for track in result["tracks"]:
645                # Playlist tracks sometimes do not have a valid artist id
646                # In that case, call the API for track details based on track id
647                try:
648                    track = self._parse_track(track)
649                    if track:
650                        tracks.append(track)
651                except InvalidDataError:
652                    if track := await self.get_track(track["videoId"]):
653                        tracks.append(track)
654            return tracks
655        return []
656
657    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
658        """Return the content details for the given track when it will be streamed."""
659        if media_type == MediaType.PODCAST_EPISODE:
660            item_id = item_id.split(PODCAST_EPISODE_SPLITTER)[1]
661        stream_format = await self._get_stream_format(item_id=item_id)
662        self.logger.debug("Found stream_format: %s for song %s", stream_format["format"], item_id)
663        url = stream_format["url"]
664        expiration = DEFAULT_STREAM_URL_EXPIRATION
665        if parsed := parse_qs(urlparse(url).query):
666            if expire_ts := parsed.get("expire", [None])[0]:
667                expiration = int(expire_ts) - int(time.time())
668        stream_details = StreamDetails(
669            provider=self.instance_id,
670            item_id=item_id,
671            audio_format=AudioFormat(
672                content_type=ContentType.try_parse(stream_format["audio_ext"]),
673            ),
674            stream_type=StreamType.HTTP,
675            path=url,
676            can_seek=True,
677            allow_seek=True,
678            expiration=expiration,
679        )
680        if (audio_channels := stream_format.get("audio_channels")) and str(
681            audio_channels
682        ).isdigit():
683            stream_details.audio_format.channels = int(audio_channels)
684        if asr := stream_format.get("asr"):
685            stream_details.audio_format.sample_rate = int(asr)
686        return stream_details
687
688    async def get_recommendations(self) -> list[RecommendationFolder]:
689        """Get this provider's available recommendation rows, without items."""
690        rows = await self._recommendation_rows_from_payload()
691        rows.append(
692            RecommendationFolder(
693                name="Mixed for you",
694                translation_key="mixed_for_you",
695                item_id=f"{self.instance_id}_mixed_for_you",
696                provider=self.instance_id,
697                icon="mdi:shuffle-variant",
698            )
699        )
700        return rows
701
702    async def get_recommendation_items(
703        self, item_id: str
704    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
705        """
706        Get the items for a single recommendation row.
707
708        :param item_id: The item_id of the row, as returned by get_recommendations.
709        """
710        if item_id == f"{self.instance_id}_mixed_for_you":
711            return (await self._get_mixed_for_you_folder()).items
712        return await self._recommendation_items_from_payload(item_id)
713
714    async def _fetch_recommendation_payload(self) -> list[RecommendationFolder]:
715        """Fetch the home feed and parse its sections into recommendation folders with items."""
716        recommendations = await get_home(self._headers, self.language, user=self._yt_user)
717
718        def _parse_sections() -> list[RecommendationFolder]:
719            # building model objects from the raw payload is CPU-bound; run off the event loop
720            folders: list[RecommendationFolder] = []
721            for section in recommendations:
722                folder = RecommendationFolder(
723                    name=section["title"],
724                    item_id=f"{self.instance_id}_{section['title']}",
725                    provider=self.instance_id,
726                    icon=determine_recommendation_icon(section["title"]),
727                )
728                for recommended_item in section.get("contents", []):
729                    if not recommended_item:
730                        continue  # yeah this seems to happen sometimes ?!
731                    if recommended_item.get("videoId"):
732                        # Probably a track
733                        try:
734                            track = self._parse_track(recommended_item)
735                            folder.items.append(track)
736                        except InvalidDataError:
737                            self.logger.debug(
738                                "Invalid track in recommendations: %s", recommended_item
739                            )
740                            continue
741                    elif recommended_item.get("playlistId"):
742                        # Probably a playlist
743                        recommended_item["id"] = recommended_item["playlistId"]
744                        del recommended_item["playlistId"]
745                        folder.items.append(self._parse_playlist(recommended_item))
746                    elif recommended_item.get("browseId"):
747                        if podcast := self._parse_browse_podcast(recommended_item):
748                            folder.items.append(podcast)
749                        else:
750                            # Probably an album
751                            folder.items.append(self._parse_album(recommended_item))
752                    elif recommended_item.get("subscribers"):
753                        # Probably artist
754                        folder.items.append(self._parse_album(recommended_item))
755                    elif recommended_item.get("videoType") == "MUSIC_VIDEO_TYPE_PODCAST_EPISODE":
756                        # Podcast episodes show up here without a videoId/browseId,
757                        # so there is no playable item to build from them
758                        self.logger.debug(
759                            "Skipping podcast episode in recommendation folder: %s",
760                            recommended_item.get("title"),
761                        )
762                        continue
763                    else:
764                        self.logger.warning(
765                            "Unknown item type in recommendation folder: %s",
766                            recommended_item,
767                        )
768                        continue
769                folders.append(folder)
770            return folders
771
772        return await asyncio.to_thread(_parse_sections)
773
774    @use_cache(3600 * 24, allow_expired_cache=True)  # Cache for 24 hours
775    async def _get_mixed_for_you_folder(self) -> RecommendationFolder:
776        """
777        Build the "Mixed for you" recommendation folder from the user's personal mixes.
778
779        :return: The folder, which has no items when no personal mixes are available.
780        """
781        mixed_for_you_folder = RecommendationFolder(
782            name="Mixed for you",
783            translation_key="mixed_for_you",
784            item_id=f"{self.instance_id}_mixed_for_you",
785            provider=self.instance_id,
786            icon="mdi:shuffle-variant",
787        )
788
789        async def _get_personal_playlist_preview(playlist_id: str) -> Playlist | None:
790            if playlist_id == YT_LIKED_SONGS_PLAYLIST_ID:
791                return None
792            try:
793                playlist_obj = await get_playlist(
794                    prov_playlist_id=playlist_id,
795                    headers=self._headers,
796                    language=self.language,
797                    user=self._yt_user,
798                    limit=5,  # limit fetched tracks here to reduce payload size
799                )
800                return self._parse_playlist(playlist_obj)
801            except MediaNotFoundError:
802                # personal playlist might not be available for all users, ignore if not found
803                return None
804            except Exception as err:
805                self.logger.debug("Failed to fetch personal mix %s: %s", playlist_id, err)
806                return None
807
808        playlist_previews = await asyncio.gather(
809            *(_get_personal_playlist_preview(playlist_id) for playlist_id in YT_PERSONAL_PLAYLISTS)
810        )
811        mixed_for_you_folder.items.extend(
812            preview for preview in playlist_previews if preview is not None
813        )
814        return mixed_for_you_folder
815
816    async def _post_data(self, endpoint: str, data: dict[str, str], **kwargs: Any) -> Any:
817        """Post data to the given endpoint."""
818        url = f"{YTM_BASE_URL}{endpoint}"
819        data.update(self._context)
820        async with self.mass.http_session.post(
821            url,
822            headers=self._headers,
823            json=data,
824            ssl=False,
825            cookies=self._cookies,
826        ) as response:
827            return await response.json()
828
829    async def _get_data(self, url: str, params: dict[str, Any] | None = None) -> str:
830        """Get data from the given URL."""
831        async with self.mass.http_session.get(
832            url, headers=self._headers, params=params, cookies=self._cookies
833        ) as response:
834            return await response.text()
835
836    def _initialize_headers(self) -> None:
837        """Initialize the headers to include in the requests."""
838        headers = {
839            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:72.0) Gecko/20100101 Firefox/72.0",
840            "Accept": "*/*",
841            "Accept-Language": "en-US,en;q=0.5",
842            "Content-Type": "application/json",
843            "X-Goog-AuthUser": "0",
844            "x-origin": YTM_DOMAIN,
845            "Cookie": self._cookie,
846        }
847        if "__Secure-3PAPISID" not in self._cookie:
848            raise LoginFailed(
849                "Invalid Cookie detected. Cookie is missing the __Secure-3PAPISID field. "
850                "Please ensure you are passing the correct cookie. "
851                "You can verify this by checking if the string "
852                "'__Secure-3PAPISID' is present in the cookie string."
853            )
854        sapisid = sapisid_from_cookie(self._cookie)
855        headers["Authorization"] = get_authorization(sapisid + " " + YTM_DOMAIN)
856        self._headers = headers
857
858    def _initialize_context(self) -> None:
859        """Initialize the context to use in requests."""
860        self._context = {
861            "context": {
862                "client": {"clientName": "WEB_REMIX", "clientVersion": "0.1"},
863                "user": {},
864            }
865        }
866
867    def _parse_album(self, album_obj: dict[str, Any], album_id: str | None = None) -> Album:
868        """Parse a YT Album response to an Album model object."""
869        album_id = album_id or album_obj.get("id") or album_obj.get("browseId")
870
871        if not album_id:
872            raise InvalidDataError("Album ID is required but not found")
873
874        if "title" in album_obj:
875            name, version = parse_title_and_version(album_obj["title"])
876        elif "name" in album_obj:
877            name, version = parse_title_and_version(album_obj["name"])
878        else:
879            name, version = "", ""
880        album = Album(
881            item_id=album_id,
882            name=name,
883            version=version,
884            provider=self.instance_id,
885            provider_mappings={
886                ProviderMapping(
887                    item_id=str(album_id),
888                    provider_domain=self.domain,
889                    provider_instance=self.instance_id,
890                    url=f"{YTM_DOMAIN}/playlist?list={album_obj.get('audioPlaylistId')}",
891                )
892            },
893            favorite=album_obj.get("likeStatus", "INDIFFERENT") == "LIKE",
894        )
895        if album_obj.get("year") and album_obj["year"].isdigit():
896            album.year = album_obj["year"]
897        if "thumbnails" in album_obj:
898            album.metadata.images = UniqueList(self._parse_thumbnails(album_obj["thumbnails"]))
899        if description := album_obj.get("description"):
900            album.metadata.description = unquote(description)
901        if "isExplicit" in album_obj:
902            album.metadata.explicit = album_obj["isExplicit"]
903        if "artists" in album_obj:
904            album.artists = UniqueList(
905                [
906                    self._get_artist_item_mapping(artist)
907                    for artist in album_obj["artists"]
908                    if artist.get("id")
909                    or artist.get("channelId")
910                    or artist.get("name") == "Various Artists"
911                ]
912            )
913        if "type" in album_obj:
914            if album_obj["type"] == "Single":
915                album_type = AlbumType.SINGLE
916            elif album_obj["type"] == "EP":
917                album_type = AlbumType.EP
918            elif album_obj["type"] == "Album":
919                album_type = AlbumType.ALBUM
920            else:
921                album_type = AlbumType.UNKNOWN
922            album.album_type = album_type
923
924        # Try inference - override if it finds something more specific
925        inferred_type = infer_album_type(name, version)
926        if inferred_type in (AlbumType.SOUNDTRACK, AlbumType.LIVE):
927            album.album_type = inferred_type
928
929        return album
930
931    def _parse_artist(self, artist_obj: dict[str, Any]) -> Artist:
932        """Parse a YT Artist response to Artist model object."""
933        artist_id = None
934        if "channelId" in artist_obj:
935            artist_id = artist_obj["channelId"]
936        elif artist_obj.get("id"):
937            artist_id = artist_obj["id"]
938        elif artist_obj["name"] == "Various Artists":
939            artist_id = VARIOUS_ARTISTS_YTM_ID
940        if not artist_id:
941            msg = "Artist does not have a valid ID"
942            raise InvalidDataError(msg)
943        artist = Artist(
944            item_id=artist_id,
945            name=artist_obj["name"],
946            provider=self.instance_id,
947            provider_mappings={
948                ProviderMapping(
949                    item_id=str(artist_id),
950                    provider_domain=self.domain,
951                    provider_instance=self.instance_id,
952                    url=f"{YTM_DOMAIN}/channel/{artist_id}",
953                )
954            },
955            favorite=artist_obj.get("likeStatus", "INDIFFERENT") == "LIKE",
956        )
957        if "description" in artist_obj:
958            artist.metadata.description = artist_obj["description"]
959        if artist_obj.get("thumbnails"):
960            artist.metadata.images = self._parse_thumbnails(artist_obj["thumbnails"])
961        return artist
962
963    def _parse_playlist(self, playlist_obj: dict[str, Any]) -> Playlist:
964        """Parse a YT Playlist response to a Playlist object."""
965        raw_playlist_id = playlist_obj["id"]
966        playlist_id = raw_playlist_id
967        playlist_name = playlist_obj["title"]
968        is_editable = playlist_obj.get("owned", playlist_obj.get("privacy", "") == "PRIVATE")
969        # Playlist ID's are not unique across instances for lists like 'Likes', 'Supermix', etc.
970        # So suffix with the instance id to make them unique
971        if playlist_id in YT_PERSONAL_PLAYLISTS:
972            playlist_id = f"{playlist_id}{YT_PLAYLIST_ID_DELIMITER}{self.instance_id}"
973            playlist_name = f"{playlist_name} ({self.name})"
974        playlist = Playlist(
975            item_id=playlist_id,
976            provider=self.instance_id,
977            name=playlist_name,
978            provider_mappings={
979                ProviderMapping(
980                    item_id=playlist_id,
981                    provider_domain=self.domain,
982                    provider_instance=self.instance_id,
983                    url=f"{YTM_DOMAIN}/playlist?list={raw_playlist_id}",
984                    is_unique=is_editable,  # user-owned playlists are unique
985                )
986            },
987            is_editable=is_editable,
988            favorite=playlist_obj.get("likeStatus", "INDIFFERENT") == "LIKE",
989        )
990        if "description" in playlist_obj:
991            playlist.metadata.description = playlist_obj["description"]
992        if playlist_obj.get("thumbnails"):
993            playlist.metadata.images = self._parse_thumbnails(playlist_obj["thumbnails"])
994
995        if authors := playlist_obj.get("author"):
996            if isinstance(authors, str):
997                playlist.owner = authors
998            elif isinstance(authors, list):
999                playlist.owner = authors[0]["name"]
1000            else:
1001                playlist.owner = authors["name"]
1002        else:
1003            playlist.owner = self.name
1004        return playlist
1005
1006    def _parse_track(self, track_obj: dict[str, Any], track_number: int = 0) -> Track:
1007        """Parse a YT Track response to a Track model object."""
1008        if not track_obj.get("videoId"):
1009            msg = "Track is missing videoId"
1010            raise InvalidDataError(msg)
1011        track_id = str(track_obj["videoId"])
1012        name, version = parse_title_and_version(track_obj["title"])
1013        track = Track(
1014            item_id=track_id,
1015            provider=self.instance_id,
1016            name=name,
1017            version=version,
1018            provider_mappings={
1019                ProviderMapping(
1020                    item_id=track_id,
1021                    provider_domain=self.domain,
1022                    provider_instance=self.instance_id,
1023                    available=track_obj.get("isAvailable", True),
1024                    url=f"{YTM_DOMAIN}/watch?v={track_id}",
1025                    audio_format=AudioFormat(
1026                        content_type=ContentType.M4A,
1027                    ),
1028                )
1029            },
1030            favorite=track_obj.get("likeStatus", "INDIFFERENT") == "LIKE",
1031            # Disc info is not available in YTM, assume a single disc
1032            disc_number=1,
1033            # Track number is "sometimes" available in the track object, otherwise approach
1034            # by counting album tracks when fetching full album details
1035            track_number=track_obj.get("trackNumber") or track_number or 0,
1036        )
1037
1038        if track_obj.get("artists"):
1039            track.artists = UniqueList(
1040                self._get_artist_item_mapping(artist)
1041                for artist in track_obj["artists"]
1042                if artist.get("id")
1043                or artist.get("channelId")
1044                or artist.get("name") == "Various Artists"
1045            )
1046        # guard that track has valid artists
1047        if not track.artists:
1048            msg = "Track is missing artists"
1049            raise InvalidDataError(msg)
1050        if track_obj.get("thumbnails"):
1051            track.metadata.images = self._parse_thumbnails(track_obj["thumbnails"])
1052        if (
1053            track_obj.get("album")
1054            and isinstance(track_obj.get("album"), dict)
1055            and track_obj["album"].get("id")
1056        ):
1057            album = track_obj["album"]
1058            track.album = self._get_item_mapping(MediaType.ALBUM, album["id"], album["name"])
1059        if "isExplicit" in track_obj:
1060            track.metadata.explicit = track_obj["isExplicit"]
1061        if "duration" in track_obj and str(track_obj["duration"]).isdigit():
1062            track.duration = int(track_obj["duration"])
1063        elif "duration_seconds" in track_obj and str(track_obj["duration_seconds"]).isdigit():
1064            track.duration = int(track_obj["duration_seconds"])
1065        return track
1066
1067    def _parse_podcast(self, podcast_obj: dict[str, Any]) -> Podcast:
1068        """Parse a YTM Podcast into a MA Podcast."""
1069        podcast = Podcast(
1070            item_id=podcast_obj["podcastId"],
1071            name=podcast_obj["title"],
1072            provider=self.instance_id,
1073            provider_mappings={
1074                ProviderMapping(
1075                    item_id=podcast_obj["podcastId"],
1076                    provider_domain=self.domain,
1077                    provider_instance=self.instance_id,
1078                )
1079            },
1080        )
1081        if description := podcast_obj.get("description"):
1082            podcast.metadata.description = description
1083        if author := podcast_obj.get("author") or podcast_obj.get("channel"):
1084            podcast.publisher = author["name"]
1085        if thumbnails := podcast_obj.get("thumbnails"):
1086            podcast.metadata.images = self._parse_thumbnails(thumbnails)
1087        return podcast
1088
1089    def _parse_browse_podcast(self, item_obj: dict[str, Any]) -> Podcast | None:
1090        """Parse a podcast from a browse or search item."""
1091        browse_id: str = item_obj.get("browseId") or ""
1092        if not browse_id.startswith("MPSP"):
1093            return None
1094        podcast_obj = dict(item_obj)
1095        # Match the unprefixed IDs used by library podcasts.
1096        podcast_obj["podcastId"] = item_obj.get("podcastId") or browse_id.removeprefix("MPSP")
1097        return self._parse_podcast(podcast_obj)
1098
1099    def _parse_podcast_episode(
1100        self, episode_obj: dict[str, Any], podcast: Podcast
1101    ) -> PodcastEpisode:
1102        """Parse a raw episode into a PodcastEpisode."""
1103        episode_id = episode_obj.get("videoId")
1104        if not episode_id:
1105            msg = "Podcast episode is missing videoId"
1106            raise InvalidDataError(msg)
1107        item_id = f"{podcast.item_id}{PODCAST_EPISODE_SPLITTER}{episode_id}"
1108        episode = PodcastEpisode(
1109            item_id=item_id,
1110            provider=self.instance_id,
1111            name=episode_obj.get("title", ""),
1112            position=0,
1113            podcast=podcast,
1114            provider_mappings={
1115                ProviderMapping(
1116                    item_id=item_id,
1117                    provider_domain=self.domain,
1118                    provider_instance=self.instance_id,
1119                    audio_format=AudioFormat(
1120                        content_type=ContentType.M4A,
1121                    ),
1122                    url=f"{YTM_DOMAIN}/watch?v={episode_id}",
1123                )
1124            },
1125        )
1126        if duration := episode_obj.get("duration"):
1127            duration_sec = parse_str_duration(duration)
1128            episode.duration = int(duration_sec)
1129        if description := episode_obj.get("description"):
1130            # the single episode endpoint returns a Description object instead of a string
1131            episode.metadata.description = (
1132                description.text if isinstance(description, Description) else description
1133            )
1134        if thumbnails := episode_obj.get("thumbnails"):
1135            episode.metadata.images = self._parse_thumbnails(thumbnails)
1136        if release_date := episode_obj.get("date"):
1137            with suppress(ValueError):
1138                episode.metadata.release_date = datetime.fromisoformat(release_date)
1139        return episode
1140
1141    async def _get_stream_format(self, item_id: str) -> dict[str, Any]:
1142        """Figure out the stream URL to use and return the highest quality."""
1143
1144        def _extract_best_stream_url_format() -> dict[str, Any]:
1145            if self._yt_dlp_module is None:
1146                self._yt_dlp_module = importlib.import_module("yt_dlp")
1147            yt_dlp = self._yt_dlp_module
1148            url = f"{YTM_DOMAIN}/watch?v={item_id}"
1149            ydl_opts = {
1150                "quiet": self.logger.level > logging.DEBUG,
1151                "verbose": self.logger.level == VERBOSE_LOG_LEVEL,
1152                "cookiefile": StringIO(self._netscape_cookie),
1153                # This enforces a player client and skips unnecessary scraping to increase speed
1154                "extractor_args": {
1155                    "youtubepot-bgutilhttp": {
1156                        "base_url": [self._po_token_server_url],
1157                    },
1158                    "youtube": {
1159                        "skip": ["translated_subs", "dash"],
1160                        "player_client": ["web_music"],
1161                    },
1162                },
1163            }
1164            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
1165                try:
1166                    info = ydl.extract_info(url, download=False)
1167                except yt_dlp.utils.DownloadError as err:
1168                    raise UnplayableMediaError(err) from err
1169                format_selector = ydl.build_format_selector("m4a/bestaudio")
1170                stream_format: dict[str, Any] | None = next(
1171                    format_selector({"formats": info["formats"]}), None
1172                )
1173                if not stream_format:
1174                    raise UnplayableMediaError("No stream formats found")
1175                return stream_format
1176
1177        return await asyncio.to_thread(_extract_best_stream_url_format)
1178
1179    def _get_item_mapping(self, media_type: MediaType, key: str, name: str) -> ItemMapping:
1180        return ItemMapping(
1181            media_type=media_type,
1182            item_id=key,
1183            provider=self.instance_id,
1184            name=name,
1185        )
1186
1187    def _get_artist_item_mapping(self, artist_obj: dict[str, Any]) -> ItemMapping:
1188        artist_id = artist_obj.get("id") or artist_obj.get("channelId")
1189        if not artist_id and artist_obj["name"] == "Various Artists":
1190            artist_id = VARIOUS_ARTISTS_YTM_ID
1191        # callers only pass artists with a valid id
1192        return self._get_item_mapping(
1193            MediaType.ARTIST, cast("str", artist_id), artist_obj.get("name", "")
1194        )
1195
1196    async def _verify_po_token_url(self) -> bool:
1197        """Ping the PO Token server and verify the response."""
1198        url = f"{self._po_token_server_url}/ping"
1199        try:
1200            async with self.mass.http_session.get(url) as response:
1201                response.raise_for_status()
1202                self.logger.debug("PO Token server responded with %s", response.status)
1203                return response.status == 200
1204        except (ClientError, TimeoutError) as err:
1205            self.logger.debug("PO Token server ping failed: %s", err)
1206            return False
1207
1208    async def _user_has_ytm_premium(self) -> bool:
1209        """Check if the user has Youtube Music Premium."""
1210        stream_format = await self._get_stream_format(YTM_PREMIUM_CHECK_TRACK_ID)
1211        # Only premium users can stream the HQ stream of this song
1212        format_id: str = stream_format["format_id"]
1213        return format_id == "141"
1214
1215    def _parse_thumbnails(self, thumbnails_obj: list[dict[str, Any]]) -> UniqueList[MediaItemImage]:
1216        """Parse and YTM thumbnails to MediaItemImage."""
1217        result: UniqueList[MediaItemImage] = UniqueList()
1218        processed_images = set()
1219        for img in sorted(thumbnails_obj, key=lambda w: w.get("width", 0), reverse=True):
1220            url: str = img["url"]
1221            url_base = url.split("=w", maxsplit=1)[0]
1222            width: int = img["width"]
1223            height: int = img["height"]
1224            image_ratio: float = width / height
1225            image_type = (
1226                ImageType.LANDSCAPE
1227                if "maxresdefault" in url or image_ratio > 2.0
1228                else ImageType.THUMB
1229            )
1230            if "=w" not in url and width < 500:
1231                continue
1232            # if the size is in the url, we can actually request a higher thumb
1233            if "=w" in url and width < 600:
1234                url = f"{url_base}=w600-h600-p"
1235                image_type = ImageType.THUMB
1236            if (url_base, image_type) in processed_images:
1237                continue
1238            processed_images.add((url_base, image_type))
1239            result.append(
1240                MediaItemImage(
1241                    type=image_type,
1242                    path=url,
1243                    provider=self.instance_id,
1244                    remotely_accessible=True,
1245                )
1246            )
1247        return result
1248
1249    async def _install_packages(self) -> None:
1250        """Install frequently changing packages dynamically."""
1251        # NOTE: Google breaks things quite often which requires us to update
1252        # some packages very frequently. Installing them dynamically prevents
1253        # us from having to update MA to ensure this provider works.
1254        for package_name in PACKAGES_TO_INSTALL:
1255            await install_package(package_name)
1256        # verify if the yt_dlp package is usable
1257        try:
1258            await import_module_in_thread("yt_dlp")
1259        except ImportError:
1260            raise SetupFailedError("Package yt_dlp failed to install")
1261