qobuz-dl/qobuz_dl/qopy.py

217 lines
7.6 KiB
Python
Raw Permalink Normal View History

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
import logging
2020-10-17 18:52:35 +02:00
import time
2020-08-10 04:47:51 +02:00
import requests
from qobuz_dl.exceptions import (
AuthenticationError,
IneligibleError,
InvalidAppIdError,
InvalidAppSecretError,
InvalidQuality,
)
from qobuz_dl.color import GREEN, YELLOW
2020-08-10 04:47:51 +02:00
RESET = "Reset your credentials with 'qobuz-dl -r'"
logger = logging.getLogger(__name__)
2020-08-10 04:47:51 +02:00
class Client:
def __init__(self, email, pwd, app_id, secrets):
logger.info(f"{YELLOW}Logging...")
self.secrets = secrets
2021-03-22 23:57:15 +01:00
self.id = str(app_id)
2020-08-10 04:47:51 +02:00
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,
2023-06-24 09:08:29 +02:00
"Content-Type": "application/json;charset=UTF-8"
2020-10-17 18:52:35 +02:00
}
)
self.base = "https://www.qobuz.com/api.json/0.2/"
2021-11-26 03:37:37 +01:00
self.sec = None
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":
params = {
"extra": "tracks",
"playlist_id": kwargs["id"],
"limit": 500,
"offset": kwargs["offset"],
}
2020-12-05 20:29:27 +01:00
elif epoint == "artist/get":
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":
params = {
"label_id": kwargs["id"],
"limit": 500,
"offset": kwargs["offset"],
"extra": "albums",
}
2021-11-26 01:34:40 +01:00
elif epoint == "favorite/getUserFavorites":
2020-08-10 04:47:51 +02:00
unix = time.time()
2021-11-27 02:56:26 +01:00
# r_sig = "userLibrarygetAlbumsList" + str(unix) + kwargs["sec"]
r_sig = "favoritegetUserFavorites" + str(unix) + kwargs["sec"]
2020-10-17 18:52:35 +02:00
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,
2021-11-26 01:34:40 +01:00
"type": "albums",
2020-08-10 04:47:51 +02:00
"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"]
if int(fmt_id) not in (5, 6, 7, 27):
raise InvalidQuality("Invalid quality id: choose between 5, 6, 7 or 27")
2020-10-17 18:52:35 +02:00
r_sig = "trackgetFileUrlformat_id{}intentstreamtrack_id{}{}{}".format(
2021-11-27 02:56:26 +01:00
fmt_id, track_id, unix, kwargs.get("sec", self.sec)
2020-10-17 18:52:35 +02:00
)
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:
2020-12-07 16:32:37 +01:00
params = kwargs
2020-08-10 04:47:51 +02:00
r = self.session.get(self.base + epoint, params=params)
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:
raise AuthenticationError("Invalid credentials.\n" + RESET)
2020-08-10 04:47:51 +02:00
elif r.status_code == 400:
raise InvalidAppIdError("Invalid app id.\n" + RESET)
2020-08-10 04:47:51 +02:00
else:
logger.info(f"{GREEN}Logged: OK")
2021-11-27 02:56:26 +01:00
elif (
epoint in ["track/getFileUrl", "favorite/getUserFavorites"]
and r.status_code == 400
):
raise InvalidAppSecretError(f"Invalid app secret: {r.json()}.\n" + RESET)
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"]
logger.info(f"{GREEN}Membership: {self.label}")
2020-08-10 04:47:51 +02: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
def get_artist_meta(self, id):
2020-12-05 20:29:27 +01:00
return self.multi_meta("artist/get", "albums_count", id, None)
def get_plist_meta(self, id):
2020-12-05 20:29:27 +01:00
return self.multi_meta("playlist/get", "tracks_count", id, None)
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-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_artists(self, query, limit):
return self.api_call("artist/search", query=query, limit=limit)
def search_playlists(self, query, limit):
return self.api_call("playlist/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 get_favorite_albums(self, offset, limit):
2020-12-07 16:32:37 +01:00
return self.api_call(
"favorite/getUserFavorites", type="albums", offset=offset, limit=limit
)
def get_favorite_tracks(self, offset, limit):
2020-12-07 16:32:37 +01:00
return self.api_call(
"favorite/getUserFavorites", type="tracks", offset=offset, limit=limit
)
def get_favorite_artists(self, offset, limit):
2020-12-07 16:32:37 +01:00
return self.api_call(
"favorite/getUserFavorites", type="artists", offset=offset, limit=limit
)
def get_user_playlists(self, limit):
return self.api_call("playlist/getUserPlaylists", limit=limit)
2020-08-10 04:47:51 +02:00
def test_secret(self, sec):
try:
2021-11-27 02:56:26 +01:00
self.api_call("track/getFileUrl", id=5966783, fmt_id=5, sec=sec)
2020-08-10 04:47:51 +02:00
return True
2021-11-27 02:56:26 +01:00
except InvalidAppSecretError:
2020-08-10 04:47:51 +02:00
return False
def cfg_setup(self):
for secret in self.secrets:
2021-11-27 02:56:26 +01:00
# Falsy secrets
if not secret:
continue
if self.test_secret(secret):
2021-11-26 03:37:37 +01:00
self.sec = secret
break
if self.sec is None:
2021-11-27 02:56:26 +01:00
raise InvalidAppSecretError("Can't find any valid app secret.\n" + RESET)