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