/
/
/
1"""YouSee Musik playlist manager."""
2
3from typing import TYPE_CHECKING
4
5from music_assistant_models.errors import MediaNotFoundError
6
7from music_assistant.providers.yousee.constants import IMAGE_SIZE
8from music_assistant.providers.yousee.parsers import parse_playlist
9
10if TYPE_CHECKING:
11 from music_assistant_models.media_items import Playlist
12
13 from music_assistant.providers.yousee.provider import YouSeeMusikProvider
14
15
16class YouSeePlaylistManager:
17 """Manages YouSee Musik playlist operations."""
18
19 def __init__(self, provider: YouSeeMusikProvider):
20 """Initialize playlist manager."""
21 self.provider = provider
22 self.api = provider.api
23 self.auth = provider.auth
24 self.logger = provider.logger
25
26 async def create(self, name: str) -> Playlist:
27 """Create a new playlist on provider with given name."""
28 query = """
29 mutation createPlaylist($title: String!, $imageSize: Int = 512) {
30 playlists {
31 create(playlist: {title: $title}) {
32 playlist {
33 id
34 title
35 description
36 tracksCount
37 createdAt
38 isOwned
39 share
40 cover(size: $imageSize)
41 }
42 }
43 }
44 }
45 """
46 variables = {"title": name, "imageSize": IMAGE_SIZE}
47 result = await self.api.post_graphql(query, variables)
48 if not result or not result.get("data", {}).get("playlists", {}).get("create", {}).get(
49 "playlist"
50 ):
51 raise MediaNotFoundError(
52 f"Could not create playlist {name}",
53 translation_key="playlist_create_failed",
54 translation_owner=self.provider.translation_owner,
55 translation_args=[name],
56 )
57
58 return await parse_playlist(
59 self.provider, result["data"]["playlists"]["create"]["playlist"]
60 )
61
62 async def add_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
63 """Add track(s) to playlist."""
64 query = """
65 mutation addToLibrary( $id: ID!, $trackIds: [ID]!) {
66 playlists {
67 addTracks(id: $id, duplicatesHandling: SKIP_DUPLICATES, trackIds: $trackIds) {
68 ok
69 }
70 }
71 }
72 """
73 variables = {"id": prov_playlist_id, "trackIds": prov_track_ids}
74 result = await self.api.post_graphql(query, variables)
75
76 if not result or not result.get("data", {}).get("playlists", {}).get("addTracks", {}).get(
77 "ok"
78 ):
79 raise MediaNotFoundError(
80 f"Could not add tracks to playlist {prov_playlist_id}: {prov_track_ids}"
81 )
82
83 async def remove_tracks(
84 self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
85 ) -> None:
86 """Remove track(s) from playlist."""
87 query = """
88 mutation addToLibrary($id: ID!, $mods: [ModifyPlaylistTrackInput!]!) {
89 playlists {
90 modifyTracks(id: $id, modifications: $mods) {
91 ok
92 }
93 }
94 }
95
96 """
97
98 mods = [
99 {"positionFrom": pos - 1, "type": "REMOVE"}
100 for pos in sorted(positions_to_remove, reverse=True)
101 ]
102
103 variables = {"id": prov_playlist_id, "mods": mods}
104
105 result = await self.api.post_graphql(query, variables)
106
107 if not result or not result.get("data", {}).get("playlists", {}).get(
108 "modifyTracks", {}
109 ).get("ok"):
110 raise MediaNotFoundError(
111 f"Could not remove tracks from playlist {prov_playlist_id}: {positions_to_remove}"
112 )
113