/
/
/
1"""Tidal music provider implementation."""
2
3from __future__ import annotations
4
5import json
6from datetime import datetime
7from typing import TYPE_CHECKING, Any
8
9from music_assistant_models.enums import MediaType, ProviderFeature
10from music_assistant_models.errors import LoginFailed
11from music_assistant_models.media_items import (
12 Album,
13 Artist,
14 ItemMapping,
15 MediaItemType,
16 Playlist,
17 RecommendationFolder,
18 SearchResults,
19 Track,
20)
21
22from music_assistant.controllers.cache import use_cache
23from music_assistant.models.music_provider import MusicProvider
24
25from .api_client import TidalAPIClient
26from .auth_manager import TidalAuthManager
27from .constants import (
28 CACHE_CATEGORY_RECOMMENDATIONS,
29 CONF_AUTH_TOKEN,
30 CONF_EXPIRY_TIME,
31 CONF_REFRESH_TOKEN,
32 CONF_USER_ID,
33)
34from .library import TidalLibraryManager
35from .media import TidalMediaManager
36from .playlist import TidalPlaylistManager
37from .recommendations import TidalRecommendationManager
38from .streaming import TidalStreamingManager
39
40if TYPE_CHECKING:
41 from collections.abc import AsyncGenerator
42
43 from music_assistant_models.config_entries import ProviderConfig
44 from music_assistant_models.provider import ProviderManifest
45 from music_assistant_models.streamdetails import StreamDetails
46
47 from music_assistant.mass import MusicAssistant
48
49
50SUPPORTED_FEATURES = {
51 ProviderFeature.LIBRARY_ARTISTS,
52 ProviderFeature.LIBRARY_ALBUMS,
53 ProviderFeature.LIBRARY_TRACKS,
54 ProviderFeature.LIBRARY_PLAYLISTS,
55 ProviderFeature.ARTIST_ALBUMS,
56 ProviderFeature.ARTIST_TOPTRACKS,
57 ProviderFeature.SEARCH,
58 ProviderFeature.LIBRARY_ARTISTS_EDIT,
59 ProviderFeature.LIBRARY_ALBUMS_EDIT,
60 ProviderFeature.LIBRARY_TRACKS_EDIT,
61 ProviderFeature.LIBRARY_PLAYLISTS_EDIT,
62 ProviderFeature.PLAYLIST_CREATE,
63 ProviderFeature.SIMILAR_TRACKS,
64 ProviderFeature.BROWSE,
65 ProviderFeature.PLAYLIST_TRACKS_EDIT,
66 ProviderFeature.RECOMMENDATIONS,
67 ProviderFeature.LYRICS,
68}
69
70
71class TidalProvider(MusicProvider):
72 """Implementation of a Tidal MusicProvider."""
73
74 def __init__(self, mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig):
75 """Initialize Tidal provider."""
76 super().__init__(mass, manifest, config, SUPPORTED_FEATURES)
77 self.auth = TidalAuthManager(
78 http_session=mass.http_session,
79 config_updater=self._update_auth_config,
80 logger=self.logger,
81 )
82 self.api = TidalAPIClient(self)
83 self.library = TidalLibraryManager(self)
84 self.media = TidalMediaManager(self)
85 self.playlists = TidalPlaylistManager(self)
86 self.recommendations_manager = TidalRecommendationManager(self)
87 self.streaming = TidalStreamingManager(self)
88
89 def _update_auth_config(self, auth_info: dict[str, Any]) -> None:
90 """Update auth config with new auth info."""
91 self._update_config_value(CONF_AUTH_TOKEN, auth_info["access_token"], encrypted=True)
92 self._update_config_value(CONF_REFRESH_TOKEN, auth_info["refresh_token"], encrypted=True)
93 self._update_config_value(CONF_EXPIRY_TIME, auth_info["expires_at"])
94 self._update_config_value(CONF_USER_ID, auth_info["userId"])
95
96 async def handle_async_init(self) -> None:
97 """Handle async initialization of the provider."""
98 access_token = self.config.get_value(CONF_AUTH_TOKEN)
99 refresh_token = self.config.get_value(CONF_REFRESH_TOKEN)
100 expires_at = self.config.get_value(CONF_EXPIRY_TIME)
101 user_id = self.config.get_value(CONF_USER_ID)
102
103 if not access_token or not refresh_token:
104 raise LoginFailed("Missing authentication data")
105
106 if isinstance(expires_at, str) and "T" in expires_at:
107 try:
108 dt = datetime.fromisoformat(expires_at)
109 expires_at = dt.timestamp()
110 self._update_config_value(CONF_EXPIRY_TIME, expires_at)
111 except ValueError:
112 expires_at = 0
113
114 auth_data = {
115 "access_token": access_token,
116 "refresh_token": refresh_token,
117 "expires_at": expires_at,
118 "userId": user_id,
119 }
120
121 if not await self.auth.initialize(json.dumps(auth_data)):
122 raise LoginFailed("Failed to authenticate with Tidal")
123
124 api_result = await self.api.get("sessions")
125 user_info = api_result[0] if isinstance(api_result, tuple) else api_result
126 logged_in_user = await self.get_user(str(user_info.get("userId")))
127 await self.auth.update_user_info(logged_in_user, str(user_info.get("sessionId")))
128
129 async def get_user(self, prov_user_id: str) -> dict[str, Any]:
130 """Get user information."""
131 return await self.api.get_data(f"users/{prov_user_id}")
132
133 @use_cache(3600 * 24 * 14)
134 async def search(
135 self, search_query: str, media_types: list[MediaType], limit: int = 5
136 ) -> SearchResults:
137 """Perform search on musicprovider."""
138 return await self.media.search(search_query, media_types, limit)
139
140 @use_cache(3600 * 24)
141 async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
142 """Get similar tracks for given track id."""
143 return await self.media.get_similar_tracks(prov_track_id, limit)
144
145 @use_cache(3600 * 24 * 30)
146 async def get_artist(self, prov_artist_id: str) -> Artist:
147 """Get artist details for given artist id."""
148 return await self.media.get_artist(prov_artist_id)
149
150 @use_cache(3600 * 24 * 30)
151 async def get_album(self, prov_album_id: str) -> Album:
152 """Get album details for given album id."""
153 return await self.media.get_album(prov_album_id)
154
155 @use_cache(3600 * 24 * 30)
156 async def get_track(self, prov_track_id: str) -> Track:
157 """Get track details for given track id."""
158 return await self.media.get_track(prov_track_id)
159
160 @use_cache(3600 * 24 * 30)
161 async def get_playlist(self, prov_playlist_id: str) -> Playlist:
162 """Get playlist details for given playlist id."""
163 return await self.media.get_playlist(prov_playlist_id)
164
165 @use_cache(3600 * 24 * 30)
166 async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
167 """Get album tracks for given album id."""
168 return await self.media.get_album_tracks(prov_album_id)
169
170 @use_cache(3600 * 24 * 7)
171 async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
172 """Get a list of all albums for the given artist."""
173 return await self.media.get_artist_albums(prov_artist_id)
174
175 @use_cache(3600 * 24 * 7)
176 async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
177 """Get a list of 10 most popular tracks for the given artist."""
178 return await self.media.get_artist_toptracks(prov_artist_id)
179
180 @use_cache(3600 * 3)
181 async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
182 """Get playlist tracks."""
183 return await self.media.get_playlist_tracks(prov_playlist_id, page)
184
185 async def get_stream_details(
186 self, item_id: str, media_type: MediaType = MediaType.TRACK
187 ) -> StreamDetails:
188 """Return the content details for the given track when it will be streamed."""
189 return await self.streaming.get_stream_details(item_id)
190
191 def get_item_mapping(self, media_type: MediaType, key: str, name: str) -> ItemMapping:
192 """Create a generic item mapping."""
193 return ItemMapping(
194 media_type=media_type,
195 item_id=key,
196 provider=self.instance_id,
197 name=name,
198 )
199
200 async def get_library_artists(self) -> AsyncGenerator[Artist, None]:
201 """Retrieve all library artists from Tidal."""
202 async for item in self.library.get_artists():
203 yield item
204
205 async def get_library_albums(self) -> AsyncGenerator[Album, None]:
206 """Retrieve all library albums from Tidal."""
207 async for item in self.library.get_albums():
208 yield item
209
210 async def get_library_tracks(self) -> AsyncGenerator[Track, None]:
211 """Retrieve library tracks from Tidal."""
212 async for item in self.library.get_tracks():
213 yield item
214
215 async def get_library_playlists(self) -> AsyncGenerator[Playlist, None]:
216 """Retrieve all library playlists from the provider."""
217 async for item in self.library.get_playlists():
218 yield item
219
220 async def library_add(self, item: MediaItemType) -> bool:
221 """Add item to library."""
222 return await self.library.add_item(item)
223
224 async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
225 """Remove item from library."""
226 return await self.library.remove_item(prov_item_id, media_type)
227
228 async def create_playlist(self, name: str) -> Playlist:
229 """Create a new playlist on provider with given name."""
230 return await self.playlists.create(name)
231
232 async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
233 """Add track(s) to playlist."""
234 await self.playlists.add_tracks(prov_playlist_id, prov_track_ids)
235
236 async def remove_playlist_tracks(
237 self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
238 ) -> None:
239 """Remove track(s) from playlist."""
240 await self.playlists.remove_tracks(prov_playlist_id, positions_to_remove)
241
242 @use_cache(expiration=3600, category=CACHE_CATEGORY_RECOMMENDATIONS)
243 async def recommendations(self) -> list[RecommendationFolder]:
244 """Get this provider's recommendations organized into folders."""
245 return await self.recommendations_manager.get_recommendations()
246