2020-08-10 04:47:51 +02:00
|
|
|
# Wrapper for Qo-DL Reborn. This is a sligthly modified version
|
|
|
|
# of qopy, originally written by Sorrow446. All credits to the
|
|
|
|
# original author.
|
|
|
|
|
|
|
|
import hashlib
|
2020-10-17 18:52:35 +02:00
|
|
|
import time
|
|
|
|
|
2020-08-10 04:47:51 +02:00
|
|
|
import requests
|
|
|
|
|
2020-10-17 18:52:35 +02:00
|
|
|
from qo_utils import spoofbuz
|
2020-11-20 00:53:23 +01:00
|
|
|
from qo_utils.exceptions import (
|
|
|
|
AuthenticationError,
|
|
|
|
IneligibleError,
|
|
|
|
InvalidAppIdError,
|
|
|
|
InvalidAppSecretError,
|
|
|
|
)
|
2020-08-10 04:47:51 +02:00
|
|
|
|
|
|
|
|
|
|
|
class Client:
|
|
|
|
def __init__(self, email, pwd):
|
2020-10-17 18:52:35 +02:00
|
|
|
print("Getting tokens...")
|
2020-08-10 04:47:51 +02:00
|
|
|
self.spoofer = spoofbuz.Spoofer()
|
|
|
|
self.id = self.spoofer.getAppId()
|
|
|
|
self.session = requests.Session()
|
2020-10-17 18:52:35 +02:00
|
|
|
self.session.headers.update(
|
|
|
|
{
|
2020-12-05 18:57:10 +01:00
|
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:83.0) Gecko/20100101 Firefox/83.0",
|
2020-10-17 18:52:35 +02:00
|
|
|
"X-App-Id": self.id,
|
|
|
|
}
|
|
|
|
)
|
|
|
|
self.base = "https://www.qobuz.com/api.json/0.2/"
|
2020-08-10 04:47:51 +02:00
|
|
|
self.auth(email, pwd)
|
|
|
|
self.cfg_setup()
|
|
|
|
|
|
|
|
def api_call(self, epoint, **kwargs):
|
2020-12-05 20:29:27 +01:00
|
|
|
if epoint == "user/login":
|
2020-10-17 18:52:35 +02:00
|
|
|
params = {
|
|
|
|
"email": kwargs["email"],
|
|
|
|
"password": kwargs["pwd"],
|
|
|
|
"app_id": self.id,
|
|
|
|
}
|
2020-12-05 20:29:27 +01:00
|
|
|
elif epoint == "track/get":
|
2020-10-17 18:52:35 +02:00
|
|
|
params = {"track_id": kwargs["id"]}
|
2020-12-05 20:29:27 +01:00
|
|
|
elif epoint == "album/get":
|
2020-10-17 18:52:35 +02:00
|
|
|
params = {"album_id": kwargs["id"]}
|
2020-12-05 20:29:27 +01:00
|
|
|
elif epoint == "playlist/get":
|
2020-11-20 00:53:23 +01:00
|
|
|
params = {
|
|
|
|
"extra": "tracks",
|
|
|
|
"playlist_id": kwargs["id"],
|
|
|
|
"limit": 500,
|
|
|
|
"offset": kwargs["offset"],
|
|
|
|
}
|
2020-12-05 20:29:27 +01:00
|
|
|
elif epoint == "artist/get":
|
2020-11-20 00:53:23 +01:00
|
|
|
params = {
|
|
|
|
"app_id": self.id,
|
|
|
|
"artist_id": kwargs["id"],
|
|
|
|
"limit": 500,
|
|
|
|
"offset": kwargs["offset"],
|
|
|
|
"extra": "albums",
|
|
|
|
}
|
2020-12-05 20:29:27 +01:00
|
|
|
elif epoint == "label/get":
|
2020-11-20 00:53:23 +01:00
|
|
|
params = {
|
|
|
|
"label_id": kwargs["id"],
|
|
|
|
"limit": 500,
|
|
|
|
"offset": kwargs["offset"],
|
|
|
|
"extra": "albums",
|
|
|
|
}
|
2020-12-05 20:29:27 +01:00
|
|
|
elif epoint == "userLibrary/getAlbumsList":
|
2020-08-10 04:47:51 +02:00
|
|
|
unix = time.time()
|
2020-10-17 18:52:35 +02:00
|
|
|
r_sig = "userLibrarygetAlbumsList" + str(unix) + kwargs["sec"]
|
|
|
|
r_sig_hashed = hashlib.md5(r_sig.encode("utf-8")).hexdigest()
|
|
|
|
params = {
|
2020-08-10 04:47:51 +02:00
|
|
|
"app_id": self.id,
|
|
|
|
"user_auth_token": self.uat,
|
|
|
|
"request_ts": unix,
|
2020-10-17 18:52:35 +02:00
|
|
|
"request_sig": r_sig_hashed,
|
|
|
|
}
|
2020-12-05 20:29:27 +01:00
|
|
|
elif epoint == "track/getFileUrl":
|
2020-08-10 04:47:51 +02:00
|
|
|
unix = time.time()
|
2020-10-17 18:52:35 +02:00
|
|
|
track_id = kwargs["id"]
|
|
|
|
fmt_id = kwargs["fmt_id"]
|
|
|
|
r_sig = "trackgetFileUrlformat_id{}intentstreamtrack_id{}{}{}".format(
|
|
|
|
fmt_id, track_id, unix, self.sec
|
|
|
|
)
|
|
|
|
r_sig_hashed = hashlib.md5(r_sig.encode("utf-8")).hexdigest()
|
|
|
|
params = {
|
2020-08-10 04:47:51 +02:00
|
|
|
"request_ts": unix,
|
|
|
|
"request_sig": r_sig_hashed,
|
|
|
|
"track_id": track_id,
|
|
|
|
"format_id": fmt_id,
|
2020-10-17 18:52:35 +02:00
|
|
|
"intent": "stream",
|
|
|
|
}
|
2020-12-05 21:49:00 +01:00
|
|
|
else:
|
|
|
|
params=kwargs
|
2020-08-10 04:47:51 +02:00
|
|
|
r = self.session.get(self.base + epoint, params=params)
|
|
|
|
# Do ref header.
|
2020-12-05 20:29:27 +01:00
|
|
|
if epoint == "user/login":
|
2020-08-10 04:47:51 +02:00
|
|
|
if r.status_code == 401:
|
2020-10-17 18:52:35 +02:00
|
|
|
raise AuthenticationError("Invalid credentials.")
|
2020-08-10 04:47:51 +02:00
|
|
|
elif r.status_code == 400:
|
2020-10-17 18:52:35 +02:00
|
|
|
raise InvalidAppIdError("Invalid app id.")
|
2020-08-10 04:47:51 +02:00
|
|
|
else:
|
2020-10-17 18:52:35 +02:00
|
|
|
print("Logged: OK")
|
2020-12-05 20:29:27 +01:00
|
|
|
elif epoint in ["track/getFileUrl", "userLibrary/getAlbumsList"]:
|
2020-08-10 04:47:51 +02:00
|
|
|
if r.status_code == 400:
|
2020-10-17 18:52:35 +02:00
|
|
|
raise InvalidAppSecretError("Invalid app secret.")
|
2020-08-10 04:47:51 +02:00
|
|
|
r.raise_for_status()
|
|
|
|
return r.json()
|
|
|
|
|
|
|
|
def auth(self, email, pwd):
|
2020-12-05 20:29:27 +01:00
|
|
|
usr_info = self.api_call("user/login", email=email, pwd=pwd)
|
2020-10-17 18:52:35 +02:00
|
|
|
if not usr_info["user"]["credential"]["parameters"]:
|
2020-08-10 04:47:51 +02:00
|
|
|
raise IneligibleError("Free accounts are not eligible to download tracks.")
|
2020-10-17 18:52:35 +02:00
|
|
|
self.uat = usr_info["user_auth_token"]
|
|
|
|
self.session.headers.update({"X-User-Auth-Token": self.uat})
|
|
|
|
self.label = usr_info["user"]["credential"]["parameters"]["short_label"]
|
|
|
|
print("Membership: {}".format(self.label))
|
2020-08-10 04:47:51 +02:00
|
|
|
|
2020-11-20 00:53:23 +01:00
|
|
|
def multi_meta(self, epoint, key, id, type):
|
|
|
|
total = 1
|
|
|
|
offset = 0
|
|
|
|
while total > 0:
|
|
|
|
if type in ["tracks", "albums"]:
|
|
|
|
j = self.api_call(epoint, id=id, offset=offset, type=type)[type]
|
|
|
|
else:
|
|
|
|
j = self.api_call(epoint, id=id, offset=offset, type=type)
|
|
|
|
if offset == 0:
|
|
|
|
yield j
|
|
|
|
total = j[key] - 500
|
|
|
|
else:
|
|
|
|
yield j
|
|
|
|
total -= 500
|
|
|
|
offset += 500
|
|
|
|
|
2020-08-10 04:47:51 +02:00
|
|
|
def get_album_meta(self, id):
|
2020-12-05 20:29:27 +01:00
|
|
|
return self.api_call("album/get", id=id)
|
2020-08-10 04:47:51 +02:00
|
|
|
|
|
|
|
def get_track_meta(self, id):
|
2020-12-05 20:29:27 +01:00
|
|
|
return self.api_call("track/get", id=id)
|
2020-08-10 04:47:51 +02:00
|
|
|
|
|
|
|
def get_track_url(self, id, fmt_id):
|
2020-12-05 20:29:27 +01:00
|
|
|
return self.api_call("track/getFileUrl", id=id, fmt_id=fmt_id)
|
2020-08-10 04:47:51 +02:00
|
|
|
|
2020-11-20 00:53:23 +01:00
|
|
|
def get_artist_meta(self, id):
|
2020-12-05 20:29:27 +01:00
|
|
|
return self.multi_meta("artist/get", "albums_count", id, None)
|
2020-11-20 00:53:23 +01:00
|
|
|
|
|
|
|
def get_plist_meta(self, id):
|
2020-12-05 20:29:27 +01:00
|
|
|
return self.multi_meta("playlist/get", "tracks_count", id, None)
|
2020-11-20 00:53:23 +01:00
|
|
|
|
|
|
|
def get_label_meta(self, id):
|
2020-12-05 20:29:27 +01:00
|
|
|
return self.multi_meta("label/get", "albums_count", id, None)
|
2020-11-20 00:53:23 +01:00
|
|
|
|
2020-08-10 04:47:51 +02:00
|
|
|
def search_albums(self, query, limit):
|
2020-12-05 20:29:27 +01:00
|
|
|
return self.api_call("album/search", query=query, limit=limit)
|
2020-08-10 04:47:51 +02:00
|
|
|
|
|
|
|
def search_tracks(self, query, limit):
|
2020-12-05 20:29:27 +01:00
|
|
|
return self.api_call("track/search", query=query, limit=limit)
|
2020-08-10 04:47:51 +02:00
|
|
|
|
|
|
|
def test_secret(self, sec):
|
|
|
|
try:
|
2020-12-05 20:29:27 +01:00
|
|
|
r = self.api_call("userLibrary/getAlbumsList", sec=sec)
|
2020-08-10 04:47:51 +02:00
|
|
|
return True
|
|
|
|
except InvalidAppSecretError:
|
|
|
|
return False
|
|
|
|
|
|
|
|
def cfg_setup(self):
|
|
|
|
for secret in self.spoofer.getSecrets().values():
|
|
|
|
if self.test_secret(secret):
|
|
|
|
self.sec = secret
|
|
|
|
break
|