140 lines
4.2 KiB
Python
140 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import uuid
|
|
|
|
from flask_restful import Resource
|
|
from flask import request, current_app, abort
|
|
import musicbrainzngs
|
|
|
|
from fred import flaskred
|
|
from config import ENCODED_SECRET_KEY
|
|
from schemas import UserSchema
|
|
from aes_encrypt import EncryptedUserRedis
|
|
|
|
"""
|
|
Flask Restful endpoints
|
|
"""
|
|
|
|
__author__ = '@tormakris'
|
|
__copyright__ = "Copyright 2020, onSpot Team"
|
|
__module_name__ = "resources"
|
|
__version__text__ = "1"
|
|
|
|
INVALID_JSON_SCHEMA_MSG = "invalid json schema"
|
|
|
|
|
|
class LoginApi(Resource):
|
|
userschema = UserSchema(many=False)
|
|
encryptor = EncryptedUserRedis(ENCODED_SECRET_KEY)
|
|
|
|
def post(self):
|
|
"""
|
|
See: https://swagger.kmlabz.com/?urls.primaryName=onSpot%20Backend#/backend/logon
|
|
"""
|
|
body = request.get_json()
|
|
|
|
try:
|
|
userobj = self.userschema.load(body)
|
|
except Exception as e:
|
|
current_app.logger.warning(e)
|
|
abort(417, INVALID_JSON_SCHEMA_MSG)
|
|
|
|
try:
|
|
musicbrainzngs.auth(userobj['name'], userobj['password'])
|
|
print(musicbrainzngs.get_collections())
|
|
musicbrainzngs.auth(None, None)
|
|
except Exception as e:
|
|
current_app.logger.warning(e)
|
|
abort(401, "login denied to musicbrainz")
|
|
|
|
self.encryptor.store(body)
|
|
token = str(uuid.uuid4())
|
|
flaskred.set(token, userobj['name'].encode('UTF-8'))
|
|
|
|
return {
|
|
'token': token
|
|
}, 200
|
|
|
|
def delete(self):
|
|
"""
|
|
See: https://swagger.kmlabz.com/?urls.primaryName=onSpot%20Backend#/backend/logoff
|
|
"""
|
|
try:
|
|
flaskred.delete(flaskred.get(request.headers.get('Authorization')).decode('UTF-8'))
|
|
flaskred.delete(request.headers.get('Authorization'))
|
|
except Exception as e:
|
|
current_app.logger.warning(e)
|
|
abort(401, "unauthorized")
|
|
|
|
return "", 204
|
|
|
|
|
|
class MeApi(Resource):
|
|
"""
|
|
See: https://swagger.kmlabz.com/?urls.primaryName=onSpot%20Backend#/backend/currentUser
|
|
"""
|
|
|
|
def get(self):
|
|
try:
|
|
currusername = flaskred.get(request.headers.get('Authorization')).decode('UTF-8')
|
|
except Exception as e:
|
|
current_app.logger.warning(e)
|
|
abort(401, "unauthorized")
|
|
return {"name": currusername}, 200
|
|
|
|
|
|
class ListsApi(Resource):
|
|
"""
|
|
See: https://swagger.kmlabz.com/?urls.primaryName=onSpot%20Backend#/backend/getAllLists
|
|
"""
|
|
encryptor = EncryptedUserRedis(ENCODED_SECRET_KEY)
|
|
|
|
def get(self):
|
|
try:
|
|
currcreds = self.encryptor.load(flaskred.get(request.headers.get('Authorization')).decode('UTF-8'))
|
|
except Exception as e:
|
|
current_app.logger.warning(e)
|
|
abort(401, "unauthorized")
|
|
|
|
musicbrainzngs.auth(currcreds['name'], currcreds['password'])
|
|
collections = musicbrainzngs.get_collections()
|
|
musicbrainzngs.auth(None, None)
|
|
elementlist = []
|
|
for collection in collections['collection-list']:
|
|
if collection['entity-type'] == 'release':
|
|
count = collection['release-count']
|
|
elif collection['entity-type'] == 'artist':
|
|
count = collection['artist-count']
|
|
elif collection['entity-type'] == 'work':
|
|
count = collection['work-count']
|
|
elif collection['entity-type'] == 'recording':
|
|
count = collection['recording-count']
|
|
elif collection['entity-type'] == 'release_group':
|
|
elementlist.append({"id": collection['id'], "name": collection['name']})
|
|
continue
|
|
else:
|
|
continue
|
|
elementlist.append({"id": collection['id'], "name": collection['name'], "element_count": count})
|
|
returndict = {"count": collections['collection-count'],
|
|
"ids": elementlist}
|
|
|
|
return returndict, 200
|
|
|
|
|
|
class SingleListApi(Resource):
|
|
"""
|
|
See: https://swagger.kmlabz.com/?urls.primaryName=onSpot%20Backend#/backend/getList
|
|
"""
|
|
|
|
def get(self, listid: str):
|
|
pass
|
|
|
|
|
|
class TrackApi(Resource):
|
|
"""
|
|
See: https://swagger.kmlabz.com/?urls.primaryName=onSpot%20Backend#/backend/getTrack
|
|
"""
|
|
|
|
def get(self, listid: str, trackid: str):
|
|
pass
|