2a639f557f
Fue un leseo la parte de recordings_of_release, pero ya esta esa parte Me faltan un par de recordings no mas :3
537 lines
16 KiB
Python
537 lines
16 KiB
Python
"""
|
|
Mediador entre musicbrainz y las urls
|
|
Su objetivo es modificar los datos que entrega musicbrainz para que sean
|
|
correspondientes con lo que debe entregar la api, se encarga de casos como
|
|
traducción.
|
|
"""
|
|
import json
|
|
from math import ceil
|
|
|
|
from country_list import countries_for_language
|
|
|
|
import fetcher.musicbrainz as mb
|
|
from fetcher import jobs
|
|
from utils import get_redis_connection
|
|
|
|
|
|
###
|
|
# Utility code
|
|
###
|
|
|
|
def full_country_name(country_code):
|
|
"""Obtiene el nombre de un pais en español dado su codigo de pais"""
|
|
return dict(countries_for_language('es')).get(country_code, country_code)
|
|
|
|
|
|
def translate_artist_type(artist_type):
|
|
"""Traduce los tipos de artista a español"""
|
|
translation = {
|
|
'Person': 'Persona',
|
|
'Group': 'Grupo',
|
|
'Orchestra': 'Orquesta',
|
|
'Choir': 'Coro',
|
|
'Character': 'Personaje',
|
|
'Other': 'Otro',
|
|
}
|
|
return translation.get(artist_type, artist_type)
|
|
|
|
|
|
def find_best_cover(mb_covers):
|
|
"""Intenta obtener la cover art mas apropiada a partir de una lista de estas"""
|
|
only_aproved_front = [x for x in mb_covers.get('images') if x.get('approved', False)
|
|
and x.get('front', False) and not x.get('back', False)]
|
|
if len(only_aproved_front) > 0:
|
|
return only_aproved_front[0]
|
|
|
|
only_aproved = [x for x in mb_covers.get('images') if x.get('approved', False)]
|
|
if len(only_aproved) > 0:
|
|
return only_aproved[0]
|
|
|
|
return mb_covers.get('images')[0]
|
|
|
|
|
|
def paginate(count, limit, page):
|
|
"""Crea un modelo de paginado a partir de la cantidad de elementos, el limite de elementos y la
|
|
pagina actual"""
|
|
return {
|
|
'total': count,
|
|
'current_page': page,
|
|
'last_page': ceil(count / limit),
|
|
'per_page': limit,
|
|
}
|
|
|
|
|
|
###
|
|
# Mapear entidades
|
|
##
|
|
|
|
def map_artist(mb_artist):
|
|
"""Mapea el modelo de artista entregado por musicbrainz a uno propio"""
|
|
artist = {
|
|
'id': mb_artist.get('id'),
|
|
'name': mb_artist.get('name'),
|
|
'sort_name': mb_artist.get('sort_name'),
|
|
'disambiguation': mb_artist.get('disambiguation'),
|
|
'type': translate_artist_type(mb_artist.get('type')),
|
|
'country': full_country_name(mb_artist.get('country')),
|
|
'tags': sorted(mb_artist.get('tags', []), key=lambda tag: tag['count'], reverse=True),
|
|
}
|
|
|
|
return artist
|
|
|
|
|
|
def map_artist_credit(mb_artist_credit):
|
|
"""Mapea el modelo de credito a artista entregado por musicbrainz a uno propio"""
|
|
return {
|
|
'id': mb_artist_credit.get('artist').get('id'),
|
|
'name': mb_artist_credit.get('artist').get('name'),
|
|
'sort_name': mb_artist_credit.get('artist').get('sort_name'),
|
|
'disambiguation': mb_artist_credit.get('artist').get('disambiguation'),
|
|
}
|
|
|
|
|
|
def map_disc(mb_disc, cover_art=None):
|
|
"""Mapea el modelo de disco entregado por musicbrainz a uno propio"""
|
|
disc = {
|
|
'id': mb_disc.get('id'),
|
|
'title': mb_disc.get('title'),
|
|
'disambiguation': mb_disc.get('disambiguation'),
|
|
'first_release_date': mb_disc.get('first_release_date'),
|
|
'primary_type': mb_disc.get('primary_type'),
|
|
'cover_art': cover_art,
|
|
}
|
|
|
|
if len(mb_disc.get('secondary_types', [])) > 0:
|
|
disc['secondary_type'] = mb_disc['secondary_types'][0]
|
|
|
|
return disc
|
|
|
|
|
|
def map_release(mb_release, cover_art=None):
|
|
"""Mapea el modelo de release entregado por musicbrainz a uno propio"""
|
|
return {
|
|
'id': mb_release.get('id'),
|
|
'title': mb_release.get('title'),
|
|
'disambiguation': mb_release.get('disambiguation'),
|
|
'status': mb_release.get('status'),
|
|
'country': mb_release.get('country'),
|
|
'date': mb_release.get('date'),
|
|
'cover_art': cover_art,
|
|
}
|
|
|
|
|
|
def map_media(mb_media):
|
|
"""Transforma una instancia de media dentro de una release a una util para mis propósitos"""
|
|
media = {
|
|
'format': mb_media.get('format'),
|
|
'position': mb_media.get('position'),
|
|
'track_count': mb_media.get('track_count'),
|
|
'recordings': [track.get('recording') for track in mb_media.get('tracks')]
|
|
}
|
|
|
|
return media
|
|
|
|
|
|
def map_recording(mb_recording):
|
|
"""Mapea el modelo de recording entregado por musicbrainz a uno propio"""
|
|
return {
|
|
'id': mb_recording.get('id'),
|
|
'title': mb_recording.get('title'),
|
|
'disambiguation': mb_recording.get('disambiguation'),
|
|
'length': mb_recording.get('length'),
|
|
}
|
|
|
|
|
|
def map_coverart(mb_cover):
|
|
"""Mapea el modelo de coverart entregado por musicbrainz a uno propio"""
|
|
return {
|
|
'image': mb_cover.get('image'),
|
|
'1200': mb_cover.get('thumbnails', {}).get('1200'),
|
|
'500': mb_cover.get('thumbnails', {}).get('500'),
|
|
'250': mb_cover.get('thumbnails', {}).get('250'),
|
|
'large': mb_cover.get('thumbnails', {}).get('large'),
|
|
'small': mb_cover.get('thumbnails', {}).get('small'),
|
|
}
|
|
|
|
|
|
##
|
|
# Obtener entidades
|
|
##
|
|
|
|
##
|
|
# Artistas
|
|
##
|
|
|
|
def get_artist(mbid):
|
|
"""Obtiene un artista desde musicbrainz incluyendo sus tags"""
|
|
|
|
with get_redis_connection() as redis:
|
|
mb_artist = redis.get(f'artist:{mbid}')
|
|
if mb_artist is None:
|
|
mb_artist = mb.get_artist_by_mbid(mbid, includes=['tags'])
|
|
else:
|
|
mb_artist = json.loads(mb_artist)
|
|
|
|
if 'error' in mb_artist:
|
|
return mb_artist
|
|
|
|
jobs.load_artist_on_cache.delay(mbid)
|
|
|
|
return map_artist(mb_artist)
|
|
|
|
|
|
def get_disc(mbid):
|
|
"""Obtiene un disco desde musicbrainz"""
|
|
|
|
with get_redis_connection() as redis:
|
|
mb_disc = redis.get(f'release_group:{mbid}')
|
|
if mb_disc is None:
|
|
mb_disc = mb.get_release_group_by_mbid(mbid, ['artists'])
|
|
else:
|
|
mb_disc = json.loads(mb_disc)
|
|
|
|
if 'error' in mb_disc:
|
|
return mb_disc
|
|
|
|
jobs.load_entities_of_release_group.delay(mbid)
|
|
return map_disc(mb_disc)
|
|
|
|
|
|
def get_discs_of_artist(mbid, limit, page):
|
|
"""Obtiene los discos de un artista desde musicbrainz incluyendo"""
|
|
|
|
offset = limit * (page - 1)
|
|
|
|
mb_discs = []
|
|
total = 0
|
|
|
|
# Si es que tengo un set de release_groups en redis me fijo si es que sus counts coinciden
|
|
# Si es que coinciden significa que se cargaron todos los discos, pero si no, quizás aun no
|
|
# terminan de guardarse, por lo que salto el código de obtención y voy directo a musicbrainz
|
|
with get_redis_connection() as redis:
|
|
key_releases = f'artist:{mbid}:release_groups'
|
|
if key_releases in redis:
|
|
if int(redis.get(f'{key_releases}:count')) == redis.zcard(key_releases):
|
|
mb_discs = [get_disc(mbid) for mbid in redis.zrange(key_releases, offset, limit)]
|
|
total = redis.zcard(key_releases)
|
|
|
|
if len(mb_discs) == 0:
|
|
# Si es que no había ningún disco, enviar a cargar al artista, quizás nunca se a guardado
|
|
# en cache antes
|
|
jobs.load_artist_on_cache.delay(mbid)
|
|
|
|
mb_discs_browse = mb.browse_release_groups(params={'artist': mbid},
|
|
includes=['artist-credits'],
|
|
limit=limit, offset=offset)
|
|
if 'error' in mb_discs_browse:
|
|
return mb_discs_browse
|
|
|
|
mb_discs = mb_discs_browse.get('release_groups')
|
|
total = mb_discs_browse.get('release_group_count')
|
|
|
|
return {
|
|
'paginate': paginate(total, limit, page),
|
|
'discs': [map_disc(disc) for disc in mb_discs]
|
|
}
|
|
|
|
|
|
def get_artist_of_disc(mbid):
|
|
"""Obtiene el artista de un disco"""
|
|
|
|
mb_artist = None
|
|
|
|
with get_redis_connection() as redis:
|
|
key_id = f'release_group:{mbid}:artist'
|
|
if key_id in redis:
|
|
mb_artist = get_artist(redis.get(key_id))
|
|
|
|
if mb_artist is None:
|
|
mb_artist_browse = mb.browse_artists(params={'release-group': mbid},
|
|
includes=['tags'],
|
|
limit=1, offset=0)
|
|
|
|
if 'error' in mb_artist_browse:
|
|
return mb_artist_browse
|
|
|
|
mb_artist = mb_artist_browse.get('artists')[0]
|
|
|
|
jobs.load_artist_on_cache.delay(mb_artist)
|
|
|
|
return {
|
|
'artist': map_artist(mb_artist)
|
|
}
|
|
|
|
|
|
##
|
|
# Releases
|
|
##
|
|
|
|
def get_release(mbid):
|
|
"""Obtiene una release desde musicbrainz incluyendo sus artistas"""
|
|
|
|
with get_redis_connection() as redis:
|
|
mb_release = redis.get(f'release:{mbid}')
|
|
if mb_release is None:
|
|
mb_release = mb.get_release_by_mbid(mbid, includes=['artists'])
|
|
else:
|
|
mb_release = json.loads(mb_release)
|
|
|
|
if 'error' in mb_release:
|
|
return mb_release
|
|
|
|
jobs.load_entities_of_release.delay(mbid)
|
|
|
|
return map_release(mb_release)
|
|
|
|
|
|
def get_releases_of_disc(mbid, limit, page):
|
|
"""Obtiene las releases de un disco desde musicbrainz"""
|
|
|
|
mb_releases = []
|
|
offset = limit * (page - 1)
|
|
total = 0
|
|
|
|
with get_redis_connection() as redis:
|
|
key_releases = f'release_group:{mbid}:releases'
|
|
if key_releases in redis:
|
|
if int(redis.get(f'{key_releases}:count')) == redis.zcard(key_releases):
|
|
mb_releases = [get_release(mbid) for mbid in redis.zrange(key_releases,
|
|
offset,
|
|
limit)]
|
|
total = redis.zcard(key_releases)
|
|
|
|
if len(mb_releases) == 0:
|
|
# Si es que no se encontraron releases antes es probable que nunca se cargo en cache el
|
|
# release group
|
|
jobs.load_entities_of_release_group.delay(mbid)
|
|
|
|
mb_releases = mb.browse_releases(params={'release-group': mbid},
|
|
includes=['artist-credits'],
|
|
limit=limit, offset=limit * (page - 1))
|
|
|
|
if 'error' in mb_releases:
|
|
return mb_releases
|
|
|
|
total = mb_releases.get('release_count')
|
|
mb_releases = mb_releases.get('releases')
|
|
|
|
return {
|
|
'paginate': paginate(total, limit, page),
|
|
'releases': [map_release(release) for release in mb_releases]
|
|
}
|
|
|
|
|
|
def get_artist_of_release(mbid, limit, page):
|
|
"""Obtiene el artista de una release"""
|
|
mb_artist = None
|
|
|
|
with get_redis_connection() as redis:
|
|
key = f'release:{mbid}:artist'
|
|
if key in redis:
|
|
mb_artist = get_artist(redis.get(key))
|
|
|
|
if mb_artist is None:
|
|
mb_artist_browse = mb.browse_artists(params={'release': mbid},
|
|
includes=['tags'],
|
|
limit=limit,
|
|
offset=limit * (page - 1))
|
|
|
|
if 'error' in mb_artist_browse:
|
|
return mb_artist_browse
|
|
|
|
mb_artist = mb_artist_browse.get('artists')[0]
|
|
jobs.load_artist_on_cache.delay(mb_artist)
|
|
|
|
return {
|
|
'artist': map_artist(mb_artist)
|
|
}
|
|
|
|
|
|
##
|
|
# Recordings
|
|
##
|
|
|
|
|
|
def get_recording(mbid):
|
|
"""Obtiene una grabación"""
|
|
|
|
with get_redis_connection() as redis:
|
|
mb_recording = redis.get(f'recording:{mbid}')
|
|
if mb_recording is None:
|
|
mb_recording = mb.get_recording_by_mbid(mbid)
|
|
else:
|
|
mb_recording = json.loads(mb_recording)
|
|
|
|
if 'error' in mb_recording:
|
|
return mb_recording
|
|
|
|
jobs.load_entities_of_recording.delay(mbid)
|
|
|
|
return map_recording(mb_recording)
|
|
|
|
|
|
def get_recordings_of_release(mbid):
|
|
"""Obtiene las grabaciones de una release
|
|
|
|
Realmente no existen grabaciones para este caso de uso, si no, media, el cual representa
|
|
los medios físicos en los que esta una grabación, asi que se entrega una lista de medias con sus
|
|
grabaciones acopladas, todo ordenado y con indexes de orden
|
|
"""
|
|
|
|
medias = []
|
|
|
|
with get_redis_connection() as redis:
|
|
medias_key = f'release:{mbid}:media'
|
|
count = redis.get(f'{medias_key}:count')
|
|
if count and redis.zcard(medias_key) == int(count):
|
|
medias = [json.loads(media) for media in redis.zrange(medias_key, 0, -1)]
|
|
for media in medias:
|
|
recordings_key = f'{medias_key}:{media.get("position")}:recordings'
|
|
recordings_id = redis.zrange(recordings_key, 0, -1)
|
|
media['recordings'] = []
|
|
for recording_id in recordings_id:
|
|
media['recordings'].append(json.loads(redis.get(f'recording:{recording_id}')))
|
|
|
|
if len(medias) == 0:
|
|
# Si es que no habían medias cargadas en cache, puede ser que la release nunca se alla
|
|
# cargado.
|
|
jobs.load_entities_of_release.delay(mbid)
|
|
|
|
mb_release = mb.get_release_by_mbid(mbid, ['recordings'])
|
|
|
|
if 'error' in mb_release:
|
|
return mb_release
|
|
|
|
medias = [map_media(media) for media in mb_release.get('media', [])]
|
|
|
|
return {'medias': medias}
|
|
|
|
|
|
def get_release_of_recording(mbid, limit, page):
|
|
"""Obtiene la release de una grabacion incluyendo los creditos a su artista"""
|
|
mb_releases = mb.browse_releases(params={'recording': mbid}, includes=['artists-credits'],
|
|
limit=limit, offset=limit * (page - 1))
|
|
|
|
if 'error' in mb_releases:
|
|
return mb_releases
|
|
|
|
return {
|
|
'paginate': paginate(mb_releases.get('release_count', 0), limit, page),
|
|
'releases': [map_release(release) for release in mb_releases.get('releases')]
|
|
}
|
|
|
|
|
|
def get_artist_of_recording(mbid, limit, page):
|
|
"""Obtiene el artista de una grabacion"""
|
|
mb_artists = mb.browse_artists(params={'recording': mbid}, limit=limit,
|
|
offset=limit * (page - 1))
|
|
|
|
if 'error' in mb_artists:
|
|
return mb_artists
|
|
|
|
return {
|
|
'paginate': paginate(mb_artists.get('artist_count', 0), limit, page),
|
|
'artists': [map_artist(artist) for artist in mb_artists['artists']]
|
|
}
|
|
|
|
|
|
##
|
|
# CoverArt
|
|
##
|
|
|
|
|
|
def get_cover_art_disc(mbid):
|
|
"""Obtiene el cover art de un disco"""
|
|
with get_redis_connection() as redis:
|
|
mb_covers = redis.get(f'release_group_cover_art_{mbid}')
|
|
if mb_covers is None:
|
|
mb_covers = mb.get_release_group_cover_art(mbid)
|
|
else:
|
|
mb_covers = json.loads(mb_covers)
|
|
|
|
if 'error' in mb_covers:
|
|
return None
|
|
|
|
jobs.load_entities_of_release_group.delay(mbid)
|
|
cover_art = map_coverart(find_best_cover(mb_covers))
|
|
|
|
return cover_art
|
|
|
|
|
|
def get_cover_art_release(mbid):
|
|
"""Obtiene el cover art de una release"""
|
|
mb_covers = mb.get_release_cover_art(mbid)
|
|
if mb_covers is None:
|
|
mb_covers = mb.get_release_cover_art(mbid)
|
|
else:
|
|
mb_covers = json.loads(mb_covers)
|
|
|
|
if 'error' in mb_covers:
|
|
return None
|
|
|
|
jobs.load_entities_of_release.delay(mbid)
|
|
cover_art = map_coverart(find_best_cover(mb_covers))
|
|
|
|
return cover_art
|
|
|
|
|
|
def get_cover_art_recording(mbid):
|
|
"""Obtiene el cover art de una grabacion"""
|
|
release = get_release_of_recording(mbid, limit=1, page=1)
|
|
|
|
if 'error' in release:
|
|
return None
|
|
|
|
return get_cover_art_release(release['releases'][0]['id'])
|
|
|
|
|
|
##
|
|
# Busqueda
|
|
##
|
|
|
|
|
|
def search_artist(query, limit, page):
|
|
"""Busca un artista dada una query"""
|
|
mb_artists = mb.search_artist(query=query, limit=limit, offset=limit * (page - 1))
|
|
|
|
if 'error' in mb_artists:
|
|
return mb_artists
|
|
|
|
return {
|
|
'paginate': paginate(mb_artists['count'], limit, page),
|
|
'artists': [map_artist(artist) for artist in mb_artists['artists']]
|
|
}
|
|
|
|
|
|
def search_disc(query, limit, page):
|
|
"""Busca un disco dada una query"""
|
|
mb_discs = mb.search_release_group(query=query, includes=['artist'], limit=limit,
|
|
offset=limit * (page - 1))
|
|
|
|
return {
|
|
'paginate': paginate(mb_discs['count'], limit, page),
|
|
'discs': [map_disc(disc) for disc in mb_discs['release_groups']]
|
|
}
|
|
|
|
|
|
def search_release(query, limit, page):
|
|
"""Busca una release dada una query"""
|
|
mb_releases = mb.search_release(query=query, includes=['artist'], limit=limit,
|
|
offset=limit * (page - 1))
|
|
|
|
return {
|
|
'paginate': paginate(mb_releases['count'], limit, page),
|
|
'releases': [map_release(release) for release in mb_releases['releases']]
|
|
}
|
|
|
|
|
|
def search_recording(query, limit, page):
|
|
"""Busca una grabacion dada una query"""
|
|
mb_recording = mb.search_recording(query=query, includes=['artist'], limit=limit,
|
|
offset=limit * (page - 1))
|
|
|
|
return {
|
|
'paginate': paginate(mb_recording['count'], limit, page),
|
|
'recordings': [map_recording(recording) for recording in mb_recording['recordings']]
|
|
}
|