/
/
/
1"""Media retrieval operations for YouSee Musik."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6
7from music_assistant_models.enums import (
8 MediaType,
9)
10from music_assistant_models.errors import MediaNotFoundError
11from music_assistant_models.media_items import Album, Artist, Playlist, SearchResults, Track
12
13from music_assistant.providers.yousee.api_client import JsonLike
14from music_assistant.providers.yousee.constants import (
15 GET_POPULAR_TRACKS_LIMIT,
16 IMAGE_SIZE,
17)
18from music_assistant.providers.yousee.parsers import (
19 parse_album,
20 parse_artist,
21 parse_lyrics,
22 parse_playlist,
23 parse_track,
24)
25
26if TYPE_CHECKING:
27 from music_assistant.providers.yousee.provider import YouSeeMusikProvider
28
29
30class YouSeeMediaManager:
31 """Handles retrieval of media items from YouSee Musik."""
32
33 def __init__(self, provider: YouSeeMusikProvider):
34 """Initialize media retriever."""
35 self.provider = provider
36 self.api = provider.api
37 self.logger = provider.logger
38
39 async def search(
40 self,
41 search_query: str,
42 media_types: list[MediaType],
43 limit: int = 5,
44 ) -> SearchResults:
45 """
46 Perform search on musicprovider.
47
48 :param search_query: Search query.
49 :param media_types: A list of media_types to include.
50 :param limit: Number of items to return in the search (per type).
51 """
52 sections = {
53 MediaType.TRACK: """
54 tracks(first: $first) {
55 items {
56 id
57 title
58 availableToStream
59 album {
60 id
61 title
62 }
63 artist {
64 id
65 title
66 cover(size: $imageSize)
67 }
68 cover(size: $imageSize)
69 duration
70 share
71 genre
72 isrc
73 featuredArtists {
74 items {
75 id
76 title
77 cover(size: $imageSize)
78 }
79 }
80 }
81 }
82 """,
83 MediaType.ALBUM: """
84 albums(first: $first) {
85 items {
86 id
87 title
88 cover(size: $imageSize)
89 artist {
90 id
91 title
92 cover(size: $imageSize)
93 }
94 }
95 }
96 """,
97 MediaType.ARTIST: """
98 artists(first: $first) {
99 items {
100 id
101 title
102 cover(size: $imageSize)
103 share
104 }
105 }
106 """,
107 MediaType.PLAYLIST: """
108 playlists(first: $first) {
109 items {
110 id
111 title
112 isOwned
113 share
114 cover(size: $imageSize)
115 description
116 }
117 }
118 """,
119 }
120
121 search_result = SearchResults()
122
123 media_types = [x for x in media_types if x in (sections)]
124
125 if not media_types:
126 return search_result
127
128 query = """
129 query searchMixedSections($criterion: String!, $imageSize: Int = 512, $first: Int = 5) {
130 search(criterion: $criterion) {
131 TRACK_SECTION
132 ALBUM_SECTION
133 PLAYLIST_SECTION
134 ARTIST_SECTION
135 }
136 }
137 """
138 for media_type, section in sections.items():
139 if media_type in media_types:
140 query = query.replace(f"{media_type.name}_SECTION", section)
141 else:
142 query = query.replace(f"{media_type.name}_SECTION", "")
143
144 variables = {
145 "criterion": search_query,
146 "imageSize": IMAGE_SIZE,
147 "first": limit,
148 }
149
150 result = await self.api.post_graphql(query, variables)
151
152 result = result.get("data", {}).get("search", {})
153
154 if not result:
155 return search_result
156
157 if "artists" in result:
158 search_result.artists = [
159 parse_artist(self.provider, item) for item in result["artists"].get("items", [])
160 ]
161 if "albums" in result:
162 search_result.albums = [
163 await parse_album(self.provider, item) for item in result["albums"].get("items", [])
164 ]
165 if "tracks" in result:
166 search_result.tracks = [
167 await parse_track(self.provider, item) for item in result["tracks"].get("items", [])
168 ]
169 if "playlists" in result:
170 search_result.playlists = [
171 await parse_playlist(self.provider, item)
172 for item in result["playlists"].get("items", [])
173 ]
174
175 return search_result
176
177 async def get_artist(self, prov_artist_id: str) -> Artist:
178 """Get full artist details by id."""
179 query = """
180 query Catalog($id: ID!, $imageSize: Int = 512) {
181 catalog {
182 artist(id: $id) {
183 id
184 title
185 cover(size: $imageSize)
186 share
187 }
188 }
189 }
190 """
191 variables = {"id": prov_artist_id, "imageSize": IMAGE_SIZE}
192
193 result = await self.api.post_graphql(query, variables)
194 if not result or not result.get("data", {}).get("catalog", {}).get("artist"):
195 raise MediaNotFoundError(f"Artist {prov_artist_id} not found")
196 return parse_artist(self.provider, result["data"]["catalog"]["artist"])
197
198 async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
199 """Get a list of all albums for the given artist."""
200 query = """
201 query Catalog($id: ID!, $imageSize: Int = 512, $first: Int = 50, $after: String) {
202 catalog {
203 artist(id: $id) {
204 id
205 albums(first: $first, after: $after) {
206 totalCount
207 pageInfo {
208 hasNextPage
209 endCursor
210 }
211 items {
212 id
213 title
214 cover(size: $imageSize)
215 }
216 }
217 }
218 }
219 }
220 """
221
222 albums = []
223 variables = {
224 "id": prov_artist_id,
225 "imageSize": IMAGE_SIZE,
226 }
227
228 async for item in self.api.paginate_graphql(
229 query,
230 variables,
231 ["data", "catalog", "artist", "albums"],
232 ):
233 albums.append(await parse_album(self.provider, item))
234
235 return albums
236
237 async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
238 """Get a list of most popular tracks for the given artist."""
239 query = """
240 query Catalog($id: ID!, $imageSize: Int = 512, $first: Int = 25) {
241 catalog {
242 artist(id: $id) {
243 id
244 title
245 cover(size: $imageSize)
246 share
247 tracks(first: $first, after: null, orderBy: POPULARITY) {
248 items {
249 id
250 title
251 cover(size: $imageSize)
252 isrc
253 duration
254 label
255 artist {
256 id
257 title
258 cover(size: $imageSize)
259 }
260 featuredArtists {
261 items {
262 id
263 title
264 cover(size: $imageSize)
265 }
266 }
267 share
268 genre
269 }
270 }
271 }
272 }
273 }
274 """
275
276 variables = {
277 "id": prov_artist_id,
278 "imageSize": IMAGE_SIZE,
279 "first": GET_POPULAR_TRACKS_LIMIT,
280 }
281
282 result = await self.api.post_graphql(query, variables)
283
284 if not result or not result.get("data", {}).get("catalog", {}).get("artist"):
285 raise MediaNotFoundError(f"Artist {prov_artist_id} not found")
286 tracks = []
287
288 for item in result["data"]["catalog"]["artist"]["tracks"]["items"]:
289 tracks.append(await parse_track(self.provider, item))
290
291 return tracks
292
293 async def get_album(self, prov_album_id: str) -> Album:
294 """Get full album details by id."""
295 query = """
296 query Catalog($id: ID!, $imageSize: Int = 512) {
297 catalog {
298 album(id: $id) {
299 id
300 title
301 tracksCount
302 genre
303 label
304 releaseDate
305 available
306 upc
307 type
308 share
309 cover(size: $imageSize)
310 artist {
311 id
312 title
313 cover(size: $imageSize)
314 }
315 featuredArtists {
316 items {
317 id
318 title
319 cover(size: $imageSize)
320 }
321 }
322 }
323 }
324 }
325 """
326 variables = {"id": prov_album_id, "imageSize": IMAGE_SIZE}
327
328 result = await self.api.post_graphql(query, variables)
329 if not result or not result.get("data", {}).get("catalog", {}).get("album"):
330 raise MediaNotFoundError(f"Album {prov_album_id} not found")
331 return await parse_album(self.provider, result["data"]["catalog"]["album"])
332
333 async def _get_lyrics(self, prov_track_id: str) -> list[JsonLike]:
334 """Attempt to retrieve lyrics for the given track id."""
335 query = """
336 query Lyric($id: ID!, $first: Int = 50, $after: String) {
337 catalog {
338 track(id: $id) {
339 lyrics {
340 lrc(first: $first, after: $after) {
341 pageInfo {
342 hasNextPage
343 endCursor
344 }
345 items {
346 startInMs
347 durationInMs
348 line
349 }
350 }
351 }
352 }
353 }
354 }
355 """
356 variables = {"id": prov_track_id}
357
358 lines = []
359
360 async for line in self.api.paginate_graphql(
361 query, variables, ["data", "catalog", "track", "lyrics", "lrc"]
362 ):
363 lines.append(line)
364
365 return lines
366
367 async def get_track(self, prov_track_id: str) -> Track:
368 """Get full track details by id."""
369 query = """
370 query getTrack($id: ID!, $imageSize: Int = 512) {
371 catalog {
372 track(id: $id) {
373 id
374 title
375 duration
376 genre
377 label
378 releaseDate
379 availableToStream
380 isrc
381 share
382 cover(size: $imageSize)
383 lyrics {
384 id
385 }
386 album {
387 id
388 title
389 }
390 artist {
391 id
392 title
393 cover(size: $imageSize)
394 }
395 featuredArtists {
396 items {
397 id
398 title
399 cover(size: $imageSize)
400 }
401 }
402 }
403 }
404 }
405 """
406 variables = {"id": prov_track_id, "imageSize": IMAGE_SIZE}
407
408 result = await self.api.post_graphql(query, variables)
409 if not result or not result.get("data", {}).get("catalog", {}).get("track"):
410 raise MediaNotFoundError(f"Track {prov_track_id} not found")
411
412 track = await parse_track(self.provider, result["data"]["catalog"]["track"])
413
414 if result["data"]["catalog"]["track"].get("lyrics"):
415 lyrics = await self._get_lyrics(prov_track_id)
416 parsed_lyrics, parsed_lrc_lyrics = await parse_lyrics(lyrics)
417
418 if parsed_lyrics:
419 self.logger.debug("Attached lyrics to track")
420 track.metadata.lyrics = parsed_lyrics
421 if parsed_lrc_lyrics:
422 self.logger.debug("Attached LRC lyrics to track")
423 track.metadata.lrc_lyrics = parsed_lrc_lyrics
424
425 return track
426
427 async def get_playlist(self, prov_playlist_id: str) -> Playlist:
428 """Get full playlist details by id."""
429 query = """
430 query getPlaylist($id: ID!, $imageSize: Int = 512) {
431 playlists {
432 playlist(id: $id) {
433 id
434 title
435 description
436 tracksCount
437 createdAt
438 isOwned
439 share
440 cover(size: $imageSize)
441 }
442 }
443 }
444 """
445 variables = {"id": prov_playlist_id, "imageSize": IMAGE_SIZE}
446
447 result = await self.api.post_graphql(query, variables)
448 if not result or not result.get("data", {}).get("playlists", {}).get("playlist"):
449 raise MediaNotFoundError(f"Playlist {prov_playlist_id} not found")
450
451 return await parse_playlist(self.provider, result["data"]["playlists"]["playlist"])
452
453 async def get_album_tracks(
454 self,
455 prov_album_id: str,
456 ) -> list[Track]:
457 """Get album tracks for given album id."""
458 query = """
459 query GetAlbum($id: ID!, $imageSize: Int = 512, $first: Int = 50, $after: String) {
460 catalog {
461 album(id: $id) {
462 id
463 tracks(first: $first, after: $after) {
464 items {
465 id
466 title
467 cover(size: $imageSize)
468 isrc
469 duration
470 label
471 artist {
472 id
473 title
474 cover(size: $imageSize)
475 }
476 featuredArtists {
477 items {
478 id
479 title
480 cover(size: $imageSize)
481 }
482 }
483 share
484 genre
485 }
486 pageInfo {
487 hasNextPage
488 endCursor
489 }
490 }
491 }
492 }
493 }
494 """
495 tracks = []
496 variables = {
497 "id": prov_album_id,
498 "imageSize": IMAGE_SIZE,
499 }
500
501 i = 1
502 async for item in self.api.paginate_graphql(
503 query,
504 variables,
505 ["data", "catalog", "album", "tracks"],
506 ):
507 track = await parse_track(self.provider, item)
508 track.position = i
509 tracks.append(track)
510 i += 1
511
512 return tracks
513
514 async def get_playlist_tracks(
515 self,
516 prov_playlist_id: str,
517 page: int = 0,
518 ) -> list[Track]:
519 """Get all playlist tracks for given playlist id."""
520 query = """
521 query getPlaylist($id: ID!, $imageSize: Int = 512, $first: Int = 50, $after: String) {
522 playlists {
523 playlist(id: $id) {
524 id
525 tracks(first: $first, after: $after) {
526 items {
527 id
528 title
529 cover(size: $imageSize)
530 isrc
531 duration
532 label
533 artist {
534 id
535 title
536 cover(size: $imageSize)
537 }
538 featuredArtists {
539 items {
540 id
541 title
542 cover(size: $imageSize)
543 }
544 }
545 share
546 genre
547 }
548 pageInfo {
549 hasNextPage
550 endCursor
551 }
552 }
553 }
554 }
555 }
556 """
557 tracks: list[Track] = []
558
559 if page > 0:
560 # paging not supported, we always return the whole list at once
561 return []
562 # TODO: access the underlying paging on the yousee api (if possible))
563
564 variables = {
565 "id": prov_playlist_id,
566 "imageSize": IMAGE_SIZE,
567 }
568
569 i = 1
570 async for item in self.api.paginate_graphql(
571 query, variables, ["data", "playlists", "playlist", "tracks"]
572 ):
573 track = await parse_track(self.provider, item)
574 track.position = i
575 tracks.append(track)
576 i += 1
577
578 return tracks
579
580 async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
581 """Retrieve a dynamic list of similar tracks based on the provided track."""
582 query = """
583 query similarTracks($id: ID!, $first: Int = 25, $imageSize: Int = 512) {
584 catalog {
585 track(id: $id) {
586 id
587 similarTracks(first: $first) {
588 items {
589 id
590 title
591 cover(size: $imageSize)
592 isrc
593 duration
594 label
595 artist {
596 id
597 title
598 cover(size: $imageSize)
599 }
600 featuredArtists {
601 items {
602 id
603 title
604 cover(size: $imageSize)
605 }
606 }
607 share
608 genre
609 }
610 }
611 }
612 }
613 }
614 """
615
616 variables = {
617 "id": prov_track_id,
618 "first": limit,
619 "imageSize": IMAGE_SIZE,
620 }
621 result = await self.api.post_graphql(query, variables)
622 if not result or not result.get("data", {}).get("catalog", {}).get("track"):
623 raise MediaNotFoundError(f"Track {prov_track_id} not found")
624
625 return [
626 await parse_track(self.provider, item)
627 for item in result["data"]["catalog"]["track"]["similarTracks"]["items"]
628 ]
629