mirror of
https://github.com/krateng/maloja.git
synced 2023-08-10 21:12:55 +03:00
Added MusicBrainz service for metadata, resolves GH-22
This commit is contained in:
Krateng
parent
fe106a3227
commit
29f88539b4
@ -1,136 +0,0 @@
|
||||
import urllib.parse, urllib.request
|
||||
import json
|
||||
import base64
|
||||
from doreah.settings import get_settings
|
||||
from doreah.logging import log
|
||||
import hashlib
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
### PICTURES
|
||||
|
||||
|
||||
apis_artists = []
|
||||
|
||||
if get_settings("LASTFM_API_KEY") not in [None,"ASK"] and get_settings("FANARTTV_API_KEY") not in [None,"ASK"]:
|
||||
apis_artists.append({
|
||||
"name":"LastFM + Fanart.tv",
|
||||
#"check":get_settings("LASTFM_API_KEY") not in [None,"ASK"] and get_settings("FANARTTV_API_KEY") not in [None,"ASK"],
|
||||
"steps":[
|
||||
("get","http://ws.audioscrobbler.com/2.0/?method=artist.getinfo&artist={artiststring}&api_key=" + str(get_settings("LASTFM_API_KEY")) + "&format=json"),
|
||||
("parse",["artist","mbid"]),
|
||||
("get","http://webservice.fanart.tv/v3/music/{var}?api_key=" + str(get_settings("FANARTTV_API_KEY"))),
|
||||
("parse",["artistthumb",0,"url"])
|
||||
]
|
||||
})
|
||||
|
||||
if get_settings("SPOTIFY_API_ID") not in [None,"ASK"] and get_settings("SPOTIFY_API_SECRET") not in [None,"ASK"]:
|
||||
apis_artists.append({
|
||||
"name":"Spotify",
|
||||
#"check":get_settings("SPOTIFY_API_ID") not in [None,"ASK"] and get_settings("SPOTIFY_API_SECRET") not in [None,"ASK"],
|
||||
"steps":[
|
||||
("post","https://accounts.spotify.com/api/token",{"Authorization":"Basic " + base64.b64encode(bytes(get_settings("SPOTIFY_API_ID") + ":" + get_settings("SPOTIFY_API_SECRET"),encoding="utf-8")).decode("utf-8")},{"grant_type":"client_credentials"}),
|
||||
("parse",["access_token"]),
|
||||
("get","https://api.spotify.com/v1/search?q={artiststring}&type=artist&access_token={var}"),
|
||||
("parse",["artists","items",0,"images",0,"url"])
|
||||
]
|
||||
})
|
||||
|
||||
apis_tracks = []
|
||||
|
||||
if get_settings("LASTFM_API_KEY") not in [None,"ASK"]:
|
||||
apis_tracks.append({
|
||||
"name":"LastFM",
|
||||
#"check":get_settings("LASTFM_API_KEY") not in [None,"ASK"],
|
||||
"steps":[
|
||||
("get","https://ws.audioscrobbler.com/2.0/?method=track.getinfo&track={titlestring}&artist={artiststring}&api_key=" + get_settings("LASTFM_API_KEY") + "&format=json"),
|
||||
("parse",["track","album","image",3,"#text"])
|
||||
]
|
||||
})
|
||||
|
||||
if get_settings("SPOTIFY_API_ID") not in [None,"ASK"] and get_settings("SPOTIFY_API_SECRET") not in [None,"ASK"]:
|
||||
apis_tracks.append({
|
||||
"name":"Spotify",
|
||||
#"check":get_settings("SPOTIFY_API_ID") not in [None,"ASK"] and get_settings("SPOTIFY_API_SECRET") not in [None,"ASK"],
|
||||
"steps":[
|
||||
("post","https://accounts.spotify.com/api/token",{"Authorization":"Basic " + base64.b64encode(bytes(get_settings("SPOTIFY_API_ID") + ":" + get_settings("SPOTIFY_API_SECRET"),encoding="utf-8")).decode("utf-8")},{"grant_type":"client_credentials"}),
|
||||
("parse",["access_token"]),
|
||||
("get","https://api.spotify.com/v1/search?q={artiststring}%20{titlestring}&type=track&access_token={var}"),
|
||||
("parse",["tracks","items",0,"album","images",0,"url"])
|
||||
]
|
||||
})
|
||||
|
||||
|
||||
def api_request_artist(artist):
|
||||
for api in apis_artists:
|
||||
if True:
|
||||
try:
|
||||
artiststring = urllib.parse.quote(artist)
|
||||
var = artiststring
|
||||
for step in api["steps"]:
|
||||
if step[0] == "get":
|
||||
response = urllib.request.urlopen(step[1].format(artiststring=artiststring,var=var))
|
||||
var = json.loads(response.read())
|
||||
elif step[0] == "post":
|
||||
keys = {
|
||||
"url":step[1].format(artiststring=artiststring,var=var),
|
||||
"method":"POST",
|
||||
"headers":step[2],
|
||||
"data":bytes(urllib.parse.urlencode(step[3]),encoding="utf-8")
|
||||
}
|
||||
req = urllib.request.Request(**keys)
|
||||
response = urllib.request.urlopen(req)
|
||||
var = json.loads(response.read())
|
||||
elif step[0] == "parse":
|
||||
for node in step[1]:
|
||||
var = var[node]
|
||||
assert isinstance(var,str) and var != ""
|
||||
except Exception as e:
|
||||
log("Error while getting artist image from " + api["name"],module="external")
|
||||
log(str(e),module="external")
|
||||
continue
|
||||
|
||||
return var
|
||||
else:
|
||||
pass
|
||||
return None
|
||||
|
||||
def api_request_track(track):
|
||||
artists, title = track
|
||||
for api in apis_tracks:
|
||||
if True:
|
||||
try:
|
||||
artiststring = urllib.parse.quote(", ".join(artists))
|
||||
titlestring = urllib.parse.quote(title)
|
||||
var = artiststring + titlestring
|
||||
for step in api["steps"]:
|
||||
if step[0] == "get":
|
||||
response = urllib.request.urlopen(step[1].format(artiststring=artiststring,titlestring=titlestring,var=var))
|
||||
var = json.loads(response.read())
|
||||
elif step[0] == "post":
|
||||
keys = {
|
||||
"url":step[1].format(artiststring=artiststring,titlestring=titlestring,var=var),
|
||||
"method":"POST",
|
||||
"headers":step[2],
|
||||
"data":bytes(urllib.parse.urlencode(step[3]),encoding="utf-8")
|
||||
}
|
||||
req = urllib.request.Request(**keys)
|
||||
response = urllib.request.urlopen(req)
|
||||
var = json.loads(response.read())
|
||||
elif step[0] == "parse":
|
||||
for node in step[1]:
|
||||
var = var[node]
|
||||
assert isinstance(var,str) and var != ""
|
||||
except:
|
||||
if len(artists) != 1:
|
||||
# try the same track with every single artist
|
||||
for a in artists:
|
||||
result = api_request_track(([a],title))
|
||||
if result is not None:
|
||||
return result
|
||||
continue
|
||||
|
||||
return var
|
||||
else:
|
||||
pass
|
||||
|
||||
return None
|
||||
15
maloja/thirdparty/__init__.py
vendored
15
maloja/thirdparty/__init__.py
vendored
@ -30,11 +30,19 @@ def proxy_scrobble_all(artists,title,timestamp):
|
||||
def get_image_track_all(track):
|
||||
for service in services["metadata"]:
|
||||
res = service.get_image_track(track)
|
||||
if res is not None: return res
|
||||
if res is not None:
|
||||
log("Got track image for " + str(track) + " from " + service.name)
|
||||
return res
|
||||
else:
|
||||
log("Could not get track image for " + str(track) + " from " + service.name)
|
||||
def get_image_artist_all(artist):
|
||||
for service in services["metadata"]:
|
||||
res = service.get_image_artist(artist)
|
||||
if res is not None: return res
|
||||
if res is not None:
|
||||
log("Got artist image for " + str(artist) + " from " + service.name)
|
||||
return res
|
||||
else:
|
||||
log("Could not get artist image for " + str(artist) + " from " + service.name)
|
||||
|
||||
|
||||
|
||||
@ -189,6 +197,7 @@ def b64(inp):
|
||||
|
||||
__all__ = [
|
||||
"lastfm",
|
||||
"spotify"
|
||||
"spotify",
|
||||
"musicbrainz"
|
||||
]
|
||||
from . import *
|
||||
|
||||
64
maloja/thirdparty/musicbrainz.py
vendored
Normal file
64
maloja/thirdparty/musicbrainz.py
vendored
Normal file
@ -0,0 +1,64 @@
|
||||
from . import MetadataInterface, utf, b64
|
||||
import hashlib
|
||||
import urllib.parse, urllib.request
|
||||
import json
|
||||
import time
|
||||
import threading
|
||||
from ..__pkginfo__ import versionstr, author, links
|
||||
|
||||
class MusicBrainz(MetadataInterface):
|
||||
name = "MusicBrainz"
|
||||
|
||||
# musicbrainz is rate-limited
|
||||
lock = threading.Lock()
|
||||
useragent = "Maloja/" + versionstr + " ( https://github.com/" + author["github"] + "/" + links["github"] + " )"
|
||||
|
||||
settings = {
|
||||
}
|
||||
|
||||
metadata = {
|
||||
"response_type":"json",
|
||||
"response_parse_tree_track": ["images",0,"image"],
|
||||
"required_settings": [],
|
||||
"activated_setting": "METADATA_MUSICBRAINZ"
|
||||
}
|
||||
|
||||
def get_image_artist(self,artist):
|
||||
return None
|
||||
# not supported
|
||||
|
||||
|
||||
def get_image_track(self,track):
|
||||
self.lock.acquire()
|
||||
try:
|
||||
artists, title = track
|
||||
artiststring = urllib.parse.quote(", ".join(artists))
|
||||
titlestring = urllib.parse.quote(title)
|
||||
querystr = urllib.parse.urlencode({
|
||||
"fmt":"json",
|
||||
"query":"{title} {artist}".format(artist=artiststring,title=titlestring)
|
||||
})
|
||||
req = urllib.request.Request(**{
|
||||
"url":"https://musicbrainz.org/ws/2/release?" + querystr,
|
||||
"method":"GET",
|
||||
"headers":{
|
||||
"User-Agent":self.useragent
|
||||
}
|
||||
})
|
||||
response = urllib.request.urlopen(req)
|
||||
responsedata = response.read()
|
||||
data = json.loads(responsedata)
|
||||
mbid = data["releases"][0]["id"]
|
||||
response = urllib.request.urlopen(
|
||||
"https://coverartarchive.org/release/{mbid}?fmt=json".format(mbid=mbid)
|
||||
)
|
||||
responsedata = response.read()
|
||||
data = json.loads(responsedata)
|
||||
return self.metadata_parse_response_track(data)
|
||||
|
||||
except:
|
||||
raise
|
||||
return None
|
||||
finally:
|
||||
time.sleep(2)
|
||||
self.lock.release()
|
||||
1
maloja/thirdparty/spotify.py
vendored
1
maloja/thirdparty/spotify.py
vendored
@ -34,5 +34,4 @@ class Spotify(MetadataInterface):
|
||||
req = urllib.request.Request(**keys)
|
||||
response = urllib.request.urlopen(req)
|
||||
self.settings["token"] = json.loads(response.read())["access_token"]
|
||||
print(self.settings)
|
||||
return True
|
||||
|
||||
Loading…
Reference in New Issue
Block a user