Fix token fetcher (close vitiko98#150)

qobuz_dl/bundle.py

@@ -0,0 +1,79 @@
+import base64
+import logging
+import re
+from collections import OrderedDict
+
+from requests import Session
+
+# Modified code based on DashLt's spoofbuz
+
+logger = logging.getLogger(__name__)
+
+_SEED_TIMEZONE_REGEX = re.compile(
+    r'[a-z]\.initialSeed\("(?P<seed>[\w=]+)",window\.utimezone\.(?P<timezone>[a-z]+)\)'
+)
+_INFO_EXTRAS_REGEX = r'name:"\w+/(?P<timezone>{timezones})",info:"(?P<info>[\w=]+)",extras:"(?P<extras>[\w=]+)"'
+_APP_ID_REGEX = re.compile(
+    r'{app_id:"(?P<app_id>\d{9})",app_secret:"\w{32}",base_port:"80",base_url:"https://www\.qobuz\.com",base_method:"/api\.json/0\.2/"},n'
+)
+
+_BUNDLE_URL_REGEX = re.compile(
+    r'<script src="(/resources/\d+\.\d+\.\d+-[a-z]\d{3}/bundle\.js)"></script>'
+)
+
+_BASE_URL = "https://play.qobuz.com"
+_BUNDLE_URL_REGEX = re.compile(
+    r'<script src="(/resources/\d+\.\d+\.\d+-[a-z]\d{3}/bundle\.js)"></script>'
+)
+
+
+class Bundle:
+    def __init__(self):
+        self._session = Session()
+
+        logger.debug("Getting logging page")
+        response = self._session.get(f"{_BASE_URL}/login")
+        response.raise_for_status()
+
+        bundle_url_match = _BUNDLE_URL_REGEX.search(response.text)
+        if not bundle_url_match:
+            raise NotImplementedError("Bundle URL found")
+
+        bundle_url = bundle_url_match.group(1)
+
+        logger.debug("Getting bundle")
+        response = self._session.get(_BASE_URL + bundle_url)
+        response.raise_for_status()
+
+        self._bundle = response.text
+
+    def get_app_id(self):
+        match = _APP_ID_REGEX.search(self._bundle)
+        if not match:
+            raise NotImplementedError("Failed to match APP ID")
+
+        return match.group("app_id")
+
+    def get_secrets(self):
+        logger.debug("Getting secrets")
+        seed_matches = _SEED_TIMEZONE_REGEX.finditer(self._bundle)
+        secrets = OrderedDict()
+
+        for match in seed_matches:
+            seed, timezone = match.group("seed", "timezone")
+            secrets[timezone] = [seed]
+
+        keypairs = list(secrets.items())
+        secrets.move_to_end(keypairs[1][0], last=False)
+        info_extras_regex = _INFO_EXTRAS_REGEX.format(
+            timezones="|".join([timezone.capitalize() for timezone in secrets])
+        )
+        info_extras_matches = re.finditer(info_extras_regex, self._bundle)
+        for match in info_extras_matches:
+            timezone, info, extras = match.group("timezone", "info", "extras")
+            secrets[timezone.lower()] += [info, extras]
+        for secret_pair in secrets:
+            secrets[secret_pair] = base64.standard_b64decode(
+                "".join(secrets[secret_pair])[:-44]
+            ).decode("utf-8")
+        return secrets

qobuz_dl/cli.py

@@ -5,7 +5,7 @@
 import os
 import sys
 
-import qobuz_dl.spoofbuz as spoofbuz
+from qobuz_dl.bundle import Bundle
 from qobuz_dl.color import GREEN, RED, YELLOW
 from qobuz_dl.commands import qobuz_dl_args
 from qobuz_dl.core import QobuzDL
@@ -53,9 +53,9 @@ def _reset_config(config_file):
     config["DEFAULT"]["no_cover"] = "false"
     config["DEFAULT"]["no_database"] = "false"
     logging.info(f"{YELLOW}Getting tokens. Please wait...")
-    spoofer = spoofbuz.Spoofer()
-    config["DEFAULT"]["app_id"] = str(spoofer.getAppId())
-    config["DEFAULT"]["secrets"] = ",".join(spoofer.getSecrets().values())
+    bundle = Bundle()
+    config["DEFAULT"]["app_id"] = str(bundle.get_app_id())
+    config["DEFAULT"]["secrets"] = ",".join(bundle.get_secrets().values())
     config["DEFAULT"]["folder_format"] = DEFAULT_FOLDER
     config["DEFAULT"]["track_format"] = DEFAULT_TRACK
     config["DEFAULT"]["smart_discography"] = "false"

qobuz_dl/core.py

@@ -6,7 +6,7 @@
 from bs4 import BeautifulSoup as bso
 from pathvalidate import sanitize_filename
 
-import qobuz_dl.spoofbuz as spoofbuz
+from qobuz_dl.bundle import Bundle
 from qobuz_dl import downloader, qopy
 from qobuz_dl.color import CYAN, OFF, RED, YELLOW, DF, RESET
 from qobuz_dl.exceptions import NonStreamable
@@ -74,10 +74,10 @@ def initialize_client(self, email, pwd, app_id, secrets):
         logger.info(f"{YELLOW}Set max quality: {QUALITIES[int(self.quality)]}\n")
 
     def get_tokens(self):
-        spoofer = spoofbuz.Spoofer()
-        self.app_id = spoofer.getAppId()
+        bundle = Bundle()
+        self.app_id = bundle.get_app_id()
         self.secrets = [
-            secret for secret in spoofer.getSecrets().values() if secret
+            secret for secret in bundle.get_secrets().values() if secret
         ]  # avoid empty fields
 
     def download_from_id(self, item_id, album=True, alt_path=None):