1e548be114
Resulta que musicbrainz puede responde con 503 a pesar de ser buena gente y mantener el ratelimit, esto puede ser por varias razones documentadas en https://musicbrainz.org/doc/XML_Web_Service/Rate_Limiting
350 lines
12 KiB
Python
350 lines
12 KiB
Python
"""
|
|
Modulo que se encarga de realizar requests a musicbrainz, mayormente devuelve el resultado que
|
|
musicbrainz entrega exceptuando los errores.
|
|
"""
|
|
|
|
from urllib.parse import quote, urlencode
|
|
import logging
|
|
import requests
|
|
from utils import sanitize_keys, ratelimit
|
|
|
|
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')
|
|
if not HEADERS['user-agent']:
|
|
raise ValueError('User Agent isn\'t set')
|
|
|
|
_log.info('Doing request to "%s" with headers %s', url, HEADERS)
|
|
response = requests.get(url, headers=HEADERS)
|
|
|
|
_log.info('Request returned with status code %s', response.status_code)
|
|
|
|
return response
|
|
|
|
|
|
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
|
|
"""
|
|
|
|
while True:
|
|
ratelimit()
|
|
response = _do_request(url)
|
|
if response.status_code != 503:
|
|
break
|
|
|
|
if response.status_code == 200:
|
|
response = response.json(object_hook=sanitize_keys)
|
|
elif response.status_code == 500:
|
|
response = {'status': response.status_code, 'error': 'Error del servidor'}
|
|
else:
|
|
response = {'status': response.status_code, 'error': response.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
|
|
"""
|
|
|
|
response = _do_request(url)
|
|
|
|
if response.status_code == 200:
|
|
response = response.json(object_hook=sanitize_keys)
|
|
elif response.status_code == 500:
|
|
response = {'status': response.status_code, 'error': 'Error del servidor'}
|
|
elif response.status_code == 400:
|
|
response = {'status': response.status_code, 'error': 'El mbid es invalido'}
|
|
elif response.status_code == 404:
|
|
response = {'status': response.status_code, 'error': 'No encontrado'}
|
|
elif response.status_code == 405:
|
|
response = {'status': response.status_code, 'error': 'Metodo erroneo'}
|
|
elif response.status_code == 503:
|
|
response = {'status': response.status_code, 'error': 'Rate limit exedido'}
|
|
else:
|
|
response = {'status': response.status_code, 'error': response.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('Getting %s of mbid %s', entity_type, mbid)
|
|
|
|
return _do_request_mb(f'{MB_HOST}/{entity_type}/{mbid}')
|
|
|
|
_log.info('Getting %s of mbid %s including %s', entity_type, mbid, 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('Searching %s with query "%s" at offset %s with limit of %s with includes %s',
|
|
entity_type, query, offset, limit, 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('Browsing %s with parameters %s at offset %s with limit of %s',
|
|
entity_type, params, offset, limit)
|
|
_query = urlencode({**params, 'limit': limit, 'offset': offset})
|
|
else:
|
|
_log.info('Browsing %s with parameters %s including %s at offset %s with limit of %s',
|
|
entity_type, params, includes, offset, 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('Obtaining the cover art of the entity with type %s and mbid %s', entity_type, mbid)
|
|
|
|
_url = f'{CA_HOST}/{entity_type}/{mbid}'
|
|
|
|
return _do_request_ca(_url)
|
|
|
|
|
|
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)
|
|
|
|
|
|
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)
|
|
|
|
|
|
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)
|
|
|
|
|
|
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)
|
|
|
|
|
|
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)
|
|
|
|
|
|
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)
|
|
|
|
|
|
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)
|
|
|
|
|
|
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)
|
|
|
|
|
|
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)
|
|
|
|
|
|
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)
|
|
|
|
|
|
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)
|
|
|
|
|
|
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)
|
|
|
|
|
|
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)
|
|
|
|
|
|
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)
|