Fix token fetcher (close #150)

This commit is contained in:
vitiko98 2022-05-21 19:49:26 -04:00
parent b9b1d134f1
commit 8aeaf4cd23
4 changed files with 87 additions and 59 deletions

79
qobuz_dl/bundle.py Normal file
View File

@ -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

View File

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

View File

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

View File

@ -1,51 +0,0 @@
import base64
import re
from collections import OrderedDict
import requests
class Spoofer:
def __init__(self):
self.seed_timezone_regex = r'[a-z]\.initialSeed\("(?P<seed>[\w=]+)",window\.utimezone\.(?P<timezone>[a-z]+)\)'
# note: {timezones} should be replaced with every capitalized timezone joined by a |
self.info_extras_regex = r'name:"\w+/(?P<timezone>{timezones})",info:"(?P<info>[\w=]+)",extras:"(?P<extras>[\w=]+)"'
self.appId_regex = 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\.base_url="https://play\.qobuz\.com"'
login_page_request = requests.get("https://play.qobuz.com/login")
login_page = login_page_request.text
bundle_url_match = re.search(
r'<script src="(/resources/\d+\.\d+\.\d+-[a-z]\d{3}/bundle\.js)"></script>',
login_page,
)
bundle_url = bundle_url_match.group(1)
bundle_req = requests.get("https://play.qobuz.com" + bundle_url)
self.bundle = bundle_req.text
def getAppId(self):
return re.search(self.appId_regex, self.bundle).group("app_id")
def getSecrets(self):
seed_matches = re.finditer(self.seed_timezone_regex, self.bundle)
secrets = OrderedDict()
for match in seed_matches:
seed, timezone = match.group("seed", "timezone")
secrets[timezone] = [seed]
"""The code that follows switches around the first and second timezone. Why? Read on:
Qobuz uses two ternary (a shortened if statement) conditions that should always return false.
The way Javascript's ternary syntax works, the second option listed is what runs if the condition returns false.
Because of this, we must prioritize the *second* seed/timezone pair captured, not the first.
"""
keypairs = list(secrets.items())
secrets.move_to_end(keypairs[1][0], last=False)
info_extras_regex = self.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