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