Add support for Deezloader mp3 downloads

rip/cli.py

@@ -153,7 +153,7 @@ def filter_discography(ctx, **kwargs):
     "-s",
     "--source",
     default="qobuz",
-    help="qobuz, tidal, soundcloud, or deezer",
+    help="qobuz, tidal, soundcloud, deezer, or deezloader",
 )
 @click.argument("QUERY", nargs=-1)
 @click.pass_context
@@ -257,7 +257,7 @@ def discover(ctx, **kwargs):
 @click.option(
     "-s",
     "--source",
-    help="Qobuz, Tidal, Deezer, or SoundCloud. Default: Qobuz.",
+    help="Qobuz, Tidal, Deezer, Deezloader, or SoundCloud. Default: Qobuz.",
 )
 @click.argument("URL")
 @click.pass_context

rip/constants.py

@@ -16,8 +16,7 @@ FAILED_DB_PATH = os.path.join(LOG_DIR, "failed_downloads.db")
 DOWNLOADS_DIR = os.path.join(HOME, "StreamripDownloads")
 
 URL_REGEX = re.compile(
-    r"https?://(?:www|open|play|listen)?\.?(qobuz|tidal|deezer)\.com(?:(?:/"
+    r"https?://(?:www|open|play|listen)?\.?(qobuz|tidal|deezer)\.com(?:(?:/(album|artist|track|playlist|video|label))|(?:\/[-\w]+?))+\/([-\w]+)"
-    r"(album|artist|track|playlist|video|label))|(?:\/[-\w]+?))+\/([-\w]+)"
 )
 SOUNDCLOUD_URL_REGEX = re.compile(r"https://soundcloud.com/[-\w:/]+")
 LASTFM_URL_REGEX = re.compile(r"https://www.last.fm/user/\w+/playlists/\w+")

rip/core.py

@@ -28,6 +28,7 @@ from streamrip.media import (
 from streamrip.clients import (
     Client,
     DeezerClient,
+    DeezloaderClient,
     QobuzClient,
     SoundCloudClient,
     TidalClient,

streamrip/clients.py

@@ -87,7 +87,7 @@ class Client(ABC):
         pass
 
     @abstractmethod
-    def get_file_url(self, track_id, quality=3) -> Union[dict, str]:
+    def get_file_url(self, track_id, quality=3) -> dict:
         """Get the direct download url dict for a file.
 
         :param track_id: id of the track
@@ -629,20 +629,77 @@ class DeezerClient(Client):
         }.get(filetype)
 
 
-def generate_blowfish_key(trackId: str):
+class DeezloaderClient(Client):
-    SECRET = "g4el58wc0zvf9na1"
+
-    md5_hash = hashlib.md5(trackId.encode()).hexdigest()
+    source = "deezer"
-    key = "".join(
+    max_quality = 1
-        chr(ord(md5_hash[i]) ^ ord(md5_hash[i + 16]) ^ ord(SECRET[i]))
+
-        for i in range(16)
+    def __init__(self):
+        self.session = gen_threadsafe_session()
+
+        # no login required
+        self.logged_in = True
+
+    def search(self, query: str, media_type: str = "album", limit: int = 200) -> dict:
+        """Search API for query.
+
+        :param query:
+        :type query: str
+        :param media_type:
+        :type media_type: str
+        :param limit:
+        :type limit: int
+        """
+        # TODO: use limit parameter
+        response = self.session.get(
+            f"{DEEZER_BASE}/search/{media_type}", params={"q": query}
         )
-    return key.encode()
+        response.raise_for_status()
+        return response.json()
 
+    def login(self, **kwargs):
+        """Return None.
 
-def decrypt_chunk(key, data):
+        Dummy method.
-    return Blowfish.new(
+
-        key, Blowfish.MODE_CBC, b"\x00\x01\x02\x03\x04\x05\x06\x07"
+        :param kwargs:
-    ).decrypt(data)
+        """
+        logger.debug("Deezer does not require login call, returning")
+
+    def get(self, meta_id: Union[str, int], media_type: str = "album"):
+        """Get metadata.
+
+        :param meta_id:
+        :type meta_id: Union[str, int]
+        :param type_:
+        :type type_: str
+        """
+        url = f"{DEEZER_BASE}/{media_type}/{meta_id}"
+        item = self.session.get(url).json()
+        if media_type in ("album", "playlist"):
+            tracks = self.session.get(f"{url}/tracks", params={"limit": 1000}).json()
+            item["tracks"] = tracks["data"]
+            item["track_total"] = len(tracks["data"])
+        elif media_type == "artist":
+            albums = self.session.get(f"{url}/albums").json()
+            item["albums"] = albums["data"]
+
+        logger.debug(item)
+        return item
+
+    @staticmethod
+    def get_file_url(meta_id: Union[str, int], quality: int = 6):
+        """Get downloadable url for a track.
+
+        :param meta_id: The track ID.
+        :type meta_id: Union[str, int]
+        :param quality:
+        :type quality: int
+        """
+        quality = min(DEEZER_MAX_Q, quality)
+        url = f"{DEEZER_DL}/{get_quality(quality, 'deezer')}/{DEEZER_BASE}/track/{meta_id}"
+        logger.debug(f"Download url {url}")
+        return {"url": url}
 
 
 class TidalClient(Client):

streamrip/constants.py

@@ -1,7 +1,6 @@
 """Constants that are kept in one place."""
 
 import mutagen.id3 as id3
-import re
 
 AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:83.0) Gecko/20100101 Firefox/83.0"
 

streamrip/media.py

@@ -23,7 +23,7 @@ from mutagen.mp4 import MP4, MP4Cover
 from pathvalidate import sanitize_filepath, sanitize_filename
 
 from . import converter
-from .clients import Client
+from .clients import Client, DeezloaderClient
 from .constants import FLAC_MAX_BLOCKSIZE, FOLDER_FORMAT, TRACK_FORMAT, ALBUM_KEYS
 from .exceptions import (
     InvalidQuality,
@@ -40,6 +40,7 @@ from .utils import (
     downsize_image,
     get_cover_urls,
     decrypt_mqa_file,
+    tqdm_download,
     get_container,
     DownloadStream,
     ext,
@@ -266,7 +267,6 @@ class Track(Media):
             raise NonStreamable(repr(e))
 
         if self.client.source == "qobuz":
-            assert isinstance(dl_info, dict)  # for typing
             if not self.__validate_qobuz_dl_info(dl_info):
                 # click.secho("Track is not available for download", fg="red")
                 raise NonStreamable("Track is not available for download")
@@ -276,7 +276,6 @@ class Track(Media):
 
         # --------- Download Track ----------
         if self.client.source in {"qobuz", "tidal"}:
-            assert isinstance(dl_info, dict), dl_info
             logger.debug("Downloadable URL found: %s", dl_info.get("url"))
             try:
                 download_url = dl_info["url"]
@@ -285,10 +284,12 @@ class Track(Media):
 
             _quick_download(download_url, self.path, desc=self._progress_desc)
 
+        elif isinstance(self.client, DeezloaderClient):
+            tqdm_download(dl_info["url"], self.path, desc=self._progress_desc)
+
         elif self.client.source == "deezer":
             # We can only find out if the requested quality is available
             # after the streaming request is sent for deezer
-            assert isinstance(dl_info, dict)
 
             try:
                 stream = DownloadStream(
@@ -314,7 +315,6 @@ class Track(Media):
                     file.write(chunk)
 
         elif self.client.source == "soundcloud":
-            assert isinstance(dl_info, dict)  # for typing
             self._soundcloud_download(dl_info)
 
         else:
@@ -1345,6 +1345,7 @@ class Album(Tracklist, Media):
         # Generate the folder name
         self.folder_format = kwargs.get("folder_format", FOLDER_FORMAT)
         self.quality = min(kwargs.get("quality", 3), self.client.max_quality)
+        print(f"{self.quality=} {self.client.max_quality = }")
 
         self.folder = self._get_formatted_folder(
             kwargs.get("parent_folder", "StreamripDownloads"), self.quality
@@ -1476,13 +1477,17 @@ class Album(Tracklist, Media):
         """
         fmt = {key: self.get(key) for key in ALBUM_KEYS}
 
+        # Get minimum of both bit depth and sampling rate
         stats = tuple(
-            min(bd, sr)
+            min(stat1, stat2)
-            for bd, sr in zip(
+            for stat1, stat2 in zip(
                 (self.meta.bit_depth, self.meta.sampling_rate),
                 get_stats_from_quality(self.quality),
             )
+            if stat1 is not None and stat2 is not None
         )
+        if not stats:
+            stats = (None, None)
 
         # The quality chosen is not the maximum available quality
         if stats != (fmt.get("sampling_rate"), fmt.get("bit_depth")):
@@ -1496,6 +1501,7 @@ class Album(Tracklist, Media):
             else:
                 fmt["sampling_rate"] = sr / 1000
 
+        logger.debug("Formatter: %s", fmt)
         return fmt
 
     def _get_formatted_folder(self, parent_folder: str, quality: int) -> str:

streamrip/metadata.py

@@ -198,9 +198,9 @@ class TrackMetadata:
             # not embedded
             self.explicit = bool(resp.get("parental_warning"))
             self.quality = 2
-            self.bit_depth = 16
+            self.bit_depth = None
             self.cover_urls = get_cover_urls(resp, self.__source)
-            self.sampling_rate = 44100
+            self.sampling_rate = None
             self.streamable = True
 
         elif self.__source == "soundcloud":