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