from __future__ import annotations from typing import Iterable, Mapping, Sequence import requests from src import config from src.data_info import PlaylistInfo, TokenInfo, UserInfo from ..errors import BaseApiError, DeezerApiError from .base import BaseDSPApiClient __all__ = ["DeezerApiClient"] class DeezerApiClient(BaseDSPApiClient): default_error_cls = DeezerApiError base_url = "https://api.deezer.com" base_auth_url = "https://connect.deezer.com/oauth" batch_size: int = 50 def __init__(self, *, app_id: str = config.DEEZER_APP_ID, secret: str = config.DEEZER_SECRET, **kwargs): super().__init__(**kwargs) self._app_id = app_id self._secret = secret @staticmethod def _is_auth_error(error: BaseApiError) -> bool: # deezer API token with "offline_access" permission will newer expire # but just in case :) return error.response and error.response.json()["error"]["code"] == 300 def _check_response(self, response: requests.Response) -> None: super()._check_response(response) error = None try: data = response.json() if isinstance(data, dict): error = data.get("error") except requests.JSONDecodeError: error = response.text if error: raise DeezerApiError(error, response=response) def _get_token_from_code(self, code: str, redirect_uri: str, **extra) -> TokenInfo: request = requests.Request( method="get", url=f"{self.base_auth_url}/access_token.php", params={"app_id": self._app_id, "secret": self._secret, "code": code, "output": "json"}, ) def _check_response(response: requests.Response): if response.status_code != 200 or "wrong code" in response.text: raise DeezerApiError(response.text) result = self.send_request_with_retry(request, authorize=False, check_response=_check_response).json() return TokenInfo( access_token=result["access_token"], expires_in=result["expires"], ) def _refresh_access_token(self) -> tuple[str, int]: # deezer API token with "offline_access" permission will newer expire if self._access_token is None: raise DeezerApiError("access token is not defined") return self._access_token, 0 def _is_token_expired(self) -> bool: # deezer API token with "offline_access" permission will newer expire return False def _authorize_request(self, request: requests.Request): request.params["access_token"] = self.get_access_token() def get_track_id_by_isrc(self, isrc: str) -> str: self._logger.debug(f"Getting info for track with ISRC '{isrc}'") request = requests.Request(method="get", url=f"{self.base_url}/track/isrc:{isrc}") result = self.send_request_with_retry(request, authorize=False).json() return str(result["id"]) def get_playlist_track_ids(self, playlist_id: str) -> Iterable[str]: self._logger.debug(f"Getting tracks for playlist with id '{playlist_id}'") page = 0 while True: request = requests.Request( method="get", url=f"{self.base_url}/playlist/{playlist_id}/tracks", params={ "index": page * self.batch_size, # offset "limit": self.batch_size, }, ) result = self.send_request_with_retry(request).json() for track in result["data"]: yield str(track["id"]) if "next" not in result: break page += 1 def insert_tracks(self, playlist_id: str, track_ids: Iterable[str]): self._logger.debug(f"Inserting tracks to playlist with id '{playlist_id}'") request = requests.Request( method="post", url=f"{self.base_url}/playlist/{playlist_id}/tracks", params={"songs": ",".join(track_ids)}, ) self.send_request_with_retry(request) self._logger.debug(f"Inserted tracks to playlist with id '{playlist_id}'") def delete_tracks(self, playlist_id: str, track_ids: Iterable[str]): self._logger.debug(f"Deleting tracks from playlist with id '{playlist_id}'") request = requests.Request( method="delete", url=f"{self.base_url}/playlist/{playlist_id}/tracks", params={"songs": ",".join(track_ids)}, ) self.send_request_with_retry(request) self._logger.debug(f"Deleted tracks from playlist with id '{playlist_id}'") def sort_tracks(self, playlist_id: str, track_ids: Sequence[str]): self._logger.debug(f"Sorting tracks in playlist with id '{playlist_id}'") request = requests.Request( method="post", url=f"{self.base_url}/playlist/{playlist_id}/tracks", params={"order": ",".join(track_ids)}, ) self.send_request_with_retry(request) self._logger.debug(f"Sorted tracks in playlist with id '{playlist_id}'") def get_playlist_info(self, playlist_id: str) -> PlaylistInfo: self._logger.debug(f"Getting info about playlist with id '{playlist_id}'") request = requests.Request(method="get", url=f"{self.base_url}/playlist/{playlist_id}") result = self.send_request_with_retry(request).json() self._logger.debug(f"Got info about playlist with id '{playlist_id}'") return PlaylistInfo( title=result["title"], description=result["description"], image_url=result["picture_medium"], total_tracks=result["nb_tracks"], user_id=result["creator"]["id"], user_name=result["creator"]["name"], ) def update_playlist_info(self, playlist_id: str, info: Mapping[str, str | None]): self._logger.debug(f"Updating info about playlist with id '{playlist_id}'") request = requests.Request(method="post", url=f"{self.base_url}/playlist/{playlist_id}", params=info) self.send_request_with_retry(request) self._logger.debug(f"Updated info about playlist with id '{playlist_id}'") def get_user_info(self, **extra) -> UserInfo: self._logger.debug("Getting user info") request = requests.Request(method="get", url=f"{self.base_url}/user/me") result = self.send_request_with_retry(request).json() self._logger.debug("Got user info") return UserInfo(user_identifier=result["id"], display_name=result["name"], image_url=result["picture"])