music-assistant-server

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