/
/
/
1"""Playlist management for Tidal."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import TYPE_CHECKING
7
8from aiohttp.client_exceptions import ClientError
9from music_assistant_models.errors import ResourceTemporarilyUnavailable
10
11from .constants import PLAYLISTS
12from .parsers import parse_playlist
13
14if TYPE_CHECKING:
15 from music_assistant_models.media_items import Playlist
16
17 from .provider import TidalProvider
18
19
20class TidalPlaylistManager:
21 """Manages Tidal playlist operations."""
22
23 def __init__(self, provider: TidalProvider):
24 """Initialize playlist manager."""
25 self.provider = provider
26 self.api = provider.api
27 self.auth = provider.auth
28 self.logger = provider.logger
29
30 async def create(self, name: str) -> Playlist:
31 """Create a new playlist."""
32 try:
33 result = await self.api.post(
34 f"users/{self.auth.user_id}/playlists",
35 data={"title": name, "description": ""},
36 as_form=True,
37 )
38 playlist = parse_playlist(self.provider, result)
39 # A freshly created playlist is by definition owned and editable by the
40 # authenticated user, regardless of what the create response echoes back.
41 playlist.is_editable = True
42 playlist.owner = (
43 self.auth.user.profile_name or self.auth.user.user_name or str(self.auth.user_id)
44 )
45 for mapping in playlist.provider_mappings:
46 mapping.is_unique = True
47 return playlist
48 except ClientError as err:
49 raise ResourceTemporarilyUnavailable("Failed to create playlist") from err
50
51 async def add_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
52 """Add tracks to playlist."""
53 # v1 appends at the end and skips dead/duplicate ids server-side
54 # (onArtifactNotFound / onDupes = SKIP), so no reactive healing round-trip
55 # is needed; the cache-only redirect still maps ids the churn cache already
56 # knows are stale onto their live equivalents before they would be skipped.
57 # The playlist ETag guards the write against a concurrent modification.
58 try:
59 sent = await asyncio.gather(
60 *(self.provider.redirect_cached_id(str(tid)) for tid in prov_track_ids)
61 )
62 playlist_obj, etag = await self.api.get_with_etag(f"{PLAYLISTS}/{prov_playlist_id}")
63 data = {
64 "onArtifactNotFound": "SKIP",
65 "trackIds": ",".join(sent),
66 "toIndex": playlist_obj.get("numberOfTracks", 0),
67 "onDupes": "SKIP",
68 }
69 headers = {"If-None-Match": etag} if etag else {}
70 await self.api.post(
71 f"{PLAYLISTS}/{prov_playlist_id}/items", data=data, as_form=True, headers=headers
72 )
73 except ClientError as err:
74 raise ResourceTemporarilyUnavailable("Failed to add tracks") from err
75
76 async def remove_tracks(self, prov_playlist_id: str, positions: tuple[int, ...]) -> None:
77 """Remove tracks from playlist."""
78 # MA's 1-based positions come from the same v1 track listing get_playlist_tracks
79 # returns, and v1 deletes by 0-based index, so the positions map straight onto the
80 # delete with no cross-API resolution. The ETag guards against a lost update.
81 if not positions:
82 return
83 try:
84 _, etag = await self.api.get_with_etag(f"{PLAYLISTS}/{prov_playlist_id}")
85 indices = ",".join(str(pos - 1) for pos in positions)
86 headers = {"If-None-Match": etag} if etag else {}
87 await self.api.delete(
88 f"{PLAYLISTS}/{prov_playlist_id}/items/{indices}", headers=headers
89 )
90 except ClientError as err:
91 raise ResourceTemporarilyUnavailable("Failed to remove tracks") from err
92