music-assistant-server

1"""YouSee Musik musicprovider support for MusicAssistant."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator
6from typing import TYPE_CHECKING
7
8from music_assistant_models.errors import (
9    LoginFailed,
10)
11from music_assistant_models.media_items import (
12    Album,
13    Artist,
14    MediaItemType,
15    Playlist,
16    RecommendationFolder,
17    SearchResults,
18    Track,
19)
20
21from music_assistant.constants import (
22    CONF_PASSWORD,
23    CONF_USERNAME,
24)
25from music_assistant.controllers.cache import use_cache
26from music_assistant.models.music_provider import MusicProvider
27from music_assistant.providers.yousee.api_client import YouSeeAPIClient
28from music_assistant.providers.yousee.auth_manager import YouSeeAuthManager
29from music_assistant.providers.yousee.library import YouSeeLibraryManager
30from music_assistant.providers.yousee.media import YouSeeMediaManager
31from music_assistant.providers.yousee.playlist import YouSeePlaylistManager
32from music_assistant.providers.yousee.recommendations import YouSeeRecommendationsManager
33from music_assistant.providers.yousee.streaming import YouSeeStreamingManager
34
35if TYPE_CHECKING:
36    from music_assistant_models.enums import (
37        MediaType,
38    )
39    from music_assistant_models.media_items import (
40        Album,
41        Artist,
42        MediaItemType,
43        Playlist,
44        RecommendationFolder,
45        SearchResults,
46        Track,
47    )
48    from music_assistant_models.streamdetails import StreamDetails
49
50
51class YouSeeMusikProvider(MusicProvider):
52    """Provider implementation for YouSee Musik."""
53
54    auth: YouSeeAuthManager
55
56    async def handle_async_init(self) -> None:
57        """Handle async initialization of the provider."""
58        if not self.config.get_value(CONF_USERNAME) or not self.config.get_value(CONF_PASSWORD):
59            msg = "Invalid login credentials"
60            raise LoginFailed(msg)
61        # try to get a token, raise if that fails
62        self.auth = YouSeeAuthManager(self)
63        self.api = YouSeeAPIClient(self)
64        self.library = YouSeeLibraryManager(self)
65        self.media = YouSeeMediaManager(self)
66        self.playlist = YouSeePlaylistManager(self)
67        self.streaming = YouSeeStreamingManager(self)
68        self.recommendations_manager = YouSeeRecommendationsManager(self)
69
70        token = await self.auth.auth_token()
71        if not token:
72            msg = f"Login failed for user {self.config.get_value(CONF_USERNAME)}"
73            raise LoginFailed(msg)
74
75    async def search(
76        self,
77        search_query: str,
78        media_types: list[MediaType],
79        limit: int = 5,
80    ) -> SearchResults:
81        """Perform search on musicprovider.
82
83        :param search_query: Search query.
84        :param media_types: A list of media_types to include.
85        :param limit: Number of items to return in the search (per type).
86        """
87        return await self.media.search(search_query, media_types, limit)
88
89    async def get_library_artists(self) -> AsyncGenerator[Artist, None]:
90        """Retrieve library artists from the provider."""
91        async for artist in self.library.get_artists():
92            yield artist
93
94    async def get_library_albums(self) -> AsyncGenerator[Album, None]:
95        """Retrieve library albums from the provider."""
96        async for album in self.library.get_albums():
97            yield album
98
99    async def get_library_tracks(self) -> AsyncGenerator[Track, None]:
100        """Retrieve library tracks from the provider."""
101        async for track in self.library.get_tracks():
102            yield track
103
104    async def get_library_playlists(self) -> AsyncGenerator[Playlist, None]:
105        """Retrieve library/subscribed playlists from the provider."""
106        async for playlist in self.library.get_playlists():
107            yield playlist
108
109    @use_cache(3600 * 24 * 30)  # Cache for 30 days
110    async def get_artist(self, prov_artist_id: str) -> Artist:
111        """Get full artist details by id."""
112        return await self.media.get_artist(prov_artist_id)
113
114    @use_cache(3600 * 24 * 14)  # Cache for 14 days
115    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
116        """Get a list of all albums for the given artist."""
117        return await self.media.get_artist_albums(prov_artist_id)
118
119    @use_cache(3600 * 24 * 14)  # Cache for 14 days
120    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
121        """Get a list of most popular tracks for the given artist."""
122        return await self.media.get_artist_toptracks(prov_artist_id)
123
124    @use_cache(3600 * 24 * 30)  # Cache for 30 days
125    async def get_album(self, prov_album_id: str) -> Album:
126        """Get full album details by id."""
127        return await self.media.get_album(prov_album_id)
128
129    @use_cache(3600 * 24 * 30)  # Cache for 30 days
130    async def get_track(self, prov_track_id: str) -> Track:
131        """Get full track details by id."""
132        return await self.media.get_track(prov_track_id)
133
134    @use_cache(3600 * 24 * 30)  # Cache for 30 days
135    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
136        """Get full playlist details by id."""
137        return await self.media.get_playlist(prov_playlist_id)
138
139    @use_cache(3600 * 24 * 30)  # Cache for 30 days
140    async def get_album_tracks(
141        self,
142        prov_album_id: str,
143    ) -> list[Track]:
144        """Get album tracks for given album id."""
145        return await self.media.get_album_tracks(prov_album_id)
146
147    @use_cache(3600 * 3)  # Cache for 3 hours
148    async def get_playlist_tracks(
149        self,
150        prov_playlist_id: str,
151        page: int = 0,
152    ) -> list[Track]:
153        """Get all playlist tracks for given playlist id."""
154        return await self.media.get_playlist_tracks(prov_playlist_id, page)
155
156    async def library_add(self, item: MediaItemType) -> bool:
157        """Add item to provider's library. Return true on success."""
158        return await self.library.add_item(item)
159
160    async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
161        """Remove item from provider's library. Return true on success."""
162        return await self.library.remove_item(prov_item_id, media_type)
163
164    async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
165        """Add track(s) to playlist."""
166        return await self.playlist.add_tracks(prov_playlist_id, prov_track_ids)
167
168    async def remove_playlist_tracks(
169        self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
170    ) -> None:
171        """Remove track(s) from playlist."""
172        return await self.playlist.remove_tracks(prov_playlist_id, positions_to_remove)
173
174    async def create_playlist(self, name: str) -> Playlist:
175        """Create a new playlist on provider with given name."""
176        return await self.playlist.create(name)
177
178    @use_cache(3600 * 24)  # Cache for 24 hours
179    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
180        """Retrieve a dynamic list of similar tracks based on the provided track."""
181        return await self.media.get_similar_tracks(prov_track_id, limit)
182
183    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
184        """Get streamdetails for a track."""
185        return await self.streaming.get_stream_details(item_id, media_type)
186
187    async def on_streamed(
188        self,
189        streamdetails: StreamDetails,
190    ) -> None:
191        """
192        Handle callback when given streamdetails completed streaming.
193
194        To get the number of seconds streamed, see streamdetails.seconds_streamed.
195        To get the number of seconds seeked/skipped, see streamdetails.seek_position.
196        Note that seconds_streamed is the total streamed seconds, so without seeked time.
197
198        NOTE: Due to internal and player buffering,
199        this may be called in advance of the actual completion.
200        """
201        await self.streaming.report_playback(
202            streamdetails,
203        )
204
205    @use_cache(3600 * 24)  # Cache for 1 day
206    async def recommendations(self) -> list[RecommendationFolder]:
207        """
208        Get this provider's recommendations.
209
210        Returns an actual (and often personalised) list of recommendations
211        from this provider for the user/account.
212        """
213        return await self.recommendations_manager.get_recommendations()
214