music-assistant-server

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