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