/
/
/
1"""Library management for Tidal."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any
6
7from aiohttp.client_exceptions import ClientError
8from music_assistant_models.enums import MediaType
9from music_assistant_models.errors import MediaNotFoundError, ResourceTemporarilyUnavailable
10
11from .parsers import parse_album, parse_artist, parse_playlist, parse_track
12
13if TYPE_CHECKING:
14 from collections.abc import AsyncGenerator
15
16 from music_assistant_models.media_items import Album, Artist, MediaItemType, Playlist, Track
17
18 from .provider import TidalProvider
19
20
21class TidalLibraryManager:
22 """Manages Tidal library operations."""
23
24 def __init__(self, provider: TidalProvider):
25 """Initialize library manager."""
26 self.provider = provider
27 self.api = provider.api
28 self.auth = provider.auth
29 self.logger = provider.logger
30
31 async def get_artists(self) -> AsyncGenerator[Artist, None]:
32 """Retrieve library artists."""
33 path = f"users/{self.auth.user_id}/favorites/artists"
34 async for item in self.api.paginate(path):
35 if item and item.get("item") and item["item"].get("id"):
36 yield parse_artist(self.provider, item)
37
38 async def get_albums(self) -> AsyncGenerator[Album, None]:
39 """Retrieve library albums."""
40 path = f"users/{self.auth.user_id}/favorites/albums"
41 async for item in self.api.paginate(path):
42 if item and item.get("item") and item["item"].get("id"):
43 yield parse_album(self.provider, item)
44
45 async def get_tracks(self) -> AsyncGenerator[Track, None]:
46 """Retrieve library tracks."""
47 path = f"users/{self.auth.user_id}/favorites/tracks"
48 async for item in self.api.paginate(path):
49 if item and item.get("item") and item["item"].get("id"):
50 yield parse_track(self.provider, item)
51
52 async def get_playlists(self) -> AsyncGenerator[Playlist, None]:
53 """Retrieve library playlists."""
54 # 1. Get favorite mixes
55 async for item in self.api.paginate(
56 "favorites/mixes", item_key="items", base_url=self.api.BASE_URL_V2, cursor_based=True
57 ):
58 if item and item.get("id"):
59 yield parse_playlist(self.provider, item, is_mix=True)
60
61 # 2. Get user playlists
62 path = f"users/{self.auth.user_id}/playlistsAndFavoritePlaylists"
63 async for item in self.api.paginate(path):
64 if item and item.get("playlist") and item["playlist"].get("uuid"):
65 yield parse_playlist(self.provider, item)
66
67 async def add_item(self, item: MediaItemType) -> bool:
68 """Add item to library."""
69 endpoint, data, is_mix = self._get_endpoint_data(item.item_id, item.media_type, "add")
70 if not endpoint:
71 return False
72
73 try:
74 if is_mix:
75 await self.api.put(endpoint, data=data, as_form=True)
76 else:
77 await self.api.post(
78 f"users/{self.auth.user_id}/{endpoint}", data=data, as_form=True
79 )
80 return True
81 except (ClientError, MediaNotFoundError, ResourceTemporarilyUnavailable):
82 return False
83
84 async def remove_item(self, prov_item_id: str, media_type: MediaType) -> bool:
85 """Remove item from library."""
86 endpoint, data, is_mix = self._get_endpoint_data(prov_item_id, media_type, "remove")
87 if not endpoint:
88 return False
89
90 try:
91 if is_mix:
92 await self.api.put(endpoint, data=data, as_form=True)
93 else:
94 await self.api.delete(f"users/{self.auth.user_id}/{endpoint}")
95 return True
96 except (ClientError, MediaNotFoundError, ResourceTemporarilyUnavailable):
97 return False
98
99 def _get_endpoint_data(
100 self, item_id: str, media_type: MediaType, operation: str
101 ) -> tuple[str | None, dict[str, Any], bool]:
102 """Get endpoint and data for library operations."""
103 if media_type == MediaType.PLAYLIST and item_id.startswith("mix_"):
104 mix_id = item_id[4:]
105 if operation == "add":
106 return (
107 "favorites/mixes/add",
108 {
109 "mixIds": mix_id,
110 "onArtifactNotFound": "FAIL",
111 "deviceType": "BROWSER",
112 },
113 True,
114 )
115 return (
116 "favorites/mixes/remove",
117 {"mixIds": mix_id, "deviceType": "BROWSER"},
118 True,
119 )
120
121 if media_type == MediaType.ARTIST:
122 return (
123 ("favorites/artists", {"artistId": item_id}, False)
124 if operation == "add"
125 else (f"favorites/artists/{item_id}", {}, False)
126 )
127 if media_type == MediaType.ALBUM:
128 return (
129 ("favorites/albums", {"albumId": item_id}, False)
130 if operation == "add"
131 else (f"favorites/albums/{item_id}", {}, False)
132 )
133 if media_type == MediaType.TRACK:
134 return (
135 ("favorites/tracks", {"trackId": item_id}, False)
136 if operation == "add"
137 else (f"favorites/tracks/{item_id}", {}, False)
138 )
139 if media_type == MediaType.PLAYLIST:
140 return (
141 ("favorites/playlists", {"uuids": item_id}, False)
142 if operation == "add"
143 else (f"favorites/playlists/{item_id}", {}, False)
144 )
145
146 return None, {}, False
147