/
/
/
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