music-assistant-server

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