music-assistant-server

1"""Playlist management for Tidal."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6
7from aiohttp.client_exceptions import ClientError
8from music_assistant_models.errors import ResourceTemporarilyUnavailable
9
10from .constants import PLAYLISTS
11from .parsers import parse_playlist
12
13if TYPE_CHECKING:
14    from music_assistant_models.media_items import Playlist
15
16    from .provider import TidalProvider
17
18
19class TidalPlaylistManager:
20    """Manages Tidal playlist operations."""
21
22    def __init__(self, provider: TidalProvider):
23        """Initialize playlist manager."""
24        self.provider = provider
25        self.api = provider.api
26        self.auth = provider.auth
27        self.logger = provider.logger
28
29    async def create(self, name: str) -> Playlist:
30        """Create a new playlist."""
31        try:
32            data = {"title": name, "description": ""}
33            result = await self.api.post(
34                f"users/{self.auth.user_id}/playlists", data=data, as_form=True
35            )
36            return parse_playlist(self.provider, result)
37        except ClientError as err:
38            raise ResourceTemporarilyUnavailable("Failed to create playlist") from err
39
40    async def add_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
41        """Add tracks to playlist."""
42        try:
43            # Get ETag first
44            playlist_obj, etag = await self.api.get_with_etag(f"{PLAYLISTS}/{prov_playlist_id}")
45
46            data = {
47                "onArtifactNotFound": "SKIP",
48                "trackIds": ",".join(map(str, prov_track_ids)),
49                "toIndex": playlist_obj.get("numberOfTracks", 0),
50                "onDupes": "SKIP",
51            }
52            headers = {"If-None-Match": etag} if etag else {}
53
54            await self.api.post(
55                f"{PLAYLISTS}/{prov_playlist_id}/items", data=data, as_form=True, headers=headers
56            )
57        except ClientError as err:
58            raise ResourceTemporarilyUnavailable("Failed to add tracks") from err
59
60    async def remove_tracks(self, prov_playlist_id: str, positions: tuple[int, ...]) -> None:
61        """Remove tracks from playlist."""
62        try:
63            # Get ETag first
64            _, etag = await self.api.get_with_etag(f"{PLAYLISTS}/{prov_playlist_id}")
65
66            # Tidal uses 0-based indices in URL path
67            indices = ",".join(str(pos - 1) for pos in positions)
68            headers = {"If-None-Match": etag} if etag else {}
69
70            await self.api.delete(
71                f"{PLAYLISTS}/{prov_playlist_id}/items/{indices}", headers=headers
72            )
73        except ClientError as err:
74            raise ResourceTemporarilyUnavailable("Failed to remove tracks") from err
75