music-assistant-server

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    language: str = "en",
335    user: str | None = None,
336) -> list[dict[str, Any]]:
337    """Async wrapper around the ytmusicapi search function."""
338
339    def _search() -> list[dict[str, Any]]:
340        ytm = ytmusicapi.YTMusic(auth=headers, language=language, user=user)
341        results = ytm.search(query=query, filter=ytm_filter, limit=limit)
342        # Sync result properties with uniformal objects
343        for result in results:
344            if result["resultType"] == "artist":
345                if "artists" in result and len(result["artists"]) > 0:
346                    result["id"] = result["artists"][0]["id"]
347                    result["name"] = result["artists"][0]["name"]
348                    del result["artists"]
349                else:
350                    result["id"] = result["browseId"]
351                    result["name"] = result["artist"]
352                    del result["browseId"]
353                    del result["artist"]
354            elif result["resultType"] == "playlist":
355                if "playlistId" in result:
356                    result["id"] = result["playlistId"]
357                    del result["playlistId"]
358                elif "browseId" in result:
359                    result["id"] = result["browseId"]
360                    del result["browseId"]
361        return results[:limit]
362
363    return await _run_ytmusic(_search)
364
365
366def get_playlist_checksum(playlist_obj: dict[str, Any]) -> str:
367    """Try to calculate a checksum so we can detect changes in a playlist."""
368    for key in ("duration_seconds", "trackCount", "count"):
369        if key in playlist_obj:
370            return str(playlist_obj[key])
371    return str(int(time()))
372
373
374def is_brand_account(username: str) -> bool:
375    """Check if the provided username is a brand-account."""
376    return len(username) == 21 and username.isdigit()
377
378
379def get_sec(time_str: str) -> int:
380    """Get seconds from time."""
381    parts = time_str.split(":")
382    if len(parts) == 3:
383        return int(parts[0]) * 3600 + int(parts[1]) * 60 + int(parts[2])
384    if len(parts) == 2:
385        return int(parts[0]) * 60 + int(parts[1])
386    return 0
387
388
389def convert_to_netscape(raw_cookie_str: str, domain: str) -> str:
390    """Convert a raw cookie into Netscape format, so yt-dl can use it."""
391    domain = domain.replace("https://", "")
392    cookie = SimpleCookie()
393    cookie.load(rawdata=raw_cookie_str)
394    netscape_cookie = "# Netscape HTTP Cookie File\n"
395    for morsel in cookie.values():
396        netscape_cookie += f"{domain}\tTRUE\t/\tTRUE\t0\t{morsel.key}\t{morsel.value}\n"
397    return netscape_cookie
398
399
400async def get_home(
401    headers: dict[str, str], language: str = "en", user: str | None = None, limit: int = 3
402) -> list[dict[str, Any]]:
403    """Get the recommendations from the home page."""
404
405    def _get_home() -> list[dict[str, Any]]:
406        ytm = ytmusicapi.YTMusic(auth=headers, language=language, user=user)
407        return ytm.get_home(limit=limit)
408
409    return await _run_ytmusic(_get_home)
410
411
412def determine_recommendation_icon(name: str) -> str:
413    """Determine the icon for a recommendation based on its name."""
414    query = name.lower()
415
416    if "listen again" in query:
417        return YTMRecommendationIcons.LISTEN_AGAIN
418    if "continue" in query:
419        return YTMRecommendationIcons.CONTINUE_WATCHING
420    if "your mix" in query:
421        return YTMRecommendationIcons.YOUR_MIX
422    if "new" in query:
423        return YTMRecommendationIcons.NEW_RELEASES
424    if "recommended" in query:
425        return YTMRecommendationIcons.RECOMMENDED
426    return YTMRecommendationIcons.DEFAULT
427
428
429async def _run_ytmusic[T](func: Callable[[], T]) -> T:
430    """Run a blocking ytmusicapi call in a thread, translating a signed-out session."""
431    try:
432        return await asyncio.to_thread(func)
433    except (KeyError, IndexError) as err:
434        _raise_if_signed_out(err)
435        raise
436
437
438def _raise_if_signed_out(err: Exception) -> None:
439    """Raise LoginFailed if the error carries YouTube's signed-out page."""
440    # An invalid cookie makes YouTube answer with its signed-out page instead of an auth
441    # error, so nav() fails on the unexpected payload and embeds it in the message.
442    if "signInEndpoint" not in str(err):
443        return
444    raise LoginFailed(
445        "Your YouTube Music session is no longer valid. "
446        "Please reconfigure this provider with a fresh cookie."
447    ) from err
448