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