music-assistant-server

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