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