Start paid deezer implementation

poetry.lock

@@ -90,6 +90,17 @@ category = "dev"
 optional = false
 python-versions = ">=3.5"
 
+[[package]]
+name = "deezer-py"
+version = "1.0.4"
+description = "A wrapper for all Deezer's APIs"
+category = "main"
+optional = false
+python-versions = ">=3.6"
+
+[package.dependencies]
+requests = "*"
+
 [[package]]
 name = "docutils"
 version = "0.17.1"
@@ -451,7 +462,7 @@ python-versions = "*"
 [metadata]
 lock-version = "1.1"
 python-versions = "^3.8"
-content-hash = "06048e747453dcda8fc0beb92254466e7e21bf6136be73ae25abe9468fd379a0"
+content-hash = "baac80bc5ff3ccb5a23168ac3303732f79cd16dbafad48a5e216bba531baebd7"
 
 [metadata.files]
 alabaster = [
@@ -490,6 +501,10 @@ decorator = [
     {file = "decorator-5.0.9-py3-none-any.whl", hash = "sha256:6e5c199c16f7a9f0e3a61a4a54b3d27e7dad0dbdde92b944426cb20914376323"},
     {file = "decorator-5.0.9.tar.gz", hash = "sha256:72ecfba4320a893c53f9706bebb2d55c270c1e51a28789361aa93e4a21319ed5"},
 ]
+deezer-py = [
+    {file = "deezer-py-1.0.4.tar.gz", hash = "sha256:73396d09b5ba1b0e3365b6b68b38dd16af71ccb6b825d328cf6740a0cce7a75c"},
+    {file = "deezer_py-1.0.4-py3-none-any.whl", hash = "sha256:ca60481b0799f5818976d2af52a69acb15f75b443d0bdc4d5e70e48013d933ce"},
+]
 docutils = [
     {file = "docutils-0.17.1-py2.py3-none-any.whl", hash = "sha256:cf316c8370a737a022b72b56874f6602acf974a37a9fba42ec2876387549fc61"},
     {file = "docutils-0.17.1.tar.gz", hash = "sha256:686577d2e4c32380bb50cbb22f575ed742d58168cee37e99117a854bcd88f125"},

pyproject.toml

@@ -33,6 +33,7 @@ simple-term-menu = {version = "^1.2.1", platform = 'linux or darwin'}
 pick = {version = "^1.0.0", platform = 'win32 or cygwin'}
 windows-curses = {version = "^2.2.0", platform = 'win32 or cygwin'}
 Pillow = "^8.3.0"
+deezer-py = "^1.0.4"
 
 [tool.poetry.urls]
 "Bug Reports" = "https://github.com/nathom/streamrip/issues"

streamrip/clients.py

@@ -1,11 +1,13 @@
 """The clients that interact with the service APIs."""
 
 import base64
+import binascii
 import hashlib
 import json
 import logging
 import re
 import time
+import deezer
 from abc import ABC, abstractmethod
 from typing import Generator, Sequence, Tuple, Union
 
@@ -438,10 +440,11 @@ class DeezerClient(Client):
 
     def __init__(self):
         """Create a DeezerClient."""
-        self.session = gen_threadsafe_session()
+        self.client = deezer.Deezer(accept_language="en-US,en;q=0.5")
+        # self.session = gen_threadsafe_session()
 
         # no login required
-        self.logged_in = True
+        # self.logged_in = True
 
     def search(self, query: str, media_type: str = "album", limit: int = 200) -> dict:
         """Search API for query.
@@ -467,7 +470,7 @@ class DeezerClient(Client):
 
         :param kwargs:
         """
-        logger.debug("Deezer does not require login call, returning")
+        assert self.client.login_via_arl(kwargs["arl"])
 
     def get(self, meta_id: Union[str, int], media_type: str = "album"):
         """Get metadata.
@@ -477,21 +480,31 @@ class DeezerClient(Client):
         :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
+        GET_FUNCTIONS = {
+            "track": self.client.api.get_track,
+            "album": self.client.api.get_album,
+            "playlist": self.client.api.get_playlist,
+            "artist": self.client.api.get_artist_discography,
+        }
 
-    @staticmethod
-    def get_file_url(meta_id: Union[str, int], quality: int = 6):
+        get_item = GET_FUNCTIONS[media_type]
+        return get_item(meta_id)
+
+        # 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
+
+    def get_file_url(self, meta_id: Union[str, int], quality: int = 2):
         """Get downloadable url for a track.
 
         :param meta_id: The track ID.
@@ -499,10 +512,35 @@ class DeezerClient(Client):
         :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}
+        track_info = self.client.gw.get_track(
+            meta_id,
+        )
+        token = track_info["TRACK_TOKEN"]
+        url = self.client.get_track_url(token, "FLAC")
+        if url is None:
+            md5 = track_info["MD5_ORIGIN"]
+            media_version = track_info["MEDIA_VERSION"]
+            format_number = 1
+
+            url_bytes = b"\xa4".join(
+                [
+                    md5.encode(),
+                    str(format_number).encode(),
+                    str(meta_id).encode(),
+                    str(media_version).encode(),
+                ]
+            )
+
+            md5val = hashlib.md5(url_bytes).hexdigest()
+            step2 = (
+                md5val.encode()
+                + b"\xa4"
+                + url_bytes
+                + b"\xa4"
+                + (b"." * (16 - (len(step2) % 16)))
+            )
+            urlPart = _ecbCrypt("jo6aey6haid2Teih", step2)
+            return urlPart.decode("utf-8")
 
 
 class TidalClient(Client):

streamrip/media.py

@@ -343,7 +343,8 @@ class Track(Media):
         :type path: str
         """
         os.makedirs(os.path.dirname(path), exist_ok=True)
-        shutil.move(self.path, path)
+        shutil.copy(self.path, path)
+        os.remove(self.path)
         self.path = path
 
     def _soundcloud_download(self, dl_info: dict):

streamrip/utils.py

@@ -50,6 +50,27 @@ def safe_get(d: dict, *keys: Hashable, default=None):
     return res
 
 
+__QUALITY_MAP: Dict[str, Dict[int, Union[int, str]]] = {
+    "qobuz": {
+        1: 5,
+        2: 6,
+        3: 7,
+        4: 27,
+    },
+    "deezer": {
+        0: 9,
+        1: 3,
+        2: 1,
+    },
+    "tidal": {
+        0: "LOW",  # AAC
+        1: "HIGH",  # AAC
+        2: "LOSSLESS",  # CD Quality
+        3: "HI_RES",  # MQA
+    },
+}
+
+
 def get_quality(quality_id: int, source: str) -> Union[str, int]:
     """Get the source-specific quality id.
 
@@ -59,33 +80,8 @@ def get_quality(quality_id: int, source: str) -> Union[str, int]:
     :type source: str
     :rtype: Union[str, int]
     """
-    q_map: Dict[int, Union[int, str]]
-    if source == "qobuz":
-        q_map = {
-            1: 5,
-            2: 6,
-            3: 7,
-            4: 27,
-        }
-    elif source == "tidal":
-        q_map = {
-            0: "LOW",  # AAC
-            1: "HIGH",  # AAC
-            2: "LOSSLESS",  # CD Quality
-            3: "HI_RES",  # MQA
-        }
-    elif source == "deezer":
-        q_map = {
-            0: 128,
-            1: 320,
-            2: 1411,
-        }
-    else:
-        raise InvalidSourceError(source)
 
-    possible_keys = set(q_map.keys())
-    assert quality_id in possible_keys, f"{quality_id} must be in {possible_keys}"
-    return q_map[quality_id]
+    return __QUALITY_MAP[source][quality_id]
 
 
 def get_quality_id(bit_depth: Optional[int], sampling_rate: Optional[int]):