365 lines
12 KiB
Python
365 lines
12 KiB
Python
"""
|
|
Modulo que se encarga de realizar requests a musicbrainz, mayormente devuelve el resultado que musicbrainz entrega
|
|
exceptuando los errores.
|
|
"""
|
|
|
|
import logging
|
|
import requests
|
|
from ratelimit import limits, sleep_and_retry
|
|
from urllib.parse import quote, urlencode
|
|
|
|
from utils import replace_key, sanitize_keys, pretty_print_json
|
|
from utils.cache import Cache as cache
|
|
|
|
_headers = {'accept': 'application/json', 'user-agent': 'MusicList/1.0 (danielcortes.xyz)'}
|
|
_mb_host = 'https://musicbrainz.org/ws/2'
|
|
_ca_host = 'http://coverartarchive.org'
|
|
|
|
_log = logging.getLogger(__name__)
|
|
_log.addHandler(logging.NullHandler())
|
|
|
|
|
|
def _do_request(url):
|
|
"""Does a request to an url
|
|
|
|
:param str url: URL where to do the request
|
|
:raises ValueError when url isn't set
|
|
:return the request response
|
|
"""
|
|
if not url:
|
|
raise ValueError('URL cant be empty')
|
|
|
|
_log.info(f'Doing request to "{url}" with headers {_headers}')
|
|
response = requests.get(url, headers=_headers)
|
|
|
|
_log.info(f'Request returned with status code {response.status_code}')
|
|
|
|
return response
|
|
|
|
|
|
@sleep_and_retry
|
|
@limits(calls=1, period=1)
|
|
def _do_request_mb(url):
|
|
"""Does a request to a path of musicbrainz and returns the dictionary representing the json
|
|
response
|
|
|
|
If the response is 200 it will return the full dictionary of the json that the response
|
|
contains, and if is a code 500 it will inform of the server error, any other response
|
|
code will be just appended to a dictionary and returned.
|
|
|
|
:param str url: URL where to do the request
|
|
:raises ValueError when user-agent isn't set
|
|
:return: The dictionary with the response or the status and his an error message
|
|
"""
|
|
|
|
if not _headers['user-agent']:
|
|
raise ValueError('User Agent isn\'t set')
|
|
|
|
r = _do_request(url)
|
|
|
|
if r.status_code == 200:
|
|
response = r.json(object_hook=sanitize_keys)
|
|
elif r.status_code == 500:
|
|
response = {'status': r.status_code, 'error': f'Error del servidor'}
|
|
else:
|
|
response = {'status': r.status_code, 'error': r.json()['error']}
|
|
|
|
return response
|
|
|
|
|
|
def _do_request_ca(url):
|
|
"""Does a request to a path of cover art archive and returns the dictionary representing
|
|
the json response
|
|
|
|
If the response is 200 it will return the full dictionary of the json that the response
|
|
contains, and if is a code 500 it will inform of the server error, any other response
|
|
code will be just appended to a dictionary and returned.
|
|
|
|
:param str url: URL where to do the request
|
|
:raises ValueError when user-agent isn't set
|
|
:return: The dictionary with the response or the status and his an error message
|
|
"""
|
|
if not _headers['user-agent']:
|
|
raise ValueError('User Agent isn\'t set')
|
|
|
|
r = _do_request(url)
|
|
|
|
if r.status_code == 200:
|
|
response = r.json(object_hook=sanitize_keys)
|
|
elif r.status_code == 500:
|
|
response = {'status': r.status_code, 'error': f'Error del servidor'}
|
|
elif r.status_code == 400:
|
|
response = {'status': r.status_code, 'error': f'El mbid es invalido'}
|
|
elif r.status_code == 404:
|
|
response = {'status': r.status_code, 'error': f'No encontrado'}
|
|
elif r.status_code == 405:
|
|
response = {'status': r.status_code, 'error': f'Metodo erroneo'}
|
|
elif r.status_code == 503:
|
|
response = {'status': r.status_code, 'error': f'Rate limit exedido'}
|
|
else:
|
|
response = {'status': r.status_code, 'error': r.json()['error']}
|
|
|
|
return response
|
|
|
|
|
|
def _get(entity_type, mbid, includes=None):
|
|
"""Does a get entity query to musicbrainz
|
|
|
|
:param str entity_type: Type of the entity (artist, release, recording...)
|
|
:param str mbid: MBID of the entity to get
|
|
:param list includes: List of include parameters (recording, releases...)
|
|
|
|
:return: The JSON response
|
|
:rtype: dict
|
|
"""
|
|
|
|
if includes is None:
|
|
_log.info(f'Getting {entity_type} of mbid {mbid}')
|
|
|
|
return _do_request_mb(f'{_mb_host}/{entity_type}/{mbid}')
|
|
else:
|
|
_log.info(f'Getting {entity_type} of mbid {mbid} including {includes}')
|
|
_includes = quote('+'.join(includes))
|
|
|
|
return _do_request_mb(f'{_mb_host}/{entity_type}/{mbid}?inc={_includes}')
|
|
|
|
|
|
def _search(entity_type, query, includes=None, limit=25, offset=0):
|
|
"""Does a search of an entity to musicbrainz
|
|
|
|
:param str entity_type: Type of the entity (artist, release, recording...)
|
|
:param str query: The search string
|
|
:param int limit: Limit of the search, defaults to 25 and has a max of 100
|
|
:param int offset: Offset of the search for paging purposes
|
|
|
|
:return: The JSON response
|
|
:rtype: dict
|
|
"""
|
|
|
|
_log.info(f'Searching {entity_type} with query "{query}" at offset {offset} with limit of {limit} with includes {includes}')
|
|
|
|
if limit >= 0 and offset >= 0:
|
|
_query = {'query': query, 'limit': limit, 'offset': offset}
|
|
else:
|
|
_query = {'query': query}
|
|
|
|
if includes is not None:
|
|
_query['inc'] = '+'.join(includes)
|
|
|
|
return _do_request_mb(f'{_mb_host}/{entity_type}/?{urlencode(_query)}')
|
|
|
|
|
|
def _browse(entity_type, params, includes=None, limit=25, offset=0):
|
|
"""Browses entities of a type with the given parameters
|
|
|
|
:param str entity_type: Type of the entity (artist, release, recording...)
|
|
:param dict params: Dictionary with the parameters to do the search
|
|
:param list includes: List of include parameters (recording, releases...)
|
|
:param int limit: Limit of the search, defaults to 25 and has a max of 100
|
|
:param int offset: Offset of the search for paging purposes
|
|
|
|
:return: The JSON response
|
|
:rtype: dict
|
|
|
|
:raises ValueError if params is not a dictionary
|
|
"""
|
|
|
|
if not isinstance(params, dict):
|
|
raise ValueError('Params must be a dictionary')
|
|
|
|
if includes is None:
|
|
_log.info(f'Browsing {entity_type} with parameters {params} at offset {offset} with limit of {limit}')
|
|
_query = urlencode({**params, 'limit': limit, 'offset': offset})
|
|
else:
|
|
_log.info(f'Browsing {entity_type} with parameters {params} including {includes} at offset {offset} with limit of {limit}')
|
|
_query = urlencode({**params, 'inc': '+'.join(includes), 'limit': limit, 'offset': offset})
|
|
|
|
return _do_request_mb(f'{_mb_host}/{entity_type}?{_query}')
|
|
|
|
|
|
def _ca(entity_type, mbid):
|
|
"""Gets the url of the cover art of a release or release-group
|
|
|
|
:param str entity_type: Type of the entity, could be release or release-group
|
|
:param str mbid: MBID of the entity of whom is the cover art
|
|
|
|
:return: The url of the cover art
|
|
"""
|
|
|
|
_log.info(f'Obtaining the cover art of the entity with type {entity_type} and mbid {mbid}')
|
|
|
|
_url = f'{_ca_host}/{entity_type}/{mbid}'
|
|
|
|
return _do_request_ca(_url)
|
|
|
|
|
|
@cache
|
|
def get_artist_by_mbid(mbid, includes=None):
|
|
"""Get an artist by its mbid
|
|
|
|
:param str mbid: MBID of the artist
|
|
:param list includes: List of include parameters
|
|
:return: dictionary with the response
|
|
"""
|
|
return _get('artist', mbid, includes)
|
|
|
|
|
|
@cache
|
|
def get_release_group_by_mbid(mbid, includes=None):
|
|
"""Get a release group by its mbid
|
|
|
|
:param str mbid: MBID of the release group
|
|
:param list includes: List of include parameters
|
|
:return: dictionary with the response
|
|
"""
|
|
return _get('release-group', mbid, includes)
|
|
|
|
|
|
@cache
|
|
def get_release_by_mbid(mbid, includes=None):
|
|
"""Get a release by its mbid
|
|
|
|
:param str mbid: MBID of the release
|
|
:param list includes: List of include parameters
|
|
:return: dictionary with the response
|
|
"""
|
|
return _get('release', mbid, includes)
|
|
|
|
|
|
@cache
|
|
def get_recording_by_mbid(mbid, includes=None):
|
|
"""Get a recording by its mbid
|
|
|
|
:param str mbid: MBID of the recording
|
|
:param list includes: List of include parameters
|
|
:return: dictionary with the response
|
|
"""
|
|
return _get('recording', mbid, includes)
|
|
|
|
|
|
@cache
|
|
def search_artist(query, includes=None, limit=25, offset=0):
|
|
"""Search an artist by a query string
|
|
|
|
:param str query: Query string
|
|
:param list includes: List of include parameters
|
|
:param int limit: Limit of the search, defaults to 25 and has a max of 100
|
|
:param int offset: Offset of the search for paging purposes
|
|
:return: dictionary with the response
|
|
"""
|
|
return _search('artist', query, includes, limit, offset)
|
|
|
|
|
|
@cache
|
|
def search_release(query, includes=None, limit=25, offset=0):
|
|
"""Search a release by a query string
|
|
|
|
:param str query: Query string
|
|
:param list includes: List of include parameters
|
|
:param int limit: Limit of the search, defaults to 25 and has a max of 100
|
|
:param int offset: Offset of the search for paging purposes
|
|
:return: dictionary with the response
|
|
"""
|
|
return _search('release', query, includes, limit, offset)
|
|
|
|
|
|
@cache
|
|
def search_release_group(query, includes=None, limit=25, offset=0):
|
|
"""Search a release group by a query string
|
|
|
|
:param str query: Query string
|
|
:param list includes: List of include parameters
|
|
:param int limit: Limit of the search, defaults to 25 and has a max of 100
|
|
:param int offset: Offset of the search for paging purposes
|
|
:return: dictionary with the response
|
|
"""
|
|
|
|
return _search('release-group', query, includes, limit, offset)
|
|
|
|
|
|
@cache
|
|
def search_recording(query, includes=None, limit=25, offset=0):
|
|
"""Search a recording by a query string
|
|
|
|
:param str query: Query string
|
|
:param list includes: List of include parameters
|
|
:param int limit: Limit of the search, defaults to 25 and has a max of 100
|
|
:param int offset: Offset of the search for paging purposes
|
|
:return: dictionary with the response
|
|
"""
|
|
return _search('recording', query, includes,limit, offset)
|
|
|
|
|
|
@cache
|
|
def browse_artists(params, includes=None, limit=25, offset=0):
|
|
"""Browse an artist given some params
|
|
|
|
:param dict params: Parameters to do a search
|
|
:param list includes: List of include parameters
|
|
:param int limit: Limit of the search, defaults to 25 and has a max of 100
|
|
:param int offset: Offset of the search for paging purposes
|
|
:return: dictionary with the response
|
|
"""
|
|
return _browse('artist', params, includes, limit, offset)
|
|
|
|
|
|
@cache
|
|
def browse_recordings(params, includes=None, limit=25, offset=0):
|
|
"""Browse through recordings given some params
|
|
|
|
:param dict params: Parameters to do a search
|
|
:param list includes: List of include parameters
|
|
:param int limit: Limit of the search, defaults to 25 and has a max of 100
|
|
:param int offset: Offset of the search for paging purposes
|
|
:return: dictionary with the response
|
|
"""
|
|
return _browse('recording', params, includes, limit, offset)
|
|
|
|
|
|
@cache
|
|
def browse_releases(params, includes=None, limit=25, offset=0):
|
|
"""Browse through releases given some params
|
|
|
|
:param dict params: Parameters to do a search
|
|
:param list includes: List of include parameters
|
|
:param int limit: Limit of the search, defaults to 25 and has a max of 100
|
|
:param int offset: Offset of the search for paging purposes
|
|
:return: dictionary with the response
|
|
"""
|
|
return _browse('release', params, includes, limit, offset)
|
|
|
|
|
|
@cache
|
|
def browse_release_groups(params, includes=None, limit=25, offset=0):
|
|
"""Browse through release groups given some params
|
|
|
|
:param dict params: Parameters to do a search
|
|
:param list includes: List of include parameters
|
|
:param int limit: Limit of the search, defaults to 25 and has a max of 100
|
|
:param int offset: Offset of the search for paging purposes
|
|
:return: dictionary with the response
|
|
"""
|
|
return _browse('release-group', params, includes, limit, offset)
|
|
|
|
|
|
@cache
|
|
def get_release_cover_art(mbid):
|
|
"""Gets the url of the cover art of a release
|
|
|
|
:param str mbid: MBID of the release of whom is the cover art
|
|
|
|
:return: dictionary with the response
|
|
"""
|
|
return _ca('release', mbid)
|
|
|
|
|
|
@cache
|
|
def get_release_group_cover_art(mbid):
|
|
"""Gets the url of the cover art of a release group
|
|
|
|
:param str mbid: MBID of the release group of whom is the cover art
|
|
|
|
:return: dictionary with the response
|
|
"""
|
|
return _ca('release-group', mbid)
|