feat(Playlist): Add more edit functions

Author: tabbeddev

src/deezergw/__init__.py

@@ -1,6 +1,6 @@
 import json as _json
 from os import PathLike
-from typing import Any, Generator, Iterable, List, Optional, Union
+from typing import Any, Generator, List, Optional, Union
 from deezergw.api import DeezerAPI
 from deezergw import decrypt_utils as _decrypt_utils
 from deezergw.resources.album import Album
@@ -12,7 +12,7 @@
 from deezergw.search_resources.playlist import SearchPlaylist
 from deezergw.search_resources.results import SearchResults
 from deezergw.search_resources.track import SearchTrack
-from deezergw.types import LoginDumpData
+from deezergw.types import LoginDumpData, ArrayLike
 
 
 class Client:
@@ -192,12 +192,12 @@ def search(self, query: str):
 
     # Batch functions
 
-    def get_tracks(self, ids: Iterable[str]):
+    def get_tracks(self, ids: ArrayLike[str]):
         """
         Get multiple Track objects by there ids.
 
         :param ids: The track ids in some form of a list
-        :type ids: Iterable[str]
+        :type ids: ArrayLike[str]
         :return: A list of the Tracks
         :rtype: List[Track]
         """

src/deezergw/api.py

@@ -1,9 +1,9 @@
 from datetime import datetime
 from requests import Session
-from typing import Any, Dict, Iterable, Optional, Tuple, Union
-from deezergw.exceptions import NoRightOnMedia, NotFoundException, UnauthorizedException
+from typing import Any, Dict, Optional, Tuple, Union
+from deezergw.exceptions import AlreadyExistsError, NotFoundException, RequestSpecificError, UnauthorizedException, UnknownException
 from deezergw.globals import Qualities, QualityType
-from deezergw.types import LoginDumpData, MediaData
+from deezergw.types import LoginDumpData, MediaData, PlaylistState, ArrayLike
 
 METHOD_GET_USER_DATA = "deezer.getUserData"
 METHOD_GET_USER_PROFILE = "deezer.pageProfile"
@@ -63,6 +63,7 @@
     "UpdatePlaylist",
     "mutation UpdatePlaylist($input: PlaylistUpdateMutationInput!) { updatePlaylist(input: $input) { playlist { id title description isPrivate isCollaborative picture { id } } } }",
 )
+GRAPHQL_GET_PLAYLIST_STATE = ("PlaylistState","query PlaylistState($playlistId: String!) { playlist(playlistId: $playlistId) { isPrivate isCollaborative } }")
 
 
 class DeezerAPI:
@@ -167,19 +168,14 @@ def _get_api(
                 retries -= 1
                 self._refresh_token()
                 return self._get_api(method, json_data, retries)
+            elif "ERROR_DATA_EXISTS" in error_data:
+                raise AlreadyExistsError(error_data["ERROR_DATA_EXISTS"])
             else:
-                # Catch any unknown error
-                print("[!] An unknown error was received by DeezerGW. Please report it")
-                print("[ ] JSON-Data:")
-                print(response.json())
+                # Any other error is up to the specific function to handle
 
-                if retries <= 0:
-                    raise Exception("Results are empty")
-
-                print(f"Retrying ({retries} tries left) ...")
-                retries -= 1
-                self._refresh_token()
-                return self._get_api(method, json_data, retries=retries)
+                error_type: str = tuple(error_data.keys())[0]
+                error_msg: str = error_data[error_type]
+                raise RequestSpecificError(error_type, error_msg)
 
         return results
 
@@ -206,7 +202,10 @@ def _request_graphql(
                 print("Refreshing JWT Token...")
                 self._refresh_jwt_token()
                 return self._request_graphql(query_pair, variables)
+            elif response_json["errors"][0]["type"] == "PlaylistMutationFailedException":
+                raise UnauthorizedException(response_json["errors"][0]["message"])
             else:
+                print(response_json)
                 raise Exception("GraphQL request failed. Unknown JSON Error")
 
         #check if theres data before trying to access it
@@ -264,6 +263,12 @@ def get_playlist_data(self, id: Union[str, int], start: int = 0):
 
         return data
 
+    def get_playlist_state(self, id: Union[str, int]) -> PlaylistState:
+        variables = {"playlistId": str(id)}
+        response = self._request_graphql(GRAPHQL_GET_PLAYLIST_STATE, variables)
+
+        return response["playlist"]
+    
     def create_playlist(self, title:str, description:Optional[str] = None, is_private: bool=False, is_collaborative: bool=False) -> str:
         """
         Create a new playlist. Returns the playlist ID. Playlist cannot be both private and collaborative.
@@ -347,13 +352,13 @@ def edit_playlist(self, playlist_id:str, title: Optional[str] = None, descriptio
         response = self._request_graphql(GRAPHQL_EDIT_PLAYLIST, variables)
         return response["updatePlaylist"]["playlist"]["id"]
 
-    def get_track_batch_data(self, ids: Iterable[str]):
+    def get_track_batch_data(self, ids: ArrayLike[str]):
         json_data = {"sng_ids": tuple(ids)}
         data = self._get_api(METHOD_GET_BATCH_TRACK_DATA, json_data)
 
         return data
 
-    def add_tracks_to_playlist(self, playlist_id: str, song_ids: Iterable[str],offset: int = -1) -> None:
+    def add_tracks_to_playlist(self, playlist_id: str, song_ids: ArrayLike[str],offset: int = -1) -> None:
         """
         Add songs to a playlist.
 
@@ -365,6 +370,9 @@ def add_tracks_to_playlist(self, playlist_id: str, song_ids: Iterable[str],offse
         :type offset: int
         """
 
+        if len(song_ids) == 0:
+            return
+
         # Convert song IDs to [id, position] format
         songs = [[str(song_id), i] for i, song_id in enumerate(song_ids)]
 
@@ -373,10 +381,16 @@ def add_tracks_to_playlist(self, playlist_id: str, song_ids: Iterable[str],offse
             "songs": songs,
             "offset": offset,
         }
-        
-        self._get_api(METHOD_ADD_PLAYLIST_TRACK, json_data)
 
-    def remove_tracks_from_playlist(self, playlist_id: str,song_ids: Iterable[str]) -> None:
+        try: 
+            self._get_api(METHOD_ADD_PLAYLIST_TRACK, json_data)
+        except RequestSpecificError as e:
+            if e.error_type == "REQUEST_ERROR":
+                raise UnauthorizedException(e.error_msg)
+            else:
+                raise e
+
+    def remove_tracks_from_playlist(self, playlist_id: str,song_ids: ArrayLike[str]) -> None:
         """
         Remove songs from a playlist.
 
@@ -385,6 +399,9 @@ def remove_tracks_from_playlist(self, playlist_id: str,song_ids: Iterable[str])
         :param song_ids: A list of song IDs to remove from the playlist
         :type song_ids: ArrayLike[str]
         """
+
+        if len(song_ids) == 0:
+            return
 
         songs = [[int(song_id), i] for i, song_id in enumerate(song_ids)] #convert song IDs to [id, position] format (as integers)
 
@@ -422,7 +439,7 @@ def get_media_data(
         infos = response.json()["data"][0]
 
         if "errors" in infos:
-            raise NoRightOnMedia(infos["errors"][0]["message"])
+            raise UnauthorizedException(infos["errors"][0]["message"])
 
         return infos["media"][0]
 
@@ -451,7 +468,7 @@ def get_favorited_track_ids(self):
 
         return favorited_tracks
 
-    def add_favorite_tracks(self, ids: Iterable[str]):
+    def add_favorite_tracks(self, ids: ArrayLike[str]):
         now = datetime.now()
 
         json_data = {"IDS": tuple(ids)}
@@ -460,7 +477,7 @@ def add_favorite_tracks(self, ids: Iterable[str]):
         for id in ids:
             self.favorited_ids[id] = now
 
-    def remove_favorite_tracks(self, ids: Iterable[str]):
+    def remove_favorite_tracks(self, ids: ArrayLike[str]):
         json_data = {"IDS": tuple(ids)}
         self._get_api(METHOD_REMOVE_FAVORITE_TRACKS, json_data)
 

src/deezergw/exceptions.py

@@ -1,14 +1,22 @@
 class ExpiredException(Exception):
     pass
 
-class NoRightOnMedia(Exception):
-    pass
-
 class UnknownException(Exception):
     pass
 
 class NotFoundException(Exception):
     pass
 
 class UnauthorizedException(Exception):
-    pass
+    pass
+
+class AlreadyExistsError(Exception):
+    pass
+
+class RequestSpecificError(Exception):
+    def __init__(self, error_type: str, error_msg: str) -> None:
+        self.error_type = error_type
+        self.error_msg = error_msg
+
+        self.msg = f"{error_type} - {error_msg}"
+        super().__init__(self.msg)

src/deezergw/resources/playlist.py

@@ -1,7 +1,7 @@
 from datetime import datetime
 from typing import Any, Dict, List, Optional, Union
 from deezergw.api import IMAGE_URL, DeezerAPI
-from deezergw.exceptions import UnknownException
+from deezergw.exceptions import UnauthorizedException, UnknownException
 from deezergw.resources.track import Track
 from deezergw.utils import normalize_track_ids
 
@@ -40,6 +40,11 @@ def __init__(
             if "DATA_MOD" in data
             else None
         )
+        self.editable: bool = data["STATUS"] == 1
+
+        state = api.get_playlist_state(self.id)
+        self.is_private = state["isPrivate"]
+        self.is_collaborative = state["isCollaborative"]
 
         self._author_pic: Optional[str] = (
             data["PARENT_USER_PICTURE"]
@@ -92,6 +97,9 @@ def add_tracks(self, tracks: List[Track], offset: int = -1):
         :param offset: The position to insert the songs at. Default is -1 (add to end of playlist)
         :type offset: int
         """
+        if not self.editable:
+            raise UnauthorizedException("This playlist doesn't seem editable")
+
         ids = normalize_track_ids(tracks)
         self._api.add_tracks_to_playlist(self.id, ids, offset)
 
@@ -113,6 +121,9 @@ def remove_tracks(self, tracks: List[Track]):
         :param tracks: A list of Tracks or track ids to remove from the playlist
         :type tracks: List[Union[str, Track]]
         """
+        if not self.editable:
+            raise UnauthorizedException("This playlist doesn't seem editable")
+
         ids = normalize_track_ids(tracks)
         self._api.remove_tracks_from_playlist(self.id, ids)
 
@@ -127,6 +138,30 @@ def remove_tracks(self, tracks: List[Track]):
 
         self.last_edited = datetime.now()
 
+    def edit_playlist(self, name: Optional[str] = None, description: Optional[str] = None, is_private: Optional[bool] = None, is_collaborative: Optional[bool] = None):
+        """
+        Edit a playlist. Returns the playlist ID. Playlist cannot be both private and collaborative.
+
+        :param name: The new name of the playlist
+        :type name: Optional[str]
+        :param description: The new description of the playlist
+        :type description: Optional[str]
+        :param is_private: Whether the playlist should be private
+        :type is_private: Optional[bool]
+        :param is_collaborative: Whether the playlist should be collaborative
+        :type is_collaborative: Optional[bool]
+        """
+        self._api.edit_playlist(self.id, name, description, is_private, is_collaborative)
+        if name:
+            self.name = name
+        if description:
+            self.description = description
+        if is_private is not None:
+            self.is_private = is_private
+        if is_collaborative is not None:
+            self.is_collaborative = is_collaborative
+
+
     def delete(self):
         """Delete this playlist. Use with care!"""
         self._api.delete_playlist(self.id)

src/deezergw/types.py

@@ -37,3 +37,11 @@ class DownloadInfo(TypedDict):
     quality: QualityType
     file_format: FormatType
     track_id: str
+
+
+class PlaylistState(TypedDict):
+    isPrivate: bool
+    isCollaborative: bool
+
+_array = TypeVar("_array")
+ArrayLike = Union[List[_array], Tuple[_array, ...], Set[_array]]