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