Big Debug of PMOQobuz step 3
This commit is contained in:
342
pmoqobuz/test_python_qobuz/raw.py
Executable file
342
pmoqobuz/test_python_qobuz/raw.py
Executable file
@@ -0,0 +1,342 @@
|
||||
"""
|
||||
qobuz.api.raw
|
||||
~~~~~~~~~~~~~
|
||||
|
||||
Our base api, all method are mapped like in <endpoint>_<method>
|
||||
see Qobuz API on GitHub (https://github.com/Qobuz/api-documentation)
|
||||
|
||||
:part_of: xbmc-qobuz
|
||||
:copyright: (c) 2012 by Joachim Basmaison, Cyril Leclerc
|
||||
:license: GPLv3, see LICENSE for more details.
|
||||
"""
|
||||
|
||||
import sys
|
||||
import time
|
||||
import math
|
||||
import hashlib
|
||||
import socket
|
||||
import binascii
|
||||
from itertools import cycle
|
||||
import requests
|
||||
import spoofbuz
|
||||
|
||||
socket.timeout = 5
|
||||
|
||||
_loglevel = 3
|
||||
|
||||
|
||||
def debug(s):
|
||||
if _loglevel >= 4:
|
||||
print("%s" % s, file=sys.stderr)
|
||||
|
||||
|
||||
def warn(s):
|
||||
if _loglevel >= 3:
|
||||
print("%s" % s, file=sys.stderr)
|
||||
|
||||
|
||||
class RawApi(object):
|
||||
|
||||
def __init__(self, appid, configvalue):
|
||||
if appid and configvalue:
|
||||
self.configvalue = configvalue
|
||||
self.appid = appid
|
||||
self.__set_s4()
|
||||
else:
|
||||
self.spoofer = spoofbuz.Spoofer()
|
||||
self.appid = self.spoofer.getAppId()
|
||||
|
||||
self.version = "0.2"
|
||||
self.baseUrl = "http://localhost:8080/api.json/"
|
||||
self.user_auth_token = None
|
||||
self.user_id = None
|
||||
self.error = None
|
||||
self.status_code = None
|
||||
self._baseUrl = self.baseUrl + self.version
|
||||
self.session = requests.Session()
|
||||
self.error = None
|
||||
|
||||
def _api_error_string(self, request, url="", params={}, json=""):
|
||||
return (
|
||||
"{reason} (code={status_code})\n"
|
||||
"url={url}\nparams={params}"
|
||||
"\njson={json}".format(
|
||||
reason=request.reason,
|
||||
status_code=self.status_code,
|
||||
url=url,
|
||||
params=str(["%s: %s" % (k, v) for k, v in params.items()]),
|
||||
json=str(json),
|
||||
)
|
||||
)
|
||||
|
||||
def _check_ka(self, ka, mandatory, allowed=[]):
|
||||
"""Checking parameters before sending our request
|
||||
- if mandatory parameter is missing raise error
|
||||
- if a given parameter is neither in mandatory or allowed
|
||||
raise error
|
||||
"""
|
||||
for label in mandatory:
|
||||
if not label in ka:
|
||||
raise Exception("Qobuz: missing parameter [%s]" % label)
|
||||
for label in ka:
|
||||
if label not in mandatory and label not in allowed:
|
||||
raise Exception("Qobuz: invalid parameter [%s]" % label)
|
||||
# Having no parameters set triggers a problem in the pyrequests/Qobuz dialog, don't know
|
||||
# where the bug is, but it results in a 411 (length required). So set a very high limit if
|
||||
# nothing is set
|
||||
noparams = True
|
||||
for label in ka:
|
||||
if ka[label]:
|
||||
noparams = False
|
||||
break
|
||||
if noparams:
|
||||
ka["limit"] = "10000"
|
||||
|
||||
def __set_s4(self):
|
||||
"""appid and associated secret is for this app usage only
|
||||
Any use of the API implies your full acceptance of the
|
||||
General Terms and Conditions
|
||||
(http://www.qobuz.com/apps/api/QobuzAPI-TermsofUse.pdf)
|
||||
"""
|
||||
s3b = self.configvalue.encode("ASCII")
|
||||
s3s = binascii.a2b_base64(s3b)
|
||||
bappid = self.appid.encode("ASCII")
|
||||
a = cycle(bappid)
|
||||
b = zip(s3s, a)
|
||||
self.s4 = b"".join((x ^ y).to_bytes(1, byteorder="big") for (x, y) in b)
|
||||
# print("S4: %s"% self.s4.decode('ASCII'), file=sys.stderr)
|
||||
|
||||
def __unset_s4(self, id, sec):
|
||||
a = cycle(id)
|
||||
b = zip(sec, a)
|
||||
bs4 = b"".join((x ^ y).to_bytes(1, byteorder="big") for (x, y) in b)
|
||||
value = binascii.b2a_base64(bs4)
|
||||
return value
|
||||
|
||||
def _api_request(self, params, uri, **opt):
|
||||
"""Qobuz API HTTP get request
|
||||
Arguments:
|
||||
params: parameters dictionary
|
||||
uri : service/method
|
||||
opt : Optional named parameters
|
||||
- noToken=True/False
|
||||
- useGet=True/False
|
||||
|
||||
Return None if something went wrong
|
||||
Return raw data from qobuz on success as dictionary
|
||||
|
||||
* on error you can check error and status_code
|
||||
|
||||
Example:
|
||||
|
||||
ret = api._api_request({'username':'foo',
|
||||
'password':'bar'},
|
||||
'user/login', noToken=True)
|
||||
print('Error: %s [%s]' % (api.error, api.status_code))
|
||||
|
||||
This should produce something like:
|
||||
Error: [200]
|
||||
Error: Bad Request [400]
|
||||
"""
|
||||
self.error = ""
|
||||
self.status_code = None
|
||||
url = self._baseUrl + uri
|
||||
useToken = False if (opt and "noToken" in opt) else True
|
||||
useGet = True if (opt and "useGet" in opt) else False
|
||||
headers = {
|
||||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:67.0) Gecko/20100101 Firefox/67.0"
|
||||
}
|
||||
if useToken and self.user_auth_token:
|
||||
headers["X-User-Auth-Token"] = self.user_auth_token
|
||||
headers["X-App-Id"] = self.appid
|
||||
r = None
|
||||
op = "GET" if useGet else "POST"
|
||||
# warn(f"{op} {url} params {params} headers {headers}")
|
||||
debug(f"{op} {url} params {params}")
|
||||
try:
|
||||
if useGet:
|
||||
r = self.session.get(url, params=params, headers=headers)
|
||||
else:
|
||||
r = self.session.post(url, data=params, headers=headers)
|
||||
except:
|
||||
self.error = "Post request fail"
|
||||
warn(self.error)
|
||||
return None
|
||||
debug(f"status_code: {r.status_code}\nheaders: {r.headers}\ncontent: {r.content}")
|
||||
self.status_code = int(r.status_code)
|
||||
if self.status_code != 200:
|
||||
self.error = self._api_error_string(r, url, params)
|
||||
warn(self.error)
|
||||
return None
|
||||
if not r.content:
|
||||
self.error = "Request return no content"
|
||||
warn(self.error)
|
||||
return None
|
||||
|
||||
"""Retry get if connexion fail"""
|
||||
try:
|
||||
response_json = r.json()
|
||||
except Exception as e:
|
||||
warn("Json loads failed to load... retrying!\n{}", repr(e))
|
||||
try:
|
||||
response_json = r.json()
|
||||
except:
|
||||
self.error = "Failed to load json two times...abort"
|
||||
warn(self.error)
|
||||
return None
|
||||
status = None
|
||||
try:
|
||||
status = response_json["status"]
|
||||
except:
|
||||
pass
|
||||
if status == "error":
|
||||
self.error = self._api_error_string(r, url, params, response_json)
|
||||
warn(self.error)
|
||||
return None
|
||||
return response_json
|
||||
|
||||
def logout(self):
|
||||
self.user_auth_token = None
|
||||
self.user_id = None
|
||||
self.logged_on = None
|
||||
|
||||
def user_login(self, **ka):
|
||||
self._check_ka(ka, ["username", "password"], ["email"])
|
||||
data = self._api_request(ka, "/user/login", noToken=True)
|
||||
if (
|
||||
not data
|
||||
or not "user" in data
|
||||
or not "credential" in data["user"]
|
||||
or not "id" in data["user"]
|
||||
or not "parameters" in data["user"]["credential"]
|
||||
):
|
||||
warn("/user/login returns %s" % data)
|
||||
self.logout()
|
||||
return None
|
||||
if not data["user"]["credential"]["parameters"]:
|
||||
warn("Free accounts are not eligible to download tracks.")
|
||||
return None
|
||||
self.user_id = data["user"]["id"]
|
||||
self.user_auth_token = data["user_auth_token"]
|
||||
self.label = data["user"]["credential"]["parameters"]["short_label"]
|
||||
debug("Membership: {}".format(self.label))
|
||||
data["user"]["email"] = ""
|
||||
data["user"]["firstname"] = ""
|
||||
data["user"]["lastname"] = ""
|
||||
self.setSec()
|
||||
return data
|
||||
|
||||
def setSec(self):
|
||||
global _loglevel
|
||||
savedloglevel = _loglevel
|
||||
_loglevel = 1
|
||||
for value in self.spoofer.getSecrets().values():
|
||||
self.s4 = value.encode("utf-8")
|
||||
if self.userlib_getAlbums(sec=self.s4) is not None:
|
||||
# debug("SECRET [%s]"%self.s4)
|
||||
_loglevel = savedloglevel
|
||||
return
|
||||
_loglevel = savedloglevel
|
||||
|
||||
def track_get(self, **ka):
|
||||
self._check_ka(ka, ["track_id"])
|
||||
return self._api_request(ka, "/track/get")
|
||||
|
||||
def track_getFileUrl(self, intent="stream", **ka):
|
||||
self._check_ka(ka, ["format_id", "track_id"])
|
||||
ts = str(time.time())
|
||||
track_id = str(ka["track_id"])
|
||||
fmt_id = str(ka["format_id"])
|
||||
stringvalue = (
|
||||
"trackgetFileUrlformat_id" + fmt_id + "intent" + intent + "track_id" + track_id + ts
|
||||
)
|
||||
stringvalue = stringvalue.encode("ASCII")
|
||||
stringvalue += self.s4
|
||||
rq_sig = str(hashlib.md5(stringvalue).hexdigest())
|
||||
params = {
|
||||
"format_id": fmt_id,
|
||||
"intent": intent,
|
||||
"request_ts": ts,
|
||||
"request_sig": rq_sig,
|
||||
"track_id": track_id,
|
||||
}
|
||||
return self._api_request(params, "/track/getFileUrl")
|
||||
|
||||
def userlib_getAlbums(self, **ka):
|
||||
ts = str(time.time())
|
||||
r_sig = "userLibrarygetAlbumsList" + str(ts) + str(ka["sec"])
|
||||
r_sig_hashed = hashlib.md5(r_sig.encode("utf-8")).hexdigest()
|
||||
params = {
|
||||
"app_id": self.appid,
|
||||
"user_auth_token": self.user_auth_token,
|
||||
"request_ts": ts,
|
||||
"request_sig": r_sig_hashed,
|
||||
}
|
||||
return self._api_request(params, "/userLibrary/getAlbumsList")
|
||||
|
||||
# Currently unused. Check that it works ?
|
||||
def track_search(self, **ka):
|
||||
self._check_ka(ka, ["query"], ["limit"])
|
||||
return self._api_request(ka, "/track/search")
|
||||
|
||||
def album_get(self, **ka):
|
||||
self._check_ka(ka, ["album_id"], ["extra", "limit", "offset"])
|
||||
# As of around sept 2024, using a POST for this does not work any more (always return the
|
||||
# same album, not the one requested. Probably an inadvertant change on the Qobuz side. The
|
||||
# WEB player uses a GET
|
||||
return self._api_request(ka, "/album/get", useGet=True)
|
||||
|
||||
def album_getFeatured(self, **ka):
|
||||
self._check_ka(ka, [], ["type", "genre_ids", "limit", "offset"])
|
||||
return self._api_request(ka, "/album/getFeatured", useGet=True)
|
||||
|
||||
def favorite_getUserFavorites(self, **ka):
|
||||
self._check_ka(ka, [], ["user_id", "type", "limit", "offset"])
|
||||
return self._api_request(ka, "/favorite/getUserFavorites")
|
||||
|
||||
def playlist_get(self, **ka):
|
||||
self._check_ka(ka, ["playlist_id"], ["extra", "limit", "offset"])
|
||||
return self._api_request(ka, "/playlist/get", useGet=True)
|
||||
|
||||
def playlist_getFeatured(self, **ka):
|
||||
# type is 'last-created' or 'editor-picks'
|
||||
self._check_ka(ka, ["type"], ["genre_ids", "tags", "limit", "offset"])
|
||||
for k in ("tags", "genre_ids"):
|
||||
if k in ka and ka[k] == "None":
|
||||
del ka[k]
|
||||
return self._api_request(ka, "/playlist/getFeatured", useGet=True)
|
||||
|
||||
def playlist_getUserPlaylists(self, **ka):
|
||||
self._check_ka(ka, [], ["user_id", "username", "order", "offset", "limit"])
|
||||
if not "user_id" in ka and not "username" in ka:
|
||||
ka["user_id"] = self.user_id
|
||||
return self._api_request(ka, "/playlist/getUserPlaylists")
|
||||
|
||||
def artist_getSimilarArtists(self, **ka):
|
||||
self._check_ka(ka, ["artist_id"], ["limit", "offset"])
|
||||
return self._api_request(ka, "/artist/getSimilarArtists", useGet=True)
|
||||
|
||||
def artist_get(self, **ka):
|
||||
self._check_ka(ka, ["artist_id"], ["extra", "limit", "offset"])
|
||||
return self._api_request(ka, "/artist/get", useGet=True)
|
||||
|
||||
def genre_list(self, **ka):
|
||||
self._check_ka(ka, [], ["parent_id", "limit", "offset"])
|
||||
return self._api_request(ka, "/genre/list")
|
||||
|
||||
def label_list(self, **ka):
|
||||
self._check_ka(ka, [], ["limit", "offset"])
|
||||
return self._api_request(ka, "/label/list")
|
||||
|
||||
def catalog_search(self, **ka):
|
||||
# type may be 'tracks', 'albums', 'artists' or 'playlists'
|
||||
self._check_ka(ka, ["query"], ["type", "offset", "limit"])
|
||||
return self._api_request(ka, "/catalog/search", useGet=True)
|
||||
|
||||
#### 2024-10 Except for search the /catalog/ methods still work but they're not used by the site
|
||||
#### afaics, replaced by /albums/getFeatured, /playlists/getFeatured
|
||||
def catalog_getFeatured(self, **ka):
|
||||
return self._api_request(ka, "/catalog/getFeatured")
|
||||
|
||||
def catalog_getFeaturedTypes(self, **ka):
|
||||
return self._api_request(ka, "/catalog/getFeaturedTypes")
|
||||
Reference in New Issue
Block a user