/
/
/
1"""Library management for YouSee Musik."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator
6from typing import TYPE_CHECKING
7
8from music_assistant_models.enums import MediaType
9from music_assistant_models.errors import InvalidDataError
10
11from music_assistant.constants import VERBOSE_LOG_LEVEL
12from music_assistant.providers.yousee.constants import IMAGE_SIZE
13from music_assistant.providers.yousee.parsers import (
14 parse_album,
15 parse_artist,
16 parse_playlist,
17 parse_track,
18)
19
20if TYPE_CHECKING:
21 from music_assistant_models.media_items import Album, Artist, MediaItemType, Playlist, Track
22
23 from music_assistant.providers.yousee.provider import YouSeeMusikProvider
24
25
26class YouSeeLibraryManager:
27 """Manages YouSee Musik library operations."""
28
29 def __init__(self, provider: YouSeeMusikProvider):
30 """Initialize library manager."""
31 self.provider = provider
32 self.api = provider.api
33 self.auth = provider.auth
34 self.logger = provider.logger
35
36 async def get_artists(self) -> AsyncGenerator[Artist, None]:
37 """Retrieve library artists from the provider."""
38 query = """
39 query favoriteArtists($first: Int!, $after: String, $imageSize: Int = 512) {
40 me {
41 favorites {
42 artists(first: $first, after: $after) {
43 totalCount,
44 pageInfo {
45 endCursor
46 hasNextPage
47 }
48 items {
49 id
50 title
51 cover(size: $imageSize)
52 share
53 }
54 }
55 }
56 }
57 }
58 """
59 variables = {"imageSize": IMAGE_SIZE}
60
61 async for item in self.api.paginate_graphql(
62 query, variables, ["data", "me", "favorites", "artists"]
63 ):
64 self.logger.log(VERBOSE_LOG_LEVEL, "Parsing artist item: %s", item)
65 yield parse_artist(self.provider, item)
66
67 async def get_albums(self) -> AsyncGenerator[Album, None]:
68 """Retrieve library albums from the provider."""
69 query = """
70 query favoriteAlbums($first: Int!, $after: String, $imageSize: Int = 512) {
71 me {
72 favorites {
73 albums(first: $first, after: $after) {
74 totalCount,
75 pageInfo {
76 endCursor
77 hasNextPage
78 }
79 items {
80 id
81 title
82 cover(size: $imageSize)
83 artist {
84 id
85 title
86 cover(size: $imageSize)
87 }
88 }
89 }
90 }
91 }
92 }
93 """
94 variables = {"imageSize": IMAGE_SIZE}
95
96 async for item in self.api.paginate_graphql(
97 query, variables, ["data", "me", "favorites", "albums"]
98 ):
99 self.logger.log(VERBOSE_LOG_LEVEL, "Parsing album item: %s", item)
100 yield await parse_album(self.provider, item)
101
102 async def get_tracks(self) -> AsyncGenerator[Track, None]:
103 """Retrieve library tracks from the provider."""
104 query = """
105 query favoriteTracks($first: Int!, $after: String, $imageSize: Int = 512) {
106 me {
107 favorites {
108 tracks(first: $first, after: $after) {
109 totalCount
110 pageInfo {
111 endCursor
112 hasNextPage
113 }
114 items {
115 id
116 title
117 availableToStream
118 album {
119 id
120 title
121 }
122 artist {
123 id
124 title
125 cover(size: $imageSize)
126 }
127 cover(size: $imageSize)
128 duration
129 share
130 genre
131 isrc
132 featuredArtists {
133 items {
134 id
135 title
136 cover(size: $imageSize)
137 }
138 }
139 }
140 }
141 }
142 }
143 }
144 """
145 variables = {"imageSize": IMAGE_SIZE}
146
147 async for item in self.api.paginate_graphql(
148 query, variables, ["data", "me", "favorites", "tracks"]
149 ):
150 self.logger.log(VERBOSE_LOG_LEVEL, "Parsing track item: %s", item)
151 yield await parse_track(self.provider, item)
152
153 async def get_playlists(self) -> AsyncGenerator[Playlist, None]:
154 """Retrieve library/subscribed playlists from the provider."""
155 query = """
156 query favoritePlaylists($first: Int!, $after: String, $imageSize: Int = 512) {
157 me {
158 playlists {
159 combinedPlaylists(first: $first, after: $after, orderBy: MODIFIED_DATE) {
160 totalCount
161 pageInfo {
162 hasNextPage
163 endCursor
164 }
165 items {
166 id
167 title
168 isOwned
169 share
170 cover(size: $imageSize)
171 description
172 }
173 }
174 }
175 }
176 }
177 """
178 variables = {"imageSize": IMAGE_SIZE}
179 async for item in self.api.paginate_graphql(
180 query, variables, ["data", "me", "playlists", "combinedPlaylists"]
181 ):
182 self.logger.log(VERBOSE_LOG_LEVEL, "Parsing playlist item: %s", item)
183 yield await parse_playlist(self.provider, item)
184
185 async def add_item(self, item: MediaItemType) -> bool:
186 """Add item to provider's library. Return true on success."""
187 if item.media_type not in (
188 MediaType.ARTIST,
189 MediaType.ALBUM,
190 MediaType.TRACK,
191 MediaType.PLAYLIST,
192 ):
193 raise InvalidDataError(
194 f"Cannot add media type {item.media_type} to library for provider "
195 f"{self.provider.name}"
196 )
197
198 media_type_str = item.media_type.capitalize()
199
200 query = f"""
201 mutation addToLibrary($id: ID!) {{
202 favorites {{
203 add{media_type_str} (id: $id) {{
204 ok
205 }}
206 }}
207 }}
208 """
209 variables = {"id": item.item_id}
210
211 result = await self.api.post_graphql(query, variables)
212
213 return bool(
214 result.get("data", {})
215 .get("favorites", {})
216 .get(f"add{media_type_str}", {})
217 .get("ok", False)
218 )
219
220 async def remove_item(self, prov_item_id: str, media_type: MediaType) -> bool:
221 """Remove item from provider's library. Return true on success."""
222 if media_type not in (
223 MediaType.ARTIST,
224 MediaType.ALBUM,
225 MediaType.TRACK,
226 MediaType.PLAYLIST,
227 ):
228 raise InvalidDataError(
229 f"Cannot remove media type {media_type} from library for provider "
230 f"{self.provider.name}"
231 )
232
233 media_type_str = media_type.capitalize()
234
235 query = f"""
236 mutation removeFromLibrary($id: ID!) {{
237 favorites {{
238 remove{media_type_str} (id: $id) {{
239 ok
240 }}
241 }}
242 }}
243 """
244 variables = {"id": prov_item_id}
245
246 result = await self.api.post_graphql(query, variables)
247
248 return bool(
249 result.get("data", {})
250 .get("favorites", {})
251 .get(f"remove{media_type_str}", {})
252 .get("ok", False)
253 )
254