/
/
/
1"""
2Helper module for parsing the Youtube Music API.
3
4This helpers file is an async wrapper around the excellent ytmusicapi package.
5While the ytmusicapi package does an excellent job at parsing the Youtube Music results,
6it is unfortunately not async, which is required for Music Assistant to run smoothly.
7This also nicely separates the parsing logic from the Youtube Music provider logic.
8"""
9
10import asyncio
11from collections.abc import Callable
12from http.cookies import SimpleCookie
13from time import time
14from typing import Any, Literal
15
16import ytmusicapi
17from music_assistant_models.errors import LoginFailed
18from ytmusicapi import LikeStatus
19from ytmusicapi.exceptions import YTMusicError
20
21from music_assistant.providers.ytmusic.constants import YTMRecommendationIcons
22
23# subset of ytmusicapi's accepted search filters that we use
24YTMSearchFilter = Literal["artists", "albums", "songs", "playlists", "podcasts"]
25
26
27async def get_artist(
28 prov_artist_id: str, headers: dict[str, str], language: str = "en"
29) -> dict[str, Any]:
30 """Async wrapper around the ytmusicapi get_artist function."""
31
32 def _get_artist() -> dict[str, Any]:
33 ytm = ytmusicapi.YTMusic(auth=headers, language=language)
34 try:
35 artist = ytm.get_artist(channelId=prov_artist_id)
36 # ChannelId can sometimes be different and original ID is not part of the response
37 artist["channelId"] = prov_artist_id
38 except KeyError as err:
39 _raise_if_signed_out(err)
40 try:
41 user = ytm.get_user(channelId=prov_artist_id)
42 artist = {"channelId": prov_artist_id, "name": user["name"]}
43 except KeyError as err:
44 _raise_if_signed_out(err)
45 artist = {"channelId": prov_artist_id, "name": "Unknown"}
46 return artist
47
48 return await _run_ytmusic(_get_artist)
49
50
51async def get_album(
52 headers: dict[str, str], prov_album_id: str, language: str = "en", user: str | None = None
53) -> dict[str, Any]:
54 """Async wrapper around the ytmusicapi get_album function."""
55
56 def _get_album() -> dict[str, Any]:
57 if prov_album_id.startswith("FEmusic_library_privately_owned_release"):
58 ytm = ytmusicapi.YTMusic(auth=headers, language=language, user=user)
59 album = ytm.get_library_upload_album(browseId=prov_album_id)
60 else:
61 ytm = ytmusicapi.YTMusic(language=language)
62 album = ytm.get_album(browseId=prov_album_id)
63
64 if "audioPlaylistId" in album:
65 # Track id's from album tracks do not match with actual album tracks. E.g. a track
66 # points to the videoId of the original version, while we want the album version
67 try:
68 album_playlist = ytm.get_playlist(playlistId=album["audioPlaylistId"], limit=None)
69 except YTMusicError:
70 return album
71
72 # Do some basic checks
73 if len(album_playlist.get("tracks", [])) != len(album.get("tracks", [])):
74 return album
75 # Move the correct track info to the album tracks
76 playlist_tracks_by_title = {t.get("title"): t for t in album_playlist.get("tracks", [])}
77 for album_track in album.get("tracks", []):
78 if playlist_track := playlist_tracks_by_title.get(album_track.get("title")):
79 album_track["videoId"] = playlist_track["videoId"]
80 album_track["isAvailable"] = playlist_track.get("isAvailable", True)
81 album_track["likeStatus"] = playlist_track.get("likeStatus", "INDIFFERENT")
82 return album
83
84 return await _run_ytmusic(_get_album)
85
86
87async def get_playlist(
88 prov_playlist_id: str,
89 headers: dict[str, str],
90 language: str = "en",
91 user: str | None = None,
92 limit: int | None = None,
93) -> dict[str, Any]:
94 """Async wrapper around the ytmusicapi get_playlist function."""
95
96 def _get_playlist() -> dict[str, Any]:
97 ytm = ytmusicapi.YTMusic(auth=headers, language=language, user=user)
98 playlist = ytm.get_playlist(playlistId=prov_playlist_id, limit=limit)
99 playlist["checksum"] = get_playlist_checksum(playlist)
100 # Fix missing playlist id in some edge cases
101 playlist["id"] = prov_playlist_id if not playlist.get("id") else playlist["id"]
102 return playlist
103
104 return await _run_ytmusic(_get_playlist)
105
106
107async def get_track(
108 prov_track_id: str, headers: dict[str, str], language: str = "en"
109) -> dict[str, Any] | None:
110 """Async wrapper around the ytmusicapi get_playlist function."""
111
112 def _get_song() -> dict[str, Any] | None:
113 ytm = ytmusicapi.YTMusic(auth=headers, language=language)
114 track_obj = ytm.get_song(videoId=prov_track_id)
115 track: dict[str, Any] = {}
116 if "videoDetails" not in track_obj:
117 # video that no longer exists
118 return None
119 track["videoId"] = track_obj["videoDetails"]["videoId"]
120 track["title"] = track_obj["videoDetails"]["title"]
121 track["artists"] = [
122 {
123 "channelId": track_obj["videoDetails"]["channelId"],
124 "name": track_obj["videoDetails"]["author"],
125 }
126 ]
127 track["duration"] = track_obj["videoDetails"]["lengthSeconds"]
128 track["thumbnails"] = track_obj["microformat"]["microformatDataRenderer"]["thumbnail"][
129 "thumbnails"
130 ]
131 if track_thumbs := track_obj["videoDetails"].get("thumbnail", {}).get("thumbnails"):
132 track["thumbnails"] = track.get("thumbnails", []) + track_thumbs
133 track["isAvailable"] = track_obj["playabilityStatus"]["status"] == "OK"
134 return track
135
136 return await _run_ytmusic(_get_song)
137
138
139async def get_podcast(
140 prov_podcast_id: str, headers: dict[str, str], language: str = "en"
141) -> dict[str, Any]:
142 """Async wrapper around the get_podcast function."""
143
144 def _get_podcast() -> dict[str, Any]:
145 ytm = ytmusicapi.YTMusic(auth=headers, language=language)
146 podcast_obj = ytm.get_podcast(playlistId=prov_podcast_id)
147 if "podcastId" not in podcast_obj:
148 podcast_obj["podcastId"] = prov_podcast_id
149 return podcast_obj
150
151 return await _run_ytmusic(_get_podcast)
152
153
154async def get_podcast_episode(
155 prov_episode_id: str, headers: dict[str, str], language: str = "en"
156) -> dict[str, Any]:
157 """Async wrapper around the podcast episode function."""
158
159 def _get_podcast_episode() -> dict[str, Any]:
160 ytm = ytmusicapi.YTMusic(auth=headers, language=language)
161 episode = ytm.get_episode(videoId=prov_episode_id)
162 if "videoId" not in episode:
163 episode["videoId"] = prov_episode_id
164 return episode
165
166 return await _run_ytmusic(_get_podcast_episode)
167
168
169async def get_library_artists(
170 headers: dict[str, str], language: str = "en", user: str | None = None
171) -> list[dict[str, Any]]:
172 """Async wrapper around the ytmusicapi get_library_artists function."""
173
174 def _get_library_artists() -> list[dict[str, Any]]:
175 ytm = ytmusicapi.YTMusic(auth=headers, language=language, user=user)
176 artists = ytm.get_library_subscriptions(limit=9999)
177 # Sync properties with uniformal artist object
178 for artist in artists:
179 artist["id"] = artist["browseId"]
180 artist["name"] = artist["artist"]
181 del artist["browseId"]
182 del artist["artist"]
183 return artists
184
185 return await _run_ytmusic(_get_library_artists)
186
187
188async def get_library_albums(
189 headers: dict[str, str], language: str = "en", user: str | None = None
190) -> list[dict[str, Any]]:
191 """Async wrapper around the ytmusicapi get_library_albums function."""
192
193 def _get_library_albums() -> list[dict[str, Any]]:
194 ytm = ytmusicapi.YTMusic(auth=headers, language=language, user=user)
195 return ytm.get_library_albums(limit=9999)
196
197 return await _run_ytmusic(_get_library_albums)
198
199
200async def get_library_playlists(
201 headers: dict[str, str], language: str = "en", user: str | None = None
202) -> list[dict[str, Any]]:
203 """Async wrapper around the ytmusicapi get_library_playlists function."""
204
205 def _get_library_playlists() -> list[dict[str, Any]]:
206 ytm = ytmusicapi.YTMusic(auth=headers, language=language, user=user)
207 playlists = ytm.get_library_playlists(limit=9999)
208 # Sync properties with uniformal playlist object
209 for playlist in playlists:
210 playlist["id"] = playlist["playlistId"]
211 del playlist["playlistId"]
212 playlist["checksum"] = get_playlist_checksum(playlist)
213 return playlists
214
215 return await _run_ytmusic(_get_library_playlists)
216
217
218async def get_library_tracks(
219 headers: dict[str, str], language: str = "en", user: str | None = None
220) -> list[dict[str, Any]]:
221 """Async wrapper around the ytmusicapi get_library_tracks function."""
222
223 def _get_library_tracks() -> list[dict[str, Any]]:
224 ytm = ytmusicapi.YTMusic(auth=headers, language=language, user=user)
225 return ytm.get_library_songs(limit=9999)
226
227 return await _run_ytmusic(_get_library_tracks)
228
229
230async def get_library_podcasts(
231 headers: dict[str, str], language: str = "en", user: str | None = None
232) -> list[dict[str, Any]]:
233 """Async wrapper around the ytmusic api get_library_podcasts function."""
234
235 def _get_library_podcasts() -> list[dict[str, Any]]:
236 ytm = ytmusicapi.YTMusic(auth=headers, language=language, user=user)
237 return ytm.get_library_podcasts(limit=9999)
238
239 return await _run_ytmusic(_get_library_podcasts)
240
241
242async def library_add_remove_artist(
243 headers: dict[str, str], prov_artist_id: str, add: bool = True, user: str | None = None
244) -> bool:
245 """Add or remove an artist to the user's library."""
246
247 def _library_add_remove_artist() -> bool:
248 ytm = ytmusicapi.YTMusic(auth=headers, user=user)
249 if add:
250 return "actions" in ytm.subscribe_artists(channelIds=[prov_artist_id])
251 return "actions" in ytm.unsubscribe_artists(channelIds=[prov_artist_id])
252
253 return await _run_ytmusic(_library_add_remove_artist)
254
255
256async def library_add_remove_album(
257 headers: dict[str, str], prov_item_id: str, add: bool = True, user: str | None = None
258) -> dict[str, Any]:
259 """Add or remove an album or playlist to the user's library."""
260 album = await get_album(headers=headers, prov_album_id=prov_item_id, user=user)
261
262 def _library_add_remove_album() -> dict[str, Any]:
263 ytm = ytmusicapi.YTMusic(auth=headers, user=user)
264 playlist_id = album["audioPlaylistId"]
265 if add:
266 return ytm.rate_playlist(playlist_id, LikeStatus.LIKE)
267 return ytm.rate_playlist(playlist_id, LikeStatus.INDIFFERENT)
268
269 return await _run_ytmusic(_library_add_remove_album)
270
271
272async def library_add_remove_playlist(
273 headers: dict[str, str], prov_item_id: str, add: bool = True, user: str | None = None
274) -> bool:
275 """Add or remove an album or playlist to the user's library."""
276
277 def _library_add_remove_playlist() -> bool:
278 ytm = ytmusicapi.YTMusic(auth=headers, user=user)
279 if add:
280 return "actions" in ytm.rate_playlist(prov_item_id, LikeStatus.LIKE)
281 return "actions" in ytm.rate_playlist(prov_item_id, LikeStatus.INDIFFERENT)
282
283 return await _run_ytmusic(_library_add_remove_playlist)
284
285
286async def add_remove_playlist_tracks(
287 headers: dict[str, str],
288 prov_playlist_id: str,
289 prov_track_ids: list[Any],
290 add: bool,
291 user: str | None = None,
292) -> str | dict[str, Any]:
293 """Async wrapper around adding/removing tracks to a playlist."""
294
295 def _add_playlist_tracks() -> str | dict[str, Any]:
296 ytm = ytmusicapi.YTMusic(auth=headers, user=user)
297 if add:
298 return ytm.add_playlist_items(playlistId=prov_playlist_id, videoIds=prov_track_ids)
299 return ytm.remove_playlist_items(playlistId=prov_playlist_id, videos=prov_track_ids)
300
301 return await _run_ytmusic(_add_playlist_tracks)
302
303
304async def get_song_radio_tracks(
305 headers: dict[str, str], prov_item_id: str, limit: int = 25, user: str | None = None
306) -> dict[str, Any]:
307 """Async wrapper around the ytmusicapi radio function."""
308
309 def _get_song_radio_tracks() -> dict[str, Any]:
310 ytm = ytmusicapi.YTMusic(auth=headers, user=user)
311 playlist_id = f"RDAMVM{prov_item_id}"
312 result = ytm.get_watch_playlist(
313 videoId=prov_item_id, playlistId=playlist_id, limit=limit, radio=True
314 )
315 # Replace inconsistensies for easier parsing
316 tracks = result.get("tracks")
317 if isinstance(tracks, list):
318 for track in tracks:
319 if track.get("thumbnail"):
320 track["thumbnails"] = track["thumbnail"]
321 del track["thumbnail"]
322 if track.get("length"):
323 track["duration"] = get_sec(track["length"])
324 return result
325
326 return await _run_ytmusic(_get_song_radio_tracks)
327
328
329async def search(
330 query: str,
331 headers: dict[str, str],
332 ytm_filter: YTMSearchFilter | None = None,
333 limit: int = 20,
334 user: str | None = None,
335) -> list[dict[str, Any]]:
336 """Async wrapper around the ytmusicapi search function."""
337
338 def _search() -> list[dict[str, Any]]:
339 # Always search in English: ytmusicapi (1.12.2) matches the result shelf title,
340 # which YouTube returns translated, against the English filter name, so a filtered
341 # search silently returns nothing in most other languages. English is what this
342 # provider expects anyway, as it compares result fields such as the album type
343 # against English literals. Revisit once ytmusicapi compares against the
344 # translated title.
345 ytm = ytmusicapi.YTMusic(auth=headers, language="en", user=user)
346 results = ytm.search(query=query, filter=ytm_filter, limit=limit)
347 # Sync result properties with uniformal objects
348 for result in results:
349 if result["resultType"] == "artist":
350 if "artists" in result and len(result["artists"]) > 0:
351 result["id"] = result["artists"][0]["id"]
352 result["name"] = result["artists"][0]["name"]
353 del result["artists"]
354 else:
355 result["id"] = result["browseId"]
356 result["name"] = result["artist"]
357 del result["browseId"]
358 del result["artist"]
359 elif result["resultType"] == "playlist":
360 if "playlistId" in result:
361 result["id"] = result["playlistId"]
362 del result["playlistId"]
363 elif "browseId" in result:
364 result["id"] = result["browseId"]
365 del result["browseId"]
366 return results[:limit]
367
368 return await _run_ytmusic(_search)
369
370
371def get_playlist_checksum(playlist_obj: dict[str, Any]) -> str:
372 """Try to calculate a checksum so we can detect changes in a playlist."""
373 for key in ("duration_seconds", "trackCount", "count"):
374 if key in playlist_obj:
375 return str(playlist_obj[key])
376 return str(int(time()))
377
378
379def is_brand_account(username: str) -> bool:
380 """Check if the provided username is a brand-account."""
381 return len(username) == 21 and username.isdigit()
382
383
384def get_sec(time_str: str) -> int:
385 """Get seconds from time."""
386 parts = time_str.split(":")
387 if len(parts) == 3:
388 return int(parts[0]) * 3600 + int(parts[1]) * 60 + int(parts[2])
389 if len(parts) == 2:
390 return int(parts[0]) * 60 + int(parts[1])
391 return 0
392
393
394def convert_to_netscape(raw_cookie_str: str, domain: str) -> str:
395 """Convert a raw cookie into Netscape format, so yt-dl can use it."""
396 domain = domain.replace("https://", "")
397 cookie = SimpleCookie()
398 cookie.load(rawdata=raw_cookie_str)
399 netscape_cookie = "# Netscape HTTP Cookie File\n"
400 for morsel in cookie.values():
401 netscape_cookie += f"{domain}\tTRUE\t/\tTRUE\t0\t{morsel.key}\t{morsel.value}\n"
402 return netscape_cookie
403
404
405async def get_home(
406 headers: dict[str, str], language: str = "en", user: str | None = None, limit: int = 3
407) -> list[dict[str, Any]]:
408 """Get the recommendations from the home page."""
409
410 def _get_home() -> list[dict[str, Any]]:
411 ytm = ytmusicapi.YTMusic(auth=headers, language=language, user=user)
412 return ytm.get_home(limit=limit)
413
414 return await _run_ytmusic(_get_home)
415
416
417def determine_recommendation_icon(name: str) -> str:
418 """Determine the icon for a recommendation based on its name."""
419 query = name.lower()
420
421 if "listen again" in query:
422 return YTMRecommendationIcons.LISTEN_AGAIN
423 if "continue" in query:
424 return YTMRecommendationIcons.CONTINUE_WATCHING
425 if "your mix" in query:
426 return YTMRecommendationIcons.YOUR_MIX
427 if "new" in query:
428 return YTMRecommendationIcons.NEW_RELEASES
429 if "recommended" in query:
430 return YTMRecommendationIcons.RECOMMENDED
431 return YTMRecommendationIcons.DEFAULT
432
433
434async def _run_ytmusic[T](func: Callable[[], T]) -> T:
435 """Run a blocking ytmusicapi call in a thread, translating a signed-out session."""
436 try:
437 return await asyncio.to_thread(func)
438 except (KeyError, IndexError) as err:
439 _raise_if_signed_out(err)
440 raise
441
442
443def _raise_if_signed_out(err: Exception) -> None:
444 """Raise LoginFailed if the error carries YouTube's signed-out page."""
445 # An invalid cookie makes YouTube answer with its signed-out page instead of an auth
446 # error, so nav() fails on the unexpected payload and embeds it in the message.
447 if "signInEndpoint" not in str(err):
448 return
449 raise LoginFailed(
450 "Your YouTube Music session is no longer valid. "
451 "Please reconfigure this provider with a fresh cookie."
452 ) from err
453