/
/
/
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(f"Could not create playlist {name}")
52
53 return await parse_playlist(
54 self.provider, result["data"]["playlists"]["create"]["playlist"]
55 )
56
57 async def add_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
58 """Add track(s) to playlist."""
59 query = """
60 mutation addToLibrary( $id: ID!, $trackIds: [ID]!) {
61 playlists {
62 addTracks(id: $id, duplicatesHandling: SKIP_DUPLICATES, trackIds: $trackIds) {
63 ok
64 }
65 }
66 }
67 """
68 variables = {"id": prov_playlist_id, "trackIds": prov_track_ids}
69 result = await self.api.post_graphql(query, variables)
70
71 if not result or not result.get("data", {}).get("playlists", {}).get("addTracks", {}).get(
72 "ok"
73 ):
74 raise MediaNotFoundError(
75 f"Could not add tracks to playlist {prov_playlist_id}: {prov_track_ids}"
76 )
77
78 async def remove_tracks(
79 self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
80 ) -> None:
81 """Remove track(s) from playlist."""
82 query = """
83 mutation addToLibrary($id: ID!, $mods: [ModifyPlaylistTrackInput!]!) {
84 playlists {
85 modifyTracks(id: $id, modifications: $mods) {
86 ok
87 }
88 }
89 }
90
91 """
92
93 mods = [
94 {"positionFrom": pos - 1, "type": "REMOVE"}
95 for pos in sorted(positions_to_remove, reverse=True)
96 ]
97
98 variables = {"id": prov_playlist_id, "mods": mods}
99
100 result = await self.api.post_graphql(query, variables)
101
102 if not result or not result.get("data", {}).get("playlists", {}).get(
103 "modifyTracks", {}
104 ).get("ok"):
105 raise MediaNotFoundError(
106 f"Could not remove tracks from playlist {prov_playlist_id}: {positions_to_remove}"
107 )
108