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