"""Interface to the ows-lyrics microservice.""" import json from oto import response from oto import status as status_code from owsrequest import request from backend.constants import error from backend.constants import header from backend.constants import services from backend.exceptions import RequestError from backend.utils import model_utils def get_track_lyrics(track_id): """Get lyrics for track. Args: track_id (int): id of the track Returns: oto.response.Response """ url = '/tracks/{0}/lyrics'.format(track_id) result = request.get( services.OWS_LYRICS, url, headers={header.CONTENT_TYPE: header.CONTENT_TYPE_JSON_VAL}) if result.status_code == status_code.OK: return response.Response(message=result.json()) return model_utils.process_error_response( result, code=error.OWS_LYRICS_ERROR_CODE) def put_track_lyrics(track_id, lyrics): """Put lyrics for track. Args: track_id (int): id of the track lyrics (str): Lyrics data Returns: oto.response.Response """ url = '/tracks/{0}/lyrics'.format(track_id) data = {'lyrics': lyrics} result = request.put( services.OWS_LYRICS, url, data=json.dumps(data, ensure_ascii=False).encode('utf-8'), headers={header.CONTENT_TYPE: header.CONTENT_TYPE_JSON_VAL}) if result.status_code == status_code.OK: return response.Response() return model_utils.process_error_response( result, code=error.OWS_LYRICS_ERROR_CODE) def delete_track_lyrics(track_id): """Delete lyrics for track. Args: track_id (int): id of the track Returns: oto.response.Response """ url = '/tracks/{0}/lyrics'.format(track_id) result = request.delete(services.OWS_LYRICS, url) if result.status_code in [status_code.OK, status_code.NOT_FOUND]: return response.Response() return model_utils.process_error_response( result, code=error.OWS_LYRICS_ERROR_CODE) def delete_tracks_lyrics(track_ids): """Delete lyrics for list of tracks ids. Args: track_ids (list): ids of the tracks Returns: oto.response.Response """ data = {'tuids': list(track_ids)} result = request.delete( services.OWS_LYRICS, '/tracks/lyrics', headers={header.CONTENT_TYPE: header.CONTENT_TYPE_JSON_VAL}, data=json.dumps(data)) if result.status_code == status_code.OK: return response.Response(message=result.json()) return model_utils.process_error_response( result, code=error.OWS_LYRICS_ERROR_CODE) def copy_tracks_lyrics(items): """Copy lyrics for tracks list. Args: items (list): list of source and destination tracks Returns: oto.response.Response """ data = { 'items': [ { 'source_tuid': item['source']['tuid'], 'destination_tuid': item['destination']['tuid'] } for item in items ] } result = request.post( services.OWS_LYRICS, '/tracks/copy', headers={header.CONTENT_TYPE: header.CONTENT_TYPE_JSON_VAL}, data=json.dumps(data)) if result.status_code == status_code.OK: return response.Response() return model_utils.process_error_response( result, code=error.OWS_LYRICS_ERROR_CODE) def has_explicit_lyrics(lyrics): """Find if lyrics contain are explicit. Args: lyrics (string): lyrics Returns: oto.response.Response Boolean """ result = request.post( services.OWS_LYRICS, '/has-explicit-lyrics', headers={header.CONTENT_TYPE: header.CONTENT_TYPE_JSON_VAL}, data=json.dumps({'lyrics': lyrics}, ensure_ascii=False).encode('utf-8')) if result.status_code != status_code.OK: raise RequestError( 'ows-lyrics has-explicit-lyrics failed', error_code=error.OWS_LYRICS_ERROR_CODE, http_status=result.status_code ) return result.json()['has_explicit_lyrics']