/
/
/
1"""
2A minimal client for the unofficial gw-API, which deezer is using on their website and app.
3
4Credits go out to RemixDev (https://gitlab.com/RemixDev) for figuring out, how to get the arl
5cookie based on the api_token.
6"""
7
8from __future__ import annotations
9
10from collections.abc import Mapping
11from http.cookies import BaseCookie, Morsel
12from typing import TYPE_CHECKING, Any, ClassVar, cast
13
14from aiohttp import ClientSession, ClientTimeout
15from music_assistant_models.errors import MediaNotFoundError
16from yarl import URL
17
18from music_assistant.helpers.datetime import future_timestamp, utc_timestamp
19
20if TYPE_CHECKING:
21 from music_assistant_models.streamdetails import StreamDetails
22
23USER_AGENT_HEADER = (
24 "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
25 "Chrome/79.0.3945.130 Safari/537.36"
26)
27
28GW_LIGHT_URL = "https://www.deezer.com/ajax/gw-light.php"
29
30
31class DeezerGWError(Exception):
32 """Exception type for GWClient related exceptions."""
33
34
35class GWClient:
36 """The GWClient class can be used to perform actions not being of the official API."""
37
38 _arl_token: str
39 _gw_csrf_token: str | None
40 _license: str | None
41 _license_expiration_timestamp: int
42 _user_id: int
43 session: ClientSession
44 formats: ClassVar[list[dict[str, str]]] = [
45 {"cipher": "BF_CBC_STRIPE", "format": "MP3_128"},
46 ]
47 user_country: str
48
49 def __init__(self, session: ClientSession, arl_token: str) -> None:
50 """Provide an aiohttp ClientSession and the deezer ARL token."""
51 self._arl_token = arl_token
52 self.session = session
53
54 async def _set_cookie(self) -> None:
55 cookie: Morsel[str] = Morsel()
56
57 cookie.set("arl", self._arl_token, self._arl_token)
58 cookie.update({"domain": ".deezer.com", "path": "/", "httponly": "True"})
59
60 self.session.cookie_jar.update_cookies(BaseCookie({"arl": cookie}), URL(GW_LIGHT_URL))
61
62 async def _update_user_data(self) -> None:
63 user_data = await self._gw_api_call("deezer.getUserData", False)
64 if not user_data["results"]["USER"]["USER_ID"]:
65 await self._set_cookie()
66 user_data = await self._gw_api_call("deezer.getUserData", False)
67
68 if not user_data["results"]["OFFER_ID"]:
69 msg = "Free subscriptions cannot be used in MA. Make sure you set a valid ARL."
70 raise DeezerGWError(msg)
71
72 self._gw_csrf_token = user_data["results"]["checkForm"]
73 self._user_id = int(user_data["results"]["USER"]["USER_ID"])
74 self._license = user_data["results"]["USER"]["OPTIONS"]["license_token"]
75 self._license_expiration_timestamp = user_data["results"]["USER"]["OPTIONS"][
76 "expiration_timestamp"
77 ]
78 web_qualities = user_data["results"]["USER"]["OPTIONS"]["web_sound_quality"]
79 mobile_qualities = user_data["results"]["USER"]["OPTIONS"]["mobile_sound_quality"]
80 if web_qualities["high"] or mobile_qualities["high"]:
81 self.formats.insert(0, {"cipher": "BF_CBC_STRIPE", "format": "MP3_320"})
82 if web_qualities["lossless"] or mobile_qualities["lossless"]:
83 self.formats.insert(0, {"cipher": "BF_CBC_STRIPE", "format": "FLAC"})
84
85 self.user_country = user_data["results"]["COUNTRY"]
86
87 async def setup(self) -> None:
88 """Call this to let the client get its cookies, license and tokens."""
89 await self._set_cookie()
90 await self._update_user_data()
91
92 async def _get_license(self) -> str | None:
93 if self._license_expiration_timestamp < future_timestamp(days=1):
94 await self._update_user_data()
95 return self._license
96
97 async def _gw_api_call(
98 self,
99 method: str,
100 use_csrf_token: bool = True,
101 args: dict[str, Any] | None = None,
102 params: dict[str, Any] | None = None,
103 http_method: str = "POST",
104 retry: bool = True,
105 ) -> dict[str, Any]:
106 csrf_token = self._gw_csrf_token if use_csrf_token else "null"
107 if params is None:
108 params = {}
109 parameters = {"api_version": "1.0", "api_token": csrf_token, "input": "3", "method": method}
110 parameters |= params
111 result = await self.session.request(
112 http_method,
113 GW_LIGHT_URL,
114 params=cast("Mapping[str, str]", parameters),
115 timeout=ClientTimeout(total=30),
116 json=args,
117 headers={"User-Agent": USER_AGENT_HEADER},
118 )
119 result_json = await result.json()
120
121 if result_json["error"]:
122 if retry:
123 await self._update_user_data()
124 return await self._gw_api_call(
125 method, use_csrf_token, args, params, http_method, False
126 )
127 msg = "Failed to call GW-API"
128 raise DeezerGWError(msg, result_json["error"])
129 return cast("dict[str, Any]", result_json)
130
131 # Content support descriptor for page.get â tells the API which module types to return
132 _PAGE_SUPPORT: ClassVar[dict[str, Any]] = {
133 "grid": ["channel", "album", "playlist", "artist"],
134 "horizontal-grid": ["channel", "album", "playlist", "artist"],
135 "slideshow": ["album", "playlist"],
136 "grid-preview-one": ["album", "playlist"],
137 "grid-preview-two": ["album", "playlist"],
138 "filterable-grid": ["album", "playlist"],
139 "large-card": ["album", "playlist"],
140 }
141
142 async def get_page(self, page: str, language: str = "en") -> dict[str, Any]:
143 """
144 Fetch a content page from the Deezer page.get GW API.
145
146 :param page: The page path (e.g., 'channels/audiobooks').
147 :param language: Language code for localized content.
148 """
149 result = await self._gw_api_call(
150 "page.get",
151 args={
152 "PAGE": page,
153 "VERSION": "2.5",
154 "SUPPORT": self._PAGE_SUPPORT,
155 "LANG": language,
156 "OPTIONS": [],
157 },
158 )
159 return cast("dict[str, Any]", result["results"])
160
161 async def get_deezer_track_urls(self, track_id: str) -> tuple[dict[str, Any], dict[str, Any]]:
162 """Get the URL for a given track id."""
163 dz_license = await self._get_license()
164
165 song_results = await self._gw_api_call("song.getData", args={"SNG_ID": track_id})
166
167 song_data = song_results["results"]
168 # If the song has been replaced by a newer version, the old track will
169 # not play anymore. The data for the newer song is contained in a
170 # "FALLBACK" entry in the song data. So if that is available, use that
171 # instead so we get the right track token.
172 if "FALLBACK" in song_data:
173 song_data = song_data["FALLBACK"]
174
175 track_token = song_data["TRACK_TOKEN"]
176 # Personal songs (user uploads) only support MP3_MISC format
177 is_personal = int(track_id) < 0
178 formats = (
179 [{"cipher": "BF_CBC_STRIPE", "format": "MP3_MISC"}] if is_personal else self.formats
180 )
181 url_data = {
182 "license_token": dz_license,
183 "media": [
184 {
185 "type": "FULL",
186 "formats": formats,
187 }
188 ],
189 "track_tokens": [track_token],
190 }
191 url_response = await self.session.post(
192 "https://media.deezer.com/v1/get_url",
193 json=url_data,
194 headers={"User-Agent": USER_AGENT_HEADER},
195 )
196 result_json = await url_response.json()
197
198 if error := result_json["data"][0].get("errors"):
199 error_code = error[0].get("code") if isinstance(error, list) and error else None
200 if error_code == 2002:
201 msg = f"Track {track_id} not available: insufficient streaming rights"
202 else:
203 msg = "Received an error from API"
204 raise DeezerGWError(msg, error)
205
206 media_list = result_json["data"][0].get("media", [])
207 if not media_list:
208 raise MediaNotFoundError(f"No media available for track {track_id}")
209
210 return media_list[0], song_data
211
212 async def log_listen(
213 self, next_track: str | None = None, last_track: StreamDetails | None = None
214 ) -> None:
215 """Log the next and/or previous track of the current playback queue."""
216 if not (next_track or last_track):
217 msg = "last or current track information must be provided."
218 raise DeezerGWError(msg)
219
220 payload: dict[str, Any] = {}
221
222 if next_track:
223 payload["next_media"] = {"media": {"id": next_track, "type": "song"}}
224
225 if last_track:
226 elapsed = utc_timestamp() - last_track.data["start_ts"]
227 seconds_streamed = (
228 min(elapsed, last_track.seconds_streamed)
229 if last_track.seconds_streamed is not None
230 else elapsed
231 )
232
233 payload["params"] = {
234 "media": {
235 "id": last_track.item_id,
236 "type": "song",
237 "format": last_track.data["format"],
238 },
239 "type": 1,
240 "stat": {
241 "seek": 1 if seconds_streamed < last_track.duration else 0,
242 "pause": 0,
243 "sync": 0,
244 "next": bool(next_track),
245 },
246 "lt": int(seconds_streamed),
247 "ctxt": {"t": "search_page", "id": last_track.item_id},
248 "dev": {"v": "10020230525142740", "t": 0},
249 "ls": [],
250 "ts_listen": int(last_track.data["start_ts"]),
251 "is_shuffle": False,
252 "stream_id": str(last_track.data["stream_id"]),
253 }
254
255 await self._gw_api_call("log.listen", args=payload)
256
257 async def get_personal_songs(self, start: int = 0, nb: int = 500) -> dict[str, Any]:
258 """
259 Get user-uploaded personal songs via the GW API.
260
261 :param start: Offset for pagination.
262 :param nb: Number of songs to fetch per page.
263 """
264 result = await self._gw_api_call(
265 "personal_song.getList",
266 args={"start": start, "nb": nb},
267 )
268 return cast("dict[str, Any]", result["results"])
269