This commit is contained in:
parent
fe28b1e8e1
commit
d6780481d3
105
src/resources/apiinteractionresource.py
Normal file
105
src/resources/apiinteractionresource.py
Normal file
@ -0,0 +1,105 @@
|
||||
"""
|
||||
Base resource with user handling and Spotify integration
|
||||
"""
|
||||
|
||||
__author__ = '@tormakris'
|
||||
__copyright__ = "Copyright 2020, onSpot Team"
|
||||
__module_name__ = "spotifyuserstoreresource"
|
||||
__version__text__ = "1"
|
||||
|
||||
from flask import current_app
|
||||
from spotipy.oauth2 import SpotifyClientCredentials
|
||||
import spotipy
|
||||
import musicbrainzngs
|
||||
|
||||
from resources.userstoreresource import UserStoreResource
|
||||
from utils.config import SPOTIFY_CLIENT_ID, SPOTIFY_CLIENT_SECRET
|
||||
|
||||
|
||||
class APIInteractionResource(UserStoreResource):
|
||||
spotify = spotipy.Spotify(auth_manager=SpotifyClientCredentials(client_id=SPOTIFY_CLIENT_ID,
|
||||
client_secret=SPOTIFY_CLIENT_SECRET))
|
||||
|
||||
def queryspotifyinfo(self, mb_type: str, title: str = "", artist: str = "", album: str = "") -> tuple:
|
||||
querystring = f"{artist} {album} {title}"
|
||||
if mb_type == "artist":
|
||||
spot_data = self.spotify.search(q=querystring, type="artist", limit=1)['artists']['items']
|
||||
spotify_uri = "spotify:artist:"
|
||||
if spot_data:
|
||||
imagedata = spot_data[0]['images']
|
||||
else:
|
||||
return None, None, None
|
||||
elif mb_type == "release":
|
||||
spot_data = self.spotify.search(q=querystring, type="artist", limit=1).get(
|
||||
'albums')
|
||||
if spot_data:
|
||||
spot_data = spot_data['items']
|
||||
else:
|
||||
return None, None, None
|
||||
spotify_uri = "spotify:album:"
|
||||
if spot_data:
|
||||
imagedata = spot_data[0]['images']
|
||||
else:
|
||||
return None, None, None
|
||||
elif mb_type == "work" or mb_type == "recording":
|
||||
spot_data = self.spotify.search(q=querystring, type="track", limit=1)['tracks']['items']
|
||||
spotify_uri = "spotify:track:"
|
||||
if spot_data:
|
||||
imagedata = spot_data[0]['album']['images']
|
||||
else:
|
||||
return None, None, None
|
||||
else:
|
||||
return None, None, None
|
||||
|
||||
spotify_uri += spot_data[0]['id']
|
||||
if imagedata:
|
||||
return spotify_uri, imagedata[0]['url'], imagedata[len(imagedata) - 1]['url']
|
||||
else:
|
||||
return spotify_uri, None, None
|
||||
|
||||
def getreleaseinfo(self, itemid: str) -> dict:
|
||||
currrelease = musicbrainzngs.get_release_by_id(itemid, includes=['artists'])['release']
|
||||
retdata = {"id": itemid, "album": currrelease['title'], "type": "release"}
|
||||
if 'artist-credit' in currrelease and currrelease['artist-credit']:
|
||||
retdata['artist'] = currrelease['artist-credit'][0]['artist']['name']
|
||||
retdata['spotify_id'], retdata['cover_url'], retdata['cover_url_small'] = \
|
||||
self.queryspotifyinfo('release', artist=retdata['artist'], album=retdata['album'])
|
||||
return retdata
|
||||
|
||||
def getartistinfo(self, itemid: str) -> dict:
|
||||
currartist = musicbrainzngs.get_artist_by_id(itemid)['artist']
|
||||
retdata = {"id": itemid, "artist": currartist['name'], "type": "artist"}
|
||||
retdata['spotify_id'], retdata['cover_url'], retdata['cover_url_small'] = \
|
||||
self.queryspotifyinfo('artist', artist=retdata['artist'])
|
||||
return retdata
|
||||
|
||||
def getworkinfo(self, itemid: str) -> dict:
|
||||
currwork = musicbrainzngs.get_work_by_id(itemid)['work']
|
||||
retdata = {"id": itemid, "title": currwork['title'], "type": "work"}
|
||||
retdata['spotify_id'], retdata['cover_url'], retdata['cover_url_small'] = \
|
||||
self.queryspotifyinfo('work', title=retdata['title'])
|
||||
return retdata
|
||||
|
||||
def getrecordinginfo(self, itemid: str) -> dict:
|
||||
currrecording = musicbrainzngs.get_recording_by_id(itemid, includes=['artists', 'releases'])['recording']
|
||||
retdata = {"id": itemid, "title": currrecording['title'], "type": "recording"}
|
||||
if 'artist-credit' in currrecording and currrecording['artist-credit']:
|
||||
currartist = currrecording['artist-credit'][0]['artist']
|
||||
retdata['artist'] = currartist['name']
|
||||
if 'release-list' in currrecording and currrecording['release-list']:
|
||||
currrlease = currrecording['release-list'][0]
|
||||
retdata['album'] = currrlease['title']
|
||||
try:
|
||||
imgurl = musicbrainzngs.get_image_list(currrlease['id'])['images']
|
||||
if imgurl:
|
||||
retdata['cover_url_small'] = imgurl[0]['thumbnails']['small']
|
||||
retdata['cover_url'] = imgurl[0]['thumbnails']['large']
|
||||
retdata['spotify_id'], _, _ = \
|
||||
self.queryspotifyinfo('recording', title=retdata['title'], album=retdata.get('album'),
|
||||
artist=retdata.get('artist'))
|
||||
except Exception as e:
|
||||
retdata['spotify_id'], retdata['cover_url'], retdata['cover_url_small'] = \
|
||||
self.queryspotifyinfo('recording', title=retdata['title'], album=retdata.get('album'),
|
||||
artist=retdata.get('artist'))
|
||||
current_app.logger.info(e)
|
||||
return retdata
|
@ -7,14 +7,13 @@ __copyright__ = "Copyright 2020, onSpot Team"
|
||||
__module_name__ = "itemapi"
|
||||
__version__text__ = "1"
|
||||
|
||||
import musicbrainzngs
|
||||
from flask import request, current_app, abort
|
||||
|
||||
from flaskaddons.fred import flaskred
|
||||
from resources.spotifyuserstoreresource import SpotifyUserStoreResource
|
||||
from resources.apiinteractionresource import APIInteractionResource
|
||||
|
||||
|
||||
class ItemApi(SpotifyUserStoreResource):
|
||||
class ItemApi(APIInteractionResource):
|
||||
"""
|
||||
See: https://swagger.kmlabz.com/?urls.primaryName=onSpot%20Backend#/backend/getItem
|
||||
"""
|
||||
@ -23,7 +22,7 @@ class ItemApi(SpotifyUserStoreResource):
|
||||
try:
|
||||
flaskred.get(request.headers.get('Authorization')).decode('UTF-8')
|
||||
except Exception as e:
|
||||
current_app.logger.warning(e)
|
||||
current_app.logger.info(e)
|
||||
abort(401, "unauthorized")
|
||||
try:
|
||||
item_type = flaskred.get(itemid).decode('UTF-8')
|
||||
@ -32,69 +31,12 @@ class ItemApi(SpotifyUserStoreResource):
|
||||
abort(404, "unknown item")
|
||||
|
||||
if item_type == 'release':
|
||||
currrelease = musicbrainzngs.get_release_by_id(itemid, includes=['artists'])['release']
|
||||
retdata = {"id": itemid, "album": currrelease['title']}
|
||||
if 'artist-credit' in currrelease and currrelease['artist-credit']:
|
||||
retdata['artist'] = currrelease['artist-credit'][0]['artist']['name']
|
||||
album_spot = \
|
||||
self.spotify.search(q=f"{retdata.get('artist', '')} {retdata['album']}", limit=1).get(
|
||||
'albums')
|
||||
if album_spot:
|
||||
album_spot = album_spot['items']
|
||||
if album_spot:
|
||||
retdata['spotify_id'] = f"spotify:album:{album_spot[0]['id']}"
|
||||
try:
|
||||
imgurl = musicbrainzngs.get_image_list(currrelease['id'])['images']
|
||||
if imgurl:
|
||||
retdata['cover_url_small'] = imgurl[0]['thumbnails']['small']
|
||||
retdata['cover_url'] = imgurl[0]['thumbnails']['large']
|
||||
except Exception as e:
|
||||
current_app.logger.warning(e)
|
||||
if album_spot:
|
||||
retdata['spotify_id'] = f"spotify:album:{album_spot[0]['id']}"
|
||||
return self.getreleaseinfo(itemid), 200
|
||||
elif item_type == 'artist':
|
||||
currartist = musicbrainzngs.get_artist_by_id(itemid)['artist']
|
||||
artist_spot = self.spotify.search(q=currartist['name'], type="artist", limit=1)['artists']['items']
|
||||
retdata = {"id": itemid, "artist": currartist['name']}
|
||||
if artist_spot:
|
||||
retdata['spotify_id'] = f"spotify:artist:{artist_spot[0]['id']}"
|
||||
artist_image = artist_spot[0]['images']
|
||||
if artist_image:
|
||||
retdata['cover_url'] = artist_image[0]['url']
|
||||
retdata['cover_url_small'] = artist_image[len(artist_image) - 1]['url']
|
||||
return self.getartistinfo(itemid), 200
|
||||
elif item_type == 'work':
|
||||
currwork = musicbrainzngs.get_work_by_id(itemid)['work']
|
||||
work_spot = self.spotify.search(q=currwork['title'], type="track", limit=1)['tracks']['items']
|
||||
retdata = {"id": itemid, "title": currwork['title']}
|
||||
if work_spot:
|
||||
retdata['spotify_id'] = f"spotify:track:{work_spot[0]['id']}"
|
||||
work_image = work_spot[0]['album']['images']
|
||||
if work_image:
|
||||
retdata['cover_url'] = work_image[0]['url']
|
||||
retdata['cover_url_small'] = work_image[len(work_image) - 1]['url']
|
||||
return self.getworkinfo(itemid), 200
|
||||
elif item_type == 'recording':
|
||||
currrecording = musicbrainzngs.get_recording_by_id(itemid, includes=['artists', 'releases'])['recording']
|
||||
retdata = {"id": itemid, "title": currrecording['title']}
|
||||
if 'artist-credit' in currrecording and currrecording['artist-credit']:
|
||||
currartist = currrecording['artist-credit'][0]['artist']
|
||||
retdata['artist'] = currartist['name']
|
||||
if 'release-list' in currrecording and currrecording['release-list']:
|
||||
currrlease = currrecording['release-list'][0]
|
||||
retdata['album'] = currrlease['title']
|
||||
try:
|
||||
imgurl = musicbrainzngs.get_image_list(currrlease['id'])['images']
|
||||
if imgurl:
|
||||
retdata['cover_url_small'] = imgurl[0]['thumbnails']['small']
|
||||
retdata['cover_url'] = imgurl[0]['thumbnails']['large']
|
||||
except Exception as e:
|
||||
current_app.logger.warning(e)
|
||||
recording_spot = \
|
||||
self.spotify.search(q=f"{retdata.get('artist', '')} {retdata.get('album', '')} {retdata['title']}",
|
||||
limit=1)['tracks']['items']
|
||||
if recording_spot:
|
||||
retdata['spotify_id'] = f"spotify:track:{recording_spot[0]['id']}"
|
||||
return self.getrecordinginfo(itemid), 200
|
||||
else:
|
||||
abort(417, "wrong type of item")
|
||||
|
||||
retdata['type'] = item_type
|
||||
return retdata, 200
|
||||
|
@ -23,7 +23,7 @@ class ListsApi(UserStoreResource):
|
||||
try:
|
||||
currcreds = self.encryptor.load(flaskred.get(request.headers.get('Authorization')).decode('UTF-8'))
|
||||
except Exception as e:
|
||||
current_app.logger.warning(e)
|
||||
current_app.logger.info(e)
|
||||
abort(401, "unauthorized")
|
||||
|
||||
musicbrainzngs.auth(currcreds['name'], currcreds['password'])
|
||||
|
@ -11,10 +11,10 @@ import musicbrainzngs
|
||||
from flask import request, current_app, abort
|
||||
|
||||
from flaskaddons.fred import flaskred
|
||||
from resources.spotifyuserstoreresource import SpotifyUserStoreResource
|
||||
from resources.apiinteractionresource import APIInteractionResource
|
||||
|
||||
|
||||
class SingleListApi(SpotifyUserStoreResource):
|
||||
class SingleListApi(APIInteractionResource):
|
||||
"""
|
||||
See: https://swagger.kmlabz.com/?urls.primaryName=onSpot%20Backend#/backend/getList
|
||||
"""
|
||||
@ -23,7 +23,7 @@ class SingleListApi(SpotifyUserStoreResource):
|
||||
try:
|
||||
currcreds = self.encryptor.load(flaskred.get(request.headers.get('Authorization')).decode('UTF-8'))
|
||||
except Exception as e:
|
||||
current_app.logger.warning(e)
|
||||
current_app.logger.info(e)
|
||||
abort(401, "unauthorized")
|
||||
try:
|
||||
list_type = flaskred.get(listid).decode('UTF-8')
|
||||
@ -35,90 +35,30 @@ class SingleListApi(SpotifyUserStoreResource):
|
||||
offset = int(request.args.get('offset', 0))
|
||||
if list_type == 'release':
|
||||
currdata = musicbrainzngs.get_releases_in_collection(listid, limit, offset)['collection']
|
||||
releaselist = []
|
||||
itemlist = []
|
||||
for release in currdata['release-list']:
|
||||
releasedata = {"id": release['id'], "album": release['title']}
|
||||
currrelease = musicbrainzngs.get_release_by_id(release['id'], includes=['artists'])
|
||||
if 'artist-credit' in currrelease['release'] and currrelease['release']['artist-credit']:
|
||||
currartist = currrelease['release']['artist-credit'][0]['artist']
|
||||
releasedata['artist'] = currartist['name']
|
||||
album_spot = \
|
||||
self.spotify.search(q=f"{releasedata.get('artist', '')} {release['title']}", limit=1).get(
|
||||
'albums')
|
||||
if album_spot:
|
||||
album_spot = album_spot['items']
|
||||
if album_spot:
|
||||
releasedata['spotify_id'] = f"spotify:album:{album_spot[0]['id']}"
|
||||
try:
|
||||
imgurl = musicbrainzngs.get_image_list(release['id'])['images']
|
||||
if imgurl:
|
||||
releasedata['cover_url'] = imgurl[0]['thumbnails']['large']
|
||||
releasedata['cover_url_small'] = imgurl[0]['thumbnails']['small']
|
||||
except Exception as e:
|
||||
current_app.logger.warning(e)
|
||||
releaselist.append(releasedata)
|
||||
itemlist.append(self.getreleaseinfo(release['id']))
|
||||
flaskred.set(release['id'], 'release'.encode('UTF-8'))
|
||||
retdata = {"id": currdata['id'], "element_count": currdata['release-count'], "itemlist": releaselist}
|
||||
elif list_type == 'artist':
|
||||
currdata = musicbrainzngs.get_artists_in_collection(listid, limit, offset)['collection']
|
||||
artistlist = []
|
||||
itemlist = []
|
||||
for artist in currdata['artist-list']:
|
||||
artist_data = {"id": artist['id'], "artist": artist['name']}
|
||||
artist_spot = self.spotify.search(q=artist['name'], type="artist", limit=1)['artists']['items']
|
||||
if artist_spot:
|
||||
artist_data['spotify_id'] = f"spotify:artist:{artist_spot[0]['id']}"
|
||||
artist_image = artist_spot[0]['images']
|
||||
if artist_image:
|
||||
artist_data['cover_url'] = artist_image[0]['url']
|
||||
artist_data['cover_url_small'] = artist_image[len(artist_image) - 1]['url']
|
||||
artistlist.append(artist_data)
|
||||
itemlist.append(self.getartistinfo(artist['id']))
|
||||
flaskred.set(artist['id'], 'artist'.encode('UTF-8'))
|
||||
retdata = {"id": currdata['id'], "element_count": currdata['artist-count'], "itemlist": artistlist}
|
||||
elif list_type == 'work':
|
||||
currdata = musicbrainzngs.get_works_in_collection(listid, limit, offset)['collection']
|
||||
worklist = []
|
||||
itemlist = []
|
||||
for work in currdata['work-list']:
|
||||
workdata = {"id": work['id'], "title": work['title']}
|
||||
work_spot = self.spotify.search(q=work['title'], type="track", limit=1)['tracks']['items']
|
||||
if work_spot:
|
||||
workdata['spotify_id'] = f"spotify:track:{work_spot[0]['id']}"
|
||||
work_image = work_spot[0]['album']['images']
|
||||
if work_image:
|
||||
workdata['cover_url'] = work_image[0]['url']
|
||||
workdata['cover_url_small'] = work_image[len(work_image) - 1]['url']
|
||||
worklist.append(workdata)
|
||||
itemlist.append(self.getworkinfo(work['id']))
|
||||
flaskred.set(work['id'], 'work'.encode('UTF-8'))
|
||||
retdata = {"id": currdata['id'], "element_count": currdata['work-count'], "itemlist": worklist}
|
||||
elif list_type == 'recording':
|
||||
currdata = musicbrainzngs.get_recordings_in_collection(listid, limit, offset)['collection']
|
||||
recordinglist = []
|
||||
itemlist = []
|
||||
for recording in currdata['recording-list']:
|
||||
currrec = {"id": recording['id'], "title": recording['title']}
|
||||
currrecording = musicbrainzngs.get_recording_by_id(recording['id'], includes=['artists', 'releases'])
|
||||
if 'artist-credit' in currrecording['recording'] and currrecording['recording']['artist-credit']:
|
||||
currartist = currrecording['recording']['artist-credit'][0]['artist']
|
||||
currrec['artist'] = currartist['name']
|
||||
if 'release-list' in currrecording['recording'] and currrecording['recording']['release-list']:
|
||||
currrlease = currrecording['recording']['release-list'][0]
|
||||
currrec['album'] = currrlease['title']
|
||||
try:
|
||||
imgurl = musicbrainzngs.get_image_list(currrlease['id'])['images']
|
||||
if imgurl:
|
||||
currrec['cover_url_small'] = imgurl[0]['thumbnails']['small']
|
||||
currrec['cover_url'] = imgurl[0]['thumbnails']['large']
|
||||
except Exception as e:
|
||||
current_app.logger.warning(e)
|
||||
recording_spot = \
|
||||
self.spotify.search(q=f"{currrec.get('artist', '')} {currrec.get('album', '')} {currrec['title']}",
|
||||
limit=1)['tracks']['items']
|
||||
if recording_spot:
|
||||
currrec['spotify_id'] = f"spotify:track:{recording_spot[0]['id']}"
|
||||
recordinglist.append(currrec)
|
||||
itemlist.append(self.getrecordinginfo(recording['id']))
|
||||
flaskred.set(recording['id'], 'recording'.encode('UTF-8'))
|
||||
retdata = {"id": currdata['id'], "element_count": currdata['recording-count'], "itemlist": recordinglist}
|
||||
else:
|
||||
abort(417, "wrong type of collection")
|
||||
musicbrainzngs.auth(None, None)
|
||||
retdata['type'] = list_type
|
||||
retdata['name'] = currdata['name']
|
||||
return retdata, 200
|
||||
return {"id": currdata['id'], "element_count": currdata[f'{list_type}-count'], "itemlist": itemlist,
|
||||
'type': list_type, 'name': currdata['name']}, 200
|
||||
|
@ -1,19 +0,0 @@
|
||||
"""
|
||||
Base resource with user handling and Spotify integration
|
||||
"""
|
||||
|
||||
__author__ = '@tormakris'
|
||||
__copyright__ = "Copyright 2020, onSpot Team"
|
||||
__module_name__ = "spotifyuserstoreresource"
|
||||
__version__text__ = "1"
|
||||
|
||||
from spotipy.oauth2 import SpotifyClientCredentials
|
||||
import spotipy
|
||||
|
||||
from resources.userstoreresource import UserStoreResource
|
||||
from utils.config import SPOTIFY_CLIENT_ID, SPOTIFY_CLIENT_SECRET
|
||||
|
||||
|
||||
class SpotifyUserStoreResource(UserStoreResource):
|
||||
spotify = spotipy.Spotify(auth_manager=SpotifyClientCredentials(client_id=SPOTIFY_CLIENT_ID,
|
||||
client_secret=SPOTIFY_CLIENT_SECRET))
|
Loading…
Reference in New Issue
Block a user