/
/
/
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