/
/
/
1"""A minimal client for the unofficial gw-API, which deezer is using on their website and app.
2
3Credits go out to RemixDev (https://gitlab.com/RemixDev) for figuring out, how to get the arl
4cookie based on the api_token.
5"""
6
7import datetime
8from collections.abc import Mapping
9from http.cookies import BaseCookie, Morsel
10from typing import Any, cast
11
12from aiohttp import ClientSession, ClientTimeout
13from music_assistant_models.streamdetails import StreamDetails
14from yarl import URL
15
16from music_assistant.helpers.datetime import utc_timestamp
17
18USER_AGENT_HEADER = (
19 "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
20 "Chrome/79.0.3945.130 Safari/537.36"
21)
22
23GW_LIGHT_URL = "https://www.deezer.com/ajax/gw-light.php"
24
25
26class DeezerGWError(BaseException):
27 """Exception type for GWClient related exceptions."""
28
29
30class GWClient:
31 """The GWClient class can be used to perform actions not being of the official API."""
32
33 _arl_token: str
34 _api_token: str
35 _gw_csrf_token: str | None
36 _license: str | None
37 _license_expiration_timestamp: int
38 session: ClientSession
39 formats: list[dict[str, str]] = [
40 {"cipher": "BF_CBC_STRIPE", "format": "MP3_128"},
41 ]
42 user_country: str
43
44 def __init__(self, session: ClientSession, api_token: str, arl_token: str) -> None:
45 """Provide an aiohttp ClientSession and the deezer api_token."""
46 self._api_token = api_token
47 self._arl_token = arl_token
48 self.session = session
49
50 async def _set_cookie(self) -> None:
51 cookie: Morsel[str] = Morsel()
52
53 cookie.set("arl", self._arl_token, self._arl_token)
54 cookie.update({"domain": ".deezer.com", "path": "/", "httponly": "True"})
55
56 self.session.cookie_jar.update_cookies(BaseCookie({"arl": cookie}), URL(GW_LIGHT_URL))
57
58 async def _update_user_data(self) -> None:
59 user_data = await self._gw_api_call("deezer.getUserData", False)
60 if not user_data["results"]["USER"]["USER_ID"]:
61 await self._set_cookie()
62 user_data = await self._gw_api_call("deezer.getUserData", False)
63
64 if not user_data["results"]["OFFER_ID"]:
65 msg = "Free subscriptions cannot be used in MA. Make sure you set a valid ARL."
66 raise DeezerGWError(msg)
67
68 self._gw_csrf_token = user_data["results"]["checkForm"]
69 self._license = user_data["results"]["USER"]["OPTIONS"]["license_token"]
70 self._license_expiration_timestamp = user_data["results"]["USER"]["OPTIONS"][
71 "expiration_timestamp"
72 ]
73 web_qualities = user_data["results"]["USER"]["OPTIONS"]["web_sound_quality"]
74 mobile_qualities = user_data["results"]["USER"]["OPTIONS"]["mobile_sound_quality"]
75 if web_qualities["high"] or mobile_qualities["high"]:
76 self.formats.insert(0, {"cipher": "BF_CBC_STRIPE", "format": "MP3_320"})
77 if web_qualities["lossless"] or mobile_qualities["lossless"]:
78 self.formats.insert(0, {"cipher": "BF_CBC_STRIPE", "format": "FLAC"})
79
80 self.user_country = user_data["results"]["COUNTRY"]
81
82 async def setup(self) -> None:
83 """Call this to let the client get its cookies, license and tokens."""
84 await self._set_cookie()
85 await self._update_user_data()
86
87 async def _get_license(self) -> str | None:
88 if (
89 self._license_expiration_timestamp
90 < (datetime.datetime.now() + datetime.timedelta(days=1)).timestamp()
91 ):
92 await self._update_user_data()
93 return self._license
94
95 async def _gw_api_call(
96 self,
97 method: str,
98 use_csrf_token: bool = True,
99 args: dict[str, Any] | None = None,
100 params: dict[str, Any] | None = None,
101 http_method: str = "POST",
102 retry: bool = True,
103 ) -> dict[str, Any]:
104 csrf_token = self._gw_csrf_token if use_csrf_token else "null"
105 if params is None:
106 params = {}
107 parameters = {"api_version": "1.0", "api_token": csrf_token, "input": "3", "method": method}
108 parameters |= params
109 result = await self.session.request(
110 http_method,
111 GW_LIGHT_URL,
112 params=cast("Mapping[str, str]", parameters),
113 timeout=ClientTimeout(total=30),
114 json=args,
115 headers={"User-Agent": USER_AGENT_HEADER},
116 )
117 result_json = await result.json()
118
119 if result_json["error"]:
120 if retry:
121 await self._update_user_data()
122 return await self._gw_api_call(
123 method, use_csrf_token, args, params, http_method, False
124 )
125 msg = "Failed to call GW-API"
126 raise DeezerGWError(msg, result_json["error"])
127 return cast("dict[str, Any]", result_json)
128
129 async def get_song_data(self, track_id: str) -> dict[str, Any]:
130 """Get data such as the track token for a given track."""
131 return await self._gw_api_call("song.getData", args={"SNG_ID": track_id})
132
133 async def get_deezer_track_urls(self, track_id: str) -> tuple[dict[str, Any], dict[str, Any]]:
134 """Get the URL for a given track id."""
135 dz_license = await self._get_license()
136
137 song_results = await self.get_song_data(track_id)
138
139 song_data = song_results["results"]
140 # If the song has been replaced by a newer version, the old track will
141 # not play anymore. The data for the newer song is contained in a
142 # "FALLBACK" entry in the song data. So if that is available, use that
143 # instead so we get the right track token.
144 if "FALLBACK" in song_data:
145 song_data = song_data["FALLBACK"]
146
147 track_token = song_data["TRACK_TOKEN"]
148 url_data = {
149 "license_token": dz_license,
150 "media": [
151 {
152 "type": "FULL",
153 "formats": self.formats,
154 }
155 ],
156 "track_tokens": [track_token],
157 }
158 url_response = await self.session.post(
159 "https://media.deezer.com/v1/get_url",
160 json=url_data,
161 headers={"User-Agent": USER_AGENT_HEADER},
162 )
163 result_json = await url_response.json()
164
165 if error := result_json["data"][0].get("errors"):
166 msg = "Received an error from API"
167 raise DeezerGWError(msg, error)
168
169 return result_json["data"][0]["media"][0], song_data
170
171 async def log_listen(
172 self, next_track: str | None = None, last_track: StreamDetails | None = None
173 ) -> None:
174 """Log the next and/or previous track of the current playback queue."""
175 if not (next_track or last_track):
176 msg = "last or current track information must be provided."
177 raise DeezerGWError(msg)
178
179 payload: dict[str, Any] = {}
180
181 if next_track:
182 payload["next_media"] = {"media": {"id": next_track, "type": "song"}}
183
184 if last_track:
185 seconds_streamed = min(
186 utc_timestamp() - last_track.data["start_ts"],
187 last_track.seconds_streamed,
188 )
189
190 payload["params"] = {
191 "media": {
192 "id": last_track.item_id,
193 "type": "song",
194 "format": last_track.data["format"],
195 },
196 "type": 1,
197 "stat": {
198 "seek": 1 if seconds_streamed < last_track.duration else 0,
199 "pause": 0,
200 "sync": 0,
201 "next": bool(next_track),
202 },
203 "lt": int(seconds_streamed),
204 "ctxt": {"t": "search_page", "id": last_track.item_id},
205 "dev": {"v": "10020230525142740", "t": 0},
206 "ls": [],
207 "ts_listen": int(last_track.data["start_ts"]),
208 "is_shuffle": False,
209 "stream_id": str(last_track.data["stream_id"]),
210 }
211
212 await self._gw_api_call("log.listen", args=payload)
213