Commit inicial
Habia trabajado un buen poco pero como vi que tenia que separar los repositorios perdi bastante la historia :c
This commit is contained in:
Daniel Cortes
344
fetcher/musicbrainz.py
Normal file
344
fetcher/musicbrainz.py
Normal file
@@ -0,0 +1,344 @@
|
||||
import logging
|
||||
import requests
|
||||
from ratelimit import limits, sleep_and_retry
|
||||
from urllib.parse import quote, urlencode
|
||||
|
||||
from utils import replace_key, sanitize_keys
|
||||
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('musicbrainz')
|
||||
_log.addHandler(logging.NullHandler())
|
||||
|
||||
|
||||
@sleep_and_retry
|
||||
@limits(calls=1, period=1)
|
||||
def _do_request(url, allow_redirects=True):
|
||||
"""Does a request to a path 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 307 code, it will create a dictionary containing the response code
|
||||
and the link where is redirecting, any other response code will be just appended to a
|
||||
dictionary and returning.
|
||||
|
||||
:param str url: URL where to do the request
|
||||
|
||||
:return: The whole response object
|
||||
:raises ValueError when user-agent isn't set
|
||||
:raises ValueError when url isn't set
|
||||
|
||||
:returns A dictionary with the request response
|
||||
it always has a status key with the status code by its value,
|
||||
the rest of the response will be contained in the 'response' key
|
||||
if the request didn't fail
|
||||
"""
|
||||
|
||||
if not _headers['user-agent']:
|
||||
raise ValueError('User Agent isn\'t set')
|
||||
if not url:
|
||||
raise ValueError('URL cant be empty')
|
||||
|
||||
_log.info(f'Doing request to "{url}" with headers {_headers}')
|
||||
r = requests.get(url, headers=_headers, allow_redirects=allow_redirects)
|
||||
|
||||
_log.info(f'Request returned with status code {r.status_code}')
|
||||
|
||||
if r.status_code == 200:
|
||||
response = r.json(object_hook=sanitize_keys)
|
||||
elif r.status_code == 307:
|
||||
response = {'link': r.headers['location']}
|
||||
elif r.status_code == 404:
|
||||
response = {'status': r.status_code, 'error': f'No encontrado'}
|
||||
elif r.status_code == 400:
|
||||
response = {'status': r.status_code, 'error': f'Query erronea'}
|
||||
else:
|
||||
response = {'status': r.status_code, 'error': f'Error desconocido de musicbrainz'}
|
||||
|
||||
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(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(f'{_mb_host}/{entity_type}/{mbid}?inc={_includes}')
|
||||
|
||||
|
||||
def _search(entity_type, query, 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}')
|
||||
|
||||
if limit >= 0 and offset >= 0:
|
||||
_query = urlencode({'query': query, 'limit': limit, 'offset': offset})
|
||||
else:
|
||||
_query = urlencode({'query': query})
|
||||
|
||||
return _do_request(f'{_mb_host}/{entity_type}/?{_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(f'{_mb_host}/{entity_type}?{_query}')
|
||||
|
||||
|
||||
def _ca(entity_type, mbid, size=None):
|
||||
"""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
|
||||
:param int size: Optional size of the cover art, could be 250, 500 or 1200
|
||||
any other value will be ignored
|
||||
|
||||
:return: The url of the cover art
|
||||
"""
|
||||
|
||||
_log.info(f'Obtaining the cover art of the entity with type {entity_type} and mbid {mbid} with size {size}')
|
||||
|
||||
_url = f'{_ca_host}/{entity_type}/{mbid}/front'
|
||||
if size in (250, 500, 1200):
|
||||
_url += f'-{size}'
|
||||
|
||||
return _do_request(_url, allow_redirects=False)
|
||||
|
||||
|
||||
@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
|
||||
"""
|
||||
if includes is None:
|
||||
includes = []
|
||||
|
||||
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
|
||||
"""
|
||||
if includes is None:
|
||||
includes = []
|
||||
|
||||
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
|
||||
"""
|
||||
if includes is None:
|
||||
includes = []
|
||||
|
||||
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
|
||||
"""
|
||||
if includes is None:
|
||||
includes = []
|
||||
|
||||
return _get('recording', mbid, includes)
|
||||
|
||||
|
||||
@cache
|
||||
def search_artist(query, limit=25, offset=0):
|
||||
"""Search an artist by a query string
|
||||
|
||||
:param str query: Query 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: dictionary with the response
|
||||
"""
|
||||
return _search('artist', query, limit, offset)
|
||||
|
||||
|
||||
@cache
|
||||
def search_release(query, limit=25, offset=0):
|
||||
"""Search a release by a query string
|
||||
|
||||
:param str query: Query 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: dictionary with the response
|
||||
"""
|
||||
return _search('release', query, limit, offset)
|
||||
|
||||
|
||||
@cache
|
||||
def search_release_group(query, limit=25, offset=0):
|
||||
"""Search a release group by a query string
|
||||
|
||||
:param str query: Query 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: dictionary with the response
|
||||
"""
|
||||
return _search('release-group', query, limit, offset)
|
||||
|
||||
|
||||
@cache
|
||||
def search_recording(query, limit=25, offset=0):
|
||||
"""Search a recording by a query string
|
||||
|
||||
:param str query: Query 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: dictionary with the response
|
||||
"""
|
||||
return _search('recording', query, 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
|
||||
"""
|
||||
if includes is None:
|
||||
includes = []
|
||||
|
||||
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
|
||||
"""
|
||||
if includes is None:
|
||||
includes = []
|
||||
|
||||
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
|
||||
"""
|
||||
if includes is None:
|
||||
includes = []
|
||||
|
||||
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
|
||||
"""
|
||||
if includes is None:
|
||||
includes = []
|
||||
|
||||
return _browse('release-group', params, includes, limit, offset)
|
||||
|
||||
|
||||
@cache
|
||||
def get_release_cover_art(mbid, size=None):
|
||||
"""Gets the url of the cover art of a release
|
||||
|
||||
:param str mbid: MBID of the release of whom is the cover art
|
||||
:param int size: Optional size of the cover art, could be 250, 500 or 1200
|
||||
|
||||
:return: dictionary with the response
|
||||
"""
|
||||
return _ca('release', mbid, size)
|
||||
|
||||
|
||||
@cache
|
||||
def get_release_group_cover_art(mbid, size=None):
|
||||
"""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
|
||||
:param int size: Optional size of the cover art, could be 250, 500 or 1200
|
||||
|
||||
:return: dictionary with the response
|
||||
"""
|
||||
return _ca('release-group', mbid, size)
|
||||
Reference in New Issue
Block a user