/
/
/
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.config_entries import ConfigEntry, ConfigValueOption
9from music_assistant_models.enums import ConfigEntryType
10from music_assistant_models.errors import (
11 LoginFailed,
12)
13from music_assistant_models.media_items import (
14 Album,
15 Artist,
16 MediaItemType,
17 Playlist,
18 RecommendationFolder,
19 SearchResults,
20 Track,
21)
22
23from music_assistant.constants import (
24 CONF_ENTRY_UNOFFICIAL_PROVIDER,
25 CONF_PASSWORD,
26 CONF_USERNAME,
27)
28from music_assistant.controllers.cache import use_cache
29from music_assistant.models.music_provider import MusicProvider
30from music_assistant.models.recommendation_payload import RecommendationPayloadMixin
31from music_assistant.providers.yousee.api_client import YouSeeAPIClient
32from music_assistant.providers.yousee.auth_manager import YouSeeAuthManager
33from music_assistant.providers.yousee.constants import CONF_QUALITY
34from music_assistant.providers.yousee.library import YouSeeLibraryManager
35from music_assistant.providers.yousee.media import YouSeeMediaManager
36from music_assistant.providers.yousee.playlist import YouSeePlaylistManager
37from music_assistant.providers.yousee.recommendations import YouSeeRecommendationsManager
38from music_assistant.providers.yousee.streaming import YouSeeStreamingManager
39
40if TYPE_CHECKING:
41 from music_assistant_models.enums import (
42 MediaType,
43 )
44 from music_assistant_models.media_items import (
45 Album,
46 Artist,
47 BrowseFolder,
48 ItemMapping,
49 MediaItemType,
50 Playlist,
51 RecommendationFolder,
52 SearchResults,
53 Track,
54 UniqueList,
55 )
56 from music_assistant_models.streamdetails import StreamDetails
57
58
59class YouSeeMusikProvider(RecommendationPayloadMixin, MusicProvider):
60 """Provider implementation for YouSee Musik."""
61
62 # the personalized sections barely change intraday; keep the pre-refactor 24h interval
63 recommendation_payload_ttl = 3600 * 24
64
65 auth: YouSeeAuthManager
66
67 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
68 """Return Config entries to setup this provider."""
69 return (
70 CONF_ENTRY_UNOFFICIAL_PROVIDER,
71 ConfigEntry(
72 key=CONF_QUALITY,
73 type=ConfigEntryType.INTEGER,
74 default_value=320,
75 options=[
76 ConfigValueOption(320),
77 ConfigValueOption(192),
78 ],
79 ),
80 )
81
82 async def handle_async_init(self) -> None:
83 """Handle async initialization of the provider."""
84 if not self.get_setup_value(CONF_USERNAME) or not self.get_setup_value(CONF_PASSWORD):
85 msg = "Invalid login credentials"
86 raise LoginFailed(msg)
87 # try to get a token, raise if that fails
88 self.auth = YouSeeAuthManager(self)
89 self.api = YouSeeAPIClient(self)
90 self.library = YouSeeLibraryManager(self)
91 self.media = YouSeeMediaManager(self)
92 self.playlist = YouSeePlaylistManager(self)
93 self.streaming = YouSeeStreamingManager(self)
94 self.recommendations_manager = YouSeeRecommendationsManager(self)
95
96 token = await self.auth.auth_token()
97 if not token:
98 msg = f"Login failed for user {self.get_setup_value(CONF_USERNAME)}"
99 raise LoginFailed(msg)
100
101 async def search(
102 self,
103 search_query: str,
104 media_types: list[MediaType],
105 limit: int = 5,
106 ) -> SearchResults:
107 """
108 Perform search on musicprovider.
109
110 :param search_query: Search query.
111 :param media_types: A list of media_types to include.
112 :param limit: Number of items to return in the search (per type).
113 """
114 return await self.media.search(search_query, media_types, limit)
115
116 async def get_library_artists(self) -> AsyncGenerator[Artist]:
117 """Retrieve library artists from the provider."""
118 async for artist in self.library.get_artists():
119 yield artist
120
121 async def get_library_albums(self) -> AsyncGenerator[Album]:
122 """Retrieve library albums from the provider."""
123 async for album in self.library.get_albums():
124 yield album
125
126 async def get_library_tracks(self) -> AsyncGenerator[Track]:
127 """Retrieve library tracks from the provider."""
128 async for track in self.library.get_tracks():
129 yield track
130
131 async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
132 """Retrieve library/subscribed playlists from the provider."""
133 async for playlist in self.library.get_playlists():
134 yield playlist
135
136 @use_cache(3600 * 24 * 30) # Cache for 30 days
137 async def get_artist(self, prov_artist_id: str) -> Artist:
138 """Get full artist details by id."""
139 return await self.media.get_artist(prov_artist_id)
140
141 @use_cache(3600 * 24 * 14, allow_expired_cache=True) # Cache for 14 days
142 async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
143 """Get a list of all albums for the given artist."""
144 return await self.media.get_artist_albums(prov_artist_id)
145
146 @use_cache(3600 * 24 * 14, allow_expired_cache=True) # Cache for 14 days
147 async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
148 """Get a list of most popular tracks for the given artist."""
149 return await self.media.get_artist_toptracks(prov_artist_id)
150
151 @use_cache(3600 * 24 * 30) # Cache for 30 days
152 async def get_album(self, prov_album_id: str) -> Album:
153 """Get full album details by id."""
154 return await self.media.get_album(prov_album_id)
155
156 @use_cache(3600 * 24 * 30) # Cache for 30 days
157 async def get_track(self, prov_track_id: str) -> Track:
158 """Get full track details by id."""
159 return await self.media.get_track(prov_track_id)
160
161 @use_cache(3600 * 24 * 30) # Cache for 30 days
162 async def get_playlist(self, prov_playlist_id: str) -> Playlist:
163 """Get full playlist details by id."""
164 return await self.media.get_playlist(prov_playlist_id)
165
166 @use_cache(3600 * 24 * 30, allow_expired_cache=True) # Cache for 30 days
167 async def get_album_tracks(
168 self,
169 prov_album_id: str,
170 ) -> list[Track]:
171 """Get album tracks for given album id."""
172 return await self.media.get_album_tracks(prov_album_id)
173
174 @use_cache(3600 * 3, allow_expired_cache=True) # Cache for 3 hours
175 async def get_playlist_tracks(
176 self,
177 prov_playlist_id: str,
178 page: int = 0,
179 ) -> list[Track]:
180 """Get all playlist tracks for given playlist id."""
181 return await self.media.get_playlist_tracks(prov_playlist_id, page)
182
183 async def library_add(self, item: MediaItemType) -> bool:
184 """Add item to provider's library. Return true on success."""
185 return await self.library.add_item(item)
186
187 async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
188 """Remove item from provider's library. Return true on success."""
189 return await self.library.remove_item(prov_item_id, media_type)
190
191 async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
192 """Add track(s) to playlist."""
193 return await self.playlist.add_tracks(prov_playlist_id, prov_track_ids)
194
195 async def remove_playlist_tracks(
196 self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
197 ) -> None:
198 """Remove track(s) from playlist."""
199 return await self.playlist.remove_tracks(prov_playlist_id, positions_to_remove)
200
201 async def create_playlist(self, name: str, media_types: set[MediaType]) -> Playlist:
202 """Create a new playlist on provider with given name."""
203 return await self.playlist.create(name)
204
205 @use_cache(3600 * 24, allow_expired_cache=True) # Cache for 24 hours
206 async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
207 """Retrieve a dynamic list of similar tracks based on the provided track."""
208 return await self.media.get_similar_tracks(prov_track_id, limit)
209
210 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
211 """Get streamdetails for a track."""
212 return await self.streaming.get_stream_details(item_id, media_type)
213
214 async def on_streamed(
215 self,
216 streamdetails: StreamDetails,
217 ) -> None:
218 """
219 Handle callback when given streamdetails completed streaming.
220
221 To get the number of seconds streamed, see streamdetails.seconds_streamed.
222 To get the number of seconds seeked/skipped, see streamdetails.seek_position.
223 Note that seconds_streamed is the total streamed seconds, so without seeked time.
224
225 NOTE: Due to internal and player buffering,
226 this may be called in advance of the actual completion.
227 """
228 await self.streaming.report_playback(
229 streamdetails,
230 )
231
232 async def get_recommendations(self) -> list[RecommendationFolder]:
233 """Get this provider's available recommendation rows, without items."""
234 return await self._recommendation_rows_from_payload()
235
236 async def get_recommendation_items(
237 self, item_id: str
238 ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
239 """
240 Get the items for a single recommendation row.
241
242 :param item_id: The item_id of the row, as returned by get_recommendations.
243 """
244 return await self._recommendation_items_from_payload(item_id)
245
246 async def _fetch_recommendation_payload(self) -> list[RecommendationFolder]:
247 """Fetch and parse the full recommendations payload (folders WITH items)."""
248 return await self.recommendations_manager.get_recommendations()
249