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.config_entries import ConfigEntry, ConfigValueOption
10from music_assistant_models.enums import ConfigEntryType, MediaType, ProviderFeature
11from music_assistant_models.errors import LoginFailed
12from music_assistant_models.media_items import (
13    Album,
14    Artist,
15    BrowseFolder,
16    ItemMapping,
17    MediaItemType,
18    Playlist,
19    RecommendationFolder,
20    SearchResults,
21    Track,
22    UniqueList,
23)
24
25from music_assistant.constants import CONF_ENTRY_UNOFFICIAL_PROVIDER
26from music_assistant.controllers.cache import use_cache
27from music_assistant.models.music_provider import MusicProvider
28from music_assistant.models.recommendation_payload import RecommendationPayloadMixin
29
30from .api_client import TidalAPIClient
31from .auth_manager import TidalAuthManager
32from .constants import (
33    CONF_AUTH_TOKEN,
34    CONF_EXPIRY_TIME,
35    CONF_QUALITY,
36    CONF_REFRESH_TOKEN,
37    CONF_USER_ID,
38)
39from .library import TidalLibraryManager
40from .media import TidalMediaManager
41from .playlist import TidalPlaylistManager
42from .recommendations import TidalRecommendationManager
43from .streaming import TidalStreamingManager
44
45if TYPE_CHECKING:
46    from collections.abc import AsyncGenerator
47
48    from music_assistant_models.config_entries import ProviderConfig
49    from music_assistant_models.provider import ProviderManifest
50    from music_assistant_models.streamdetails import StreamDetails
51
52    from music_assistant.mass import MusicAssistant
53
54
55SUPPORTED_FEATURES = {
56    ProviderFeature.LIBRARY_ARTISTS,
57    ProviderFeature.LIBRARY_ALBUMS,
58    ProviderFeature.LIBRARY_TRACKS,
59    ProviderFeature.LIBRARY_PLAYLISTS,
60    ProviderFeature.ARTIST_ALBUMS,
61    ProviderFeature.ARTIST_TOPTRACKS,
62    ProviderFeature.SEARCH,
63    ProviderFeature.LIBRARY_ARTISTS_EDIT,
64    ProviderFeature.LIBRARY_ALBUMS_EDIT,
65    ProviderFeature.LIBRARY_TRACKS_EDIT,
66    ProviderFeature.LIBRARY_PLAYLISTS_EDIT,
67    ProviderFeature.PLAYLIST_CREATE,
68    ProviderFeature.SIMILAR_TRACKS,
69    ProviderFeature.BROWSE,
70    ProviderFeature.PLAYLIST_TRACKS_EDIT,
71    ProviderFeature.RECOMMENDATIONS,
72    ProviderFeature.LYRICS,
73}
74
75
76class TidalProvider(RecommendationPayloadMixin, MusicProvider):
77    """Implementation of a Tidal MusicProvider."""
78
79    def __init__(self, mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig):
80        """Initialize Tidal provider."""
81        super().__init__(mass, manifest, config, SUPPORTED_FEATURES)
82        self.auth = TidalAuthManager(
83            http_session=mass.http_session,
84            config_updater=self._update_auth_config,
85            logger=self.logger,
86        )
87        self.api = TidalAPIClient(self)
88        self.library = TidalLibraryManager(self)
89        self.media = TidalMediaManager(self)
90        self.playlists = TidalPlaylistManager(self)
91        self.recommendations_manager = TidalRecommendationManager(self)
92        self.streaming = TidalStreamingManager(self)
93
94    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
95        """
96        Return the configuration (options) entries for the Tidal provider.
97
98        Authentication runs in the interactive setup flow (see ``setup_flow.py``); the only
99        genuine option configured here is the preferred streaming quality.
100        """
101        return (
102            CONF_ENTRY_UNOFFICIAL_PROVIDER,
103            ConfigEntry(
104                key=CONF_QUALITY,
105                type=ConfigEntryType.STRING,
106                required=True,
107                options=[
108                    ConfigValueOption("LOSSLESS"),
109                    ConfigValueOption("HI_RES_LOSSLESS"),
110                ],
111                default_value="HI_RES_LOSSLESS",
112            ),
113        )
114
115    def _update_auth_config(self, auth_info: dict[str, Any]) -> None:
116        """Update the persisted auth setup data with new (rotated) auth info."""
117        self._update_setup_data(CONF_AUTH_TOKEN, auth_info["access_token"])
118        self._update_setup_data(CONF_REFRESH_TOKEN, auth_info["refresh_token"])
119        self._update_setup_data(CONF_EXPIRY_TIME, auth_info["expires_at"])
120        self._update_setup_data(CONF_USER_ID, auth_info["userId"])
121
122    async def handle_async_init(self) -> None:
123        """Handle async initialization of the provider."""
124        access_token = self.get_setup_value(CONF_AUTH_TOKEN)
125        refresh_token = self.get_setup_value(CONF_REFRESH_TOKEN)
126        expires_at = self.get_setup_value(CONF_EXPIRY_TIME)
127        user_id = self.get_setup_value(CONF_USER_ID)
128
129        if not access_token or not refresh_token:
130            raise LoginFailed("Missing authentication data")
131
132        if isinstance(expires_at, str) and "T" in expires_at:
133            try:
134                dt = datetime.fromisoformat(expires_at)
135                expires_at = dt.timestamp()
136                self._update_setup_data(CONF_EXPIRY_TIME, expires_at)
137            except ValueError:
138                expires_at = 0
139
140        auth_data = {
141            "access_token": access_token,
142            "refresh_token": refresh_token,
143            "expires_at": expires_at,
144            "userId": user_id,
145        }
146
147        if not await self.auth.initialize(json.dumps(auth_data)):
148            raise LoginFailed("Failed to authenticate with Tidal")
149
150        user_info = await self.api.get("sessions")
151        logged_in_user = await self.get_user(str(user_info.get("userId")))
152        await self.auth.update_user_info(logged_in_user, str(user_info.get("sessionId")))
153
154    async def get_user(self, prov_user_id: str) -> dict[str, Any]:
155        """Get user information."""
156        return await self.api.get(f"users/{prov_user_id}")
157
158    @use_cache(3600 * 24 * 14)
159    async def search(
160        self, search_query: str, media_types: list[MediaType], limit: int = 5
161    ) -> SearchResults:
162        """Perform search on musicprovider."""
163        return await self.media.search(search_query, media_types, limit)
164
165    @use_cache(3600 * 24, allow_expired_cache=True)
166    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
167        """Get similar tracks for given track id."""
168        return await self.media.get_similar_tracks(prov_track_id, limit)
169
170    @use_cache(3600 * 24 * 30)
171    async def get_artist(self, prov_artist_id: str) -> Artist:
172        """Get artist details for given artist id."""
173        return await self.media.get_artist(prov_artist_id)
174
175    @use_cache(3600 * 24 * 30)
176    async def get_album(self, prov_album_id: str) -> Album:
177        """Get album details for given album id."""
178        return await self.media.get_album(prov_album_id)
179
180    @use_cache(3600 * 24 * 30)
181    async def get_track(self, prov_track_id: str) -> Track:
182        """Get track details for given track id."""
183        return await self.media.get_track(prov_track_id)
184
185    @use_cache(3600 * 24 * 30)
186    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
187        """Get playlist details for given playlist id."""
188        return await self.media.get_playlist(prov_playlist_id)
189
190    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
191    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
192        """Get album tracks for given album id."""
193        return await self.media.get_album_tracks(prov_album_id)
194
195    @use_cache(3600 * 24 * 7, allow_expired_cache=True)
196    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
197        """Get a list of all albums for the given artist."""
198        return await self.media.get_artist_albums(prov_artist_id)
199
200    @use_cache(3600 * 24 * 7, allow_expired_cache=True)
201    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
202        """Get a list of 10 most popular tracks for the given artist."""
203        return await self.media.get_artist_toptracks(prov_artist_id)
204
205    @use_cache(3600 * 3, allow_expired_cache=True)
206    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
207        """Get playlist tracks."""
208        return await self.media.get_playlist_tracks(prov_playlist_id, page)
209
210    async def get_stream_details(
211        self, item_id: str, media_type: MediaType = MediaType.TRACK
212    ) -> StreamDetails:
213        """Return the content details for the given track when it will be streamed."""
214        return await self.streaming.get_stream_details(item_id)
215
216    def get_item_mapping(self, media_type: MediaType, key: str, name: str) -> ItemMapping:
217        """Create a generic item mapping."""
218        return ItemMapping(
219            media_type=media_type,
220            item_id=key,
221            provider=self.instance_id,
222            name=name,
223        )
224
225    async def get_library_artists(self) -> AsyncGenerator[Artist]:
226        """Retrieve all library artists from Tidal."""
227        async for item in self.library.get_artists():
228            yield item
229
230    async def get_library_albums(self) -> AsyncGenerator[Album]:
231        """Retrieve all library albums from Tidal."""
232        async for item in self.library.get_albums():
233            yield item
234
235    async def get_library_tracks(self) -> AsyncGenerator[Track]:
236        """Retrieve library tracks from Tidal."""
237        async for item in self.library.get_tracks():
238            yield item
239
240    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
241        """Retrieve all library playlists from the provider."""
242        async for item in self.library.get_playlists():
243            yield item
244
245    async def library_add(self, item: MediaItemType) -> bool:
246        """Add item to library."""
247        return await self.library.add_item(item)
248
249    async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
250        """Remove item from library."""
251        return await self.library.remove_item(prov_item_id, media_type)
252
253    async def create_playlist(self, name: str, media_types: set[MediaType]) -> Playlist:
254        """Create a new playlist on provider with given name."""
255        return await self.playlists.create(name)
256
257    async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
258        """Add track(s) to playlist."""
259        await self.playlists.add_tracks(prov_playlist_id, prov_track_ids)
260
261    async def remove_playlist_tracks(
262        self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
263    ) -> None:
264        """Remove track(s) from playlist."""
265        await self.playlists.remove_tracks(prov_playlist_id, positions_to_remove)
266
267    async def get_recommendations(self) -> list[RecommendationFolder]:
268        """Get this provider's available recommendation rows, without items."""
269        return await self._recommendation_rows_from_payload()
270
271    async def get_recommendation_items(
272        self, item_id: str
273    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
274        """
275        Get the items for a single recommendation row.
276
277        :param item_id: The item_id of the row, as returned by get_recommendations.
278        """
279        return await self._recommendation_items_from_payload(item_id)
280
281    async def _fetch_recommendation_payload(self) -> list[RecommendationFolder]:
282        """Fetch and parse the full recommendations payload (folders WITH items)."""
283        return await self.recommendations_manager.get_recommendations()
284