mirror of
https://github.com/beetbox/beets.git
synced 2025-12-06 08:39:17 +01:00
Factor out APIAutotaggerPlugin
This commit is contained in:
Rahul Ahuja
parent
4a552595df
commit
bd0cea9f1b
4 changed files with 275 additions and 322 deletions
|
|
@ -18,11 +18,21 @@
|
|||
|
||||
from __future__ import division, absolute_import, print_function
|
||||
|
||||
import re
|
||||
from abc import abstractmethod, abstractproperty
|
||||
|
||||
from beets import logging
|
||||
from beets import config
|
||||
from beets.plugins import BeetsPlugin
|
||||
|
||||
# Parts of external interface.
|
||||
from .hooks import AlbumInfo, TrackInfo, AlbumMatch, TrackMatch # noqa
|
||||
from .hooks import (
|
||||
AlbumInfo,
|
||||
TrackInfo,
|
||||
AlbumMatch,
|
||||
TrackMatch,
|
||||
Distance,
|
||||
) # noqa
|
||||
from .match import tag_item, tag_album, Proposal # noqa
|
||||
from .match import Recommendation # noqa
|
||||
|
||||
|
|
@ -32,6 +42,7 @@ log = logging.getLogger('beets')
|
|||
|
||||
# Additional utilities for the main interface.
|
||||
|
||||
|
||||
def apply_item_metadata(item, track_info):
|
||||
"""Set an item's metadata from its matched TrackInfo object.
|
||||
"""
|
||||
|
|
@ -72,14 +83,15 @@ def apply_metadata(album_info, mapping):
|
|||
for item, track_info in mapping.items():
|
||||
# Artist or artist credit.
|
||||
if config['artist_credit']:
|
||||
item.artist = (track_info.artist_credit or
|
||||
track_info.artist or
|
||||
album_info.artist_credit or
|
||||
album_info.artist)
|
||||
item.albumartist = (album_info.artist_credit or
|
||||
album_info.artist)
|
||||
item.artist = (
|
||||
track_info.artist_credit
|
||||
or track_info.artist
|
||||
or album_info.artist_credit
|
||||
or album_info.artist
|
||||
)
|
||||
item.albumartist = album_info.artist_credit or album_info.artist
|
||||
else:
|
||||
item.artist = (track_info.artist or album_info.artist)
|
||||
item.artist = track_info.artist or album_info.artist
|
||||
item.albumartist = album_info.artist
|
||||
|
||||
# Album.
|
||||
|
|
@ -87,8 +99,9 @@ def apply_metadata(album_info, mapping):
|
|||
|
||||
# Artist sort and credit names.
|
||||
item.artist_sort = track_info.artist_sort or album_info.artist_sort
|
||||
item.artist_credit = (track_info.artist_credit or
|
||||
album_info.artist_credit)
|
||||
item.artist_credit = (
|
||||
track_info.artist_credit or album_info.artist_credit
|
||||
)
|
||||
item.albumartist_sort = album_info.artist_sort
|
||||
item.albumartist_credit = album_info.artist_credit
|
||||
|
||||
|
|
@ -179,7 +192,7 @@ def apply_metadata(album_info, mapping):
|
|||
'work',
|
||||
'mb_workid',
|
||||
'work_disambig',
|
||||
)
|
||||
),
|
||||
}
|
||||
|
||||
# Don't overwrite fields with empty values unless the
|
||||
|
|
@ -197,3 +210,160 @@ def apply_metadata(album_info, mapping):
|
|||
if value is None and not clobber:
|
||||
continue
|
||||
item[field] = value
|
||||
|
||||
|
||||
def album_distance(config, data_source, album_info):
|
||||
"""Returns the ``data_source`` weight and the maximum source weight
|
||||
for albums.
|
||||
"""
|
||||
dist = Distance()
|
||||
if album_info.data_source == data_source:
|
||||
dist.add('source', config['source_weight'].as_number())
|
||||
return dist
|
||||
|
||||
|
||||
def track_distance(config, data_source, track_info):
|
||||
"""Returns the ``data_source`` weight and the maximum source weight
|
||||
for individual tracks.
|
||||
"""
|
||||
dist = Distance()
|
||||
if track_info.data_source == data_source:
|
||||
dist.add('source', config['source_weight'].as_number())
|
||||
return dist
|
||||
|
||||
|
||||
class APIAutotaggerPlugin(BeetsPlugin):
|
||||
def __init__(self):
|
||||
super(APIAutotaggerPlugin, self).__init__()
|
||||
self.config.add({'source_weight': 0.5})
|
||||
|
||||
@abstractproperty
|
||||
def id_regex(self):
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractproperty
|
||||
def data_source(self):
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractproperty
|
||||
def search_url(self):
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractproperty
|
||||
def album_url(self):
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractproperty
|
||||
def track_url(self):
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractmethod
|
||||
def _search_api(self, query_type, filters, keywords=''):
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractmethod
|
||||
def album_for_id(self, album_id):
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractmethod
|
||||
def track_for_id(self, track_id=None, track_data=None):
|
||||
raise NotImplementedError
|
||||
|
||||
@staticmethod
|
||||
def get_artist(artists, id_key='id', name_key='name'):
|
||||
"""Returns an artist string (all artists) and an artist_id (the main
|
||||
artist) for a list of artist object dicts.
|
||||
|
||||
:param artists: Iterable of artist dicts returned by API.
|
||||
:type artists: list[dict]
|
||||
:param id_key: Key corresponding to ``artist_id`` value.
|
||||
:type id_key: str
|
||||
:param name_key: Keys corresponding to values to concatenate for ``artist``.
|
||||
:type name_key: str
|
||||
:return: Normalized artist string.
|
||||
:rtype: str
|
||||
"""
|
||||
artist_id = None
|
||||
artist_names = []
|
||||
for artist in artists:
|
||||
if not artist_id:
|
||||
artist_id = artist[id_key]
|
||||
name = artist[name_key]
|
||||
# Move articles to the front.
|
||||
name = re.sub(r'^(.*?), (a|an|the)$', r'\2 \1', name, flags=re.I)
|
||||
artist_names.append(name)
|
||||
artist = ', '.join(artist_names).replace(' ,', ',') or None
|
||||
return artist, artist_id
|
||||
|
||||
def _get_id(self, url_type, id_):
|
||||
"""Parse an ID from its URL if necessary.
|
||||
|
||||
:param url_type: Type of URL. Either 'album' or 'track'.
|
||||
:type url_type: str
|
||||
:param id_: Album/track ID or URL.
|
||||
:type id_: str
|
||||
:return: Album/track ID.
|
||||
:rtype: str
|
||||
"""
|
||||
self._log.debug(
|
||||
u"Searching {} for {} '{}'", self.data_source, url_type, id_
|
||||
)
|
||||
match = re.search(
|
||||
self.id_regex['pattern'].format(url_type=url_type), str(id_)
|
||||
)
|
||||
id_ = match.group(self.id_regex['match_group'])
|
||||
return id_ if id_ else None
|
||||
|
||||
def candidates(self, items, artist, album, va_likely):
|
||||
"""Returns a list of AlbumInfo objects for Search API results
|
||||
matching an ``album`` and ``artist`` (if not various).
|
||||
|
||||
:param items: List of items comprised by an album to be matched.
|
||||
:type items: list[beets.library.Item]
|
||||
:param artist: The artist of the album to be matched.
|
||||
:type artist: str
|
||||
:param album: The name of the album to be matched.
|
||||
:type album: str
|
||||
:param va_likely: True if the album to be matched likely has
|
||||
Various Artists.
|
||||
:type va_likely: bool
|
||||
:return: Candidate AlbumInfo objects.
|
||||
:rtype: list[beets.autotag.hooks.AlbumInfo]
|
||||
"""
|
||||
query_filters = {'album': album}
|
||||
if not va_likely:
|
||||
query_filters['artist'] = artist
|
||||
albums = self._search_api(query_type='album', filters=query_filters)
|
||||
return [self.album_for_id(album_id=album['id']) for album in albums]
|
||||
|
||||
def item_candidates(self, item, artist, title):
|
||||
"""Returns a list of TrackInfo objects for Search API results
|
||||
matching ``title`` and ``artist``.
|
||||
|
||||
:param item: Singleton item to be matched.
|
||||
:type item: beets.library.Item
|
||||
:param artist: The artist of the track to be matched.
|
||||
:type artist: str
|
||||
:param title: The title of the track to be matched.
|
||||
:type title: str
|
||||
:return: Candidate TrackInfo objects.
|
||||
:rtype: list[beets.autotag.hooks.TrackInfo]
|
||||
"""
|
||||
tracks = self._search_api(
|
||||
query_type='track', keywords=title, filters={'artist': artist}
|
||||
)
|
||||
return [self.track_for_id(track_data=track) for track in tracks]
|
||||
|
||||
def album_distance(self, items, album_info, mapping):
|
||||
return album_distance(
|
||||
data_source=self.data_source,
|
||||
album_info=album_info,
|
||||
config=self.config,
|
||||
)
|
||||
|
||||
def track_distance(self, item, track_info):
|
||||
return track_distance(
|
||||
data_source=self.data_source,
|
||||
track_info=track_info,
|
||||
config=self.config,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -17,7 +17,6 @@
|
|||
"""
|
||||
from __future__ import absolute_import, print_function
|
||||
|
||||
import re
|
||||
import collections
|
||||
|
||||
import six
|
||||
|
|
@ -25,37 +24,24 @@ import unidecode
|
|||
import requests
|
||||
|
||||
from beets import ui
|
||||
from beets.plugins import BeetsPlugin
|
||||
from beets.autotag.hooks import AlbumInfo, TrackInfo, Distance
|
||||
from beets.autotag import APIAutotaggerPlugin
|
||||
from beets.autotag.hooks import AlbumInfo, TrackInfo
|
||||
|
||||
|
||||
class DeezerPlugin(BeetsPlugin):
|
||||
class DeezerPlugin(APIAutotaggerPlugin):
|
||||
# Base URLs for the Deezer API
|
||||
# Documentation: https://developers.deezer.com/api/
|
||||
search_url = 'https://api.deezer.com/search/'
|
||||
album_url = 'https://api.deezer.com/album/'
|
||||
track_url = 'https://api.deezer.com/track/'
|
||||
data_source = 'Deezer'
|
||||
id_regex = {
|
||||
'pattern': r'(^|deezer\.com/([a-z]*/)?{url_type}/)([0-9]*)',
|
||||
'match_group': 3,
|
||||
}
|
||||
|
||||
def __init__(self):
|
||||
super(DeezerPlugin, self).__init__()
|
||||
self.config.add({'source_weight': 0.5})
|
||||
|
||||
def _get_deezer_id(self, url_type, id_):
|
||||
"""Parse a Deezer ID from its URL if necessary.
|
||||
|
||||
:param url_type: Type of Deezer URL. Either 'album', 'artist',
|
||||
'playlist', or 'track'.
|
||||
:type url_type: str
|
||||
:param id_: Deezer ID or URL.
|
||||
:type id_: str
|
||||
:return: Deezer ID.
|
||||
:rtype: str
|
||||
"""
|
||||
id_regex = r'(^|deezer\.com/([a-z]*/)?{}/)([0-9]*)'
|
||||
self._log.debug(u'Searching for {} {}', url_type, id_)
|
||||
match = re.search(id_regex.format(url_type), str(id_))
|
||||
deezer_id = match.group(3)
|
||||
return deezer_id if deezer_id else None
|
||||
|
||||
def album_for_id(self, album_id):
|
||||
"""Fetch an album by its Deezer ID or URL and return an
|
||||
|
|
@ -66,12 +52,12 @@ class DeezerPlugin(BeetsPlugin):
|
|||
:return: AlbumInfo object for album.
|
||||
:rtype: beets.autotag.hooks.AlbumInfo or None
|
||||
"""
|
||||
deezer_id = self._get_deezer_id('album', album_id)
|
||||
deezer_id = self._get_id('album', album_id)
|
||||
if deezer_id is None:
|
||||
return None
|
||||
|
||||
album_data = requests.get(self.album_url + deezer_id).json()
|
||||
artist, artist_id = self._get_artist(album_data['contributors'])
|
||||
artist, artist_id = self.get_artist(album_data['contributors'])
|
||||
|
||||
release_date = album_data['release_date']
|
||||
date_parts = [int(part) for part in release_date.split('-')]
|
||||
|
|
@ -89,7 +75,7 @@ class DeezerPlugin(BeetsPlugin):
|
|||
else:
|
||||
raise ui.UserError(
|
||||
u"Invalid `release_date` returned "
|
||||
u"by Deezer API: '{}'".format(release_date)
|
||||
u"by {} API: '{}'".format(self.data_source, release_date)
|
||||
)
|
||||
|
||||
tracks_data = requests.get(
|
||||
|
|
@ -109,7 +95,7 @@ class DeezerPlugin(BeetsPlugin):
|
|||
album=album_data['title'],
|
||||
album_id=deezer_id,
|
||||
artist=artist,
|
||||
artist_credit=self._get_artist([album_data['artist']])[0],
|
||||
artist_credit=self.get_artist([album_data['artist']])[0],
|
||||
artist_id=artist_id,
|
||||
tracks=tracks,
|
||||
albumtype=album_data['record_type'],
|
||||
|
|
@ -120,7 +106,7 @@ class DeezerPlugin(BeetsPlugin):
|
|||
day=day,
|
||||
label=album_data['label'],
|
||||
mediums=max(medium_totals.keys()),
|
||||
data_source='Deezer',
|
||||
data_source=self.data_source,
|
||||
data_url=album_data['link'],
|
||||
)
|
||||
|
||||
|
|
@ -132,7 +118,7 @@ class DeezerPlugin(BeetsPlugin):
|
|||
:return: TrackInfo object for track
|
||||
:rtype: beets.autotag.hooks.TrackInfo
|
||||
"""
|
||||
artist, artist_id = self._get_artist(
|
||||
artist, artist_id = self.get_artist(
|
||||
track_data.get('contributors', [track_data['artist']])
|
||||
)
|
||||
return TrackInfo(
|
||||
|
|
@ -144,7 +130,7 @@ class DeezerPlugin(BeetsPlugin):
|
|||
index=track_data['track_position'],
|
||||
medium=track_data['disk_number'],
|
||||
medium_index=track_data['track_position'],
|
||||
data_source='Deezer',
|
||||
data_source=self.data_source,
|
||||
data_url=track_data['link'],
|
||||
)
|
||||
|
||||
|
|
@ -162,7 +148,7 @@ class DeezerPlugin(BeetsPlugin):
|
|||
:rtype: beets.autotag.hooks.TrackInfo or None
|
||||
"""
|
||||
if track_data is None:
|
||||
deezer_id = self._get_deezer_id('track', track_id)
|
||||
deezer_id = self._get_id('track', track_id)
|
||||
if deezer_id is None:
|
||||
return None
|
||||
track_data = requests.get(self.track_url + deezer_id).json()
|
||||
|
|
@ -176,107 +162,13 @@ class DeezerPlugin(BeetsPlugin):
|
|||
).json()['data']
|
||||
medium_total = 0
|
||||
for i, track_data in enumerate(album_tracks_data, start=1):
|
||||
if track_data['disc_number'] == track.medium:
|
||||
if track_data['disk_number'] == track.medium:
|
||||
medium_total += 1
|
||||
if track_data['id'] == track.track_id:
|
||||
track.index = i
|
||||
track.medium_total = medium_total
|
||||
return track
|
||||
|
||||
@staticmethod
|
||||
def _get_artist(artists):
|
||||
"""Returns an artist string (all artists) and an artist_id (the main
|
||||
artist) for a list of Deezer artist object dicts.
|
||||
|
||||
:param artists: Iterable of ``contributors`` or ``artist`` returned
|
||||
by the Deezer Album (https://developers.deezer.com/api/album) or
|
||||
Deezer Track (https://developers.deezer.com/api/track) APIs.
|
||||
:type artists: list[dict]
|
||||
:return: Normalized artist string
|
||||
:rtype: str
|
||||
"""
|
||||
artist_id = None
|
||||
artist_names = []
|
||||
for artist in artists:
|
||||
if not artist_id:
|
||||
artist_id = artist['id']
|
||||
name = artist['name']
|
||||
# Move articles to the front.
|
||||
name = re.sub(r'^(.*?), (a|an|the)$', r'\2 \1', name, flags=re.I)
|
||||
artist_names.append(name)
|
||||
artist = ', '.join(artist_names).replace(' ,', ',') or None
|
||||
return artist, artist_id
|
||||
|
||||
def album_distance(self, items, album_info, mapping):
|
||||
"""Returns the Deezer source weight and the maximum source weight
|
||||
for albums.
|
||||
"""
|
||||
dist = Distance()
|
||||
if album_info.data_source == 'Deezer':
|
||||
dist.add('source', self.config['source_weight'].as_number())
|
||||
return dist
|
||||
|
||||
def track_distance(self, item, track_info):
|
||||
"""Returns the Deezer source weight and the maximum source weight
|
||||
for individual tracks.
|
||||
"""
|
||||
dist = Distance()
|
||||
if track_info.data_source == 'Deezer':
|
||||
dist.add('source', self.config['source_weight'].as_number())
|
||||
return dist
|
||||
|
||||
def candidates(self, items, artist, album, va_likely):
|
||||
"""Returns a list of AlbumInfo objects for Deezer Search API results
|
||||
matching an ``album`` and ``artist`` (if not various).
|
||||
|
||||
:param items: List of items comprised by an album to be matched.
|
||||
:type items: list[beets.library.Item]
|
||||
:param artist: The artist of the album to be matched.
|
||||
:type artist: str
|
||||
:param album: The name of the album to be matched.
|
||||
:type album: str
|
||||
:param va_likely: True if the album to be matched likely has
|
||||
Various Artists.
|
||||
:type va_likely: bool
|
||||
:return: Candidate AlbumInfo objects.
|
||||
:rtype: list[beets.autotag.hooks.AlbumInfo]
|
||||
"""
|
||||
query_filters = {'album': album}
|
||||
if not va_likely:
|
||||
query_filters['artist'] = artist
|
||||
response_data = self._search_deezer(
|
||||
query_type='album', filters=query_filters
|
||||
)
|
||||
if response_data is None:
|
||||
return []
|
||||
return [
|
||||
self.album_for_id(album_id=album_data['id'])
|
||||
for album_data in response_data['data']
|
||||
]
|
||||
|
||||
def item_candidates(self, item, artist, title):
|
||||
"""Returns a list of TrackInfo objects for Deezer Search API results
|
||||
matching ``title`` and ``artist``.
|
||||
|
||||
:param item: Singleton item to be matched.
|
||||
:type item: beets.library.Item
|
||||
:param artist: The artist of the track to be matched.
|
||||
:type artist: str
|
||||
:param title: The title of the track to be matched.
|
||||
:type title: str
|
||||
:return: Candidate TrackInfo objects.
|
||||
:rtype: list[beets.autotag.hooks.TrackInfo]
|
||||
"""
|
||||
response_data = self._search_deezer(
|
||||
query_type='track', keywords=title, filters={'artist': artist}
|
||||
)
|
||||
if response_data is None:
|
||||
return []
|
||||
return [
|
||||
self.track_for_id(track_data=track_data)
|
||||
for track_data in response_data['data']
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def _construct_search_query(filters=None, keywords=''):
|
||||
"""Construct a query string with the specified filters and keywords to
|
||||
|
|
@ -299,7 +191,7 @@ class DeezerPlugin(BeetsPlugin):
|
|||
query = query.decode('utf8')
|
||||
return unidecode.unidecode(query)
|
||||
|
||||
def _search_deezer(self, query_type, filters=None, keywords=''):
|
||||
def _search_api(self, query_type, filters=None, keywords=''):
|
||||
"""Query the Deezer Search API for the specified ``keywords``, applying
|
||||
the provided ``filters``.
|
||||
|
||||
|
|
@ -320,12 +212,18 @@ class DeezerPlugin(BeetsPlugin):
|
|||
)
|
||||
if not query:
|
||||
return None
|
||||
self._log.debug(u"Searching Deezer for '{}'".format(query))
|
||||
response_data = requests.get(
|
||||
self.search_url + query_type, params={'q': query}
|
||||
).json()
|
||||
num_results = len(response_data['data'])
|
||||
self._log.debug(
|
||||
u"Found {} results from Deezer for '{}'", num_results, query
|
||||
u"Searching {} for '{}'".format(self.data_source, query)
|
||||
)
|
||||
return response_data if num_results > 0 else None
|
||||
response = requests.get(
|
||||
self.search_url + query_type, params={'q': query}
|
||||
)
|
||||
response.raise_for_status()
|
||||
response_data = response.json().get('data', [])
|
||||
self._log.debug(
|
||||
u"Found {} results from {} for '{}'",
|
||||
self.data_source,
|
||||
len(response_data),
|
||||
query,
|
||||
)
|
||||
return response_data
|
||||
|
|
|
|||
|
|
@ -20,7 +20,8 @@ from __future__ import division, absolute_import, print_function
|
|||
|
||||
import beets.ui
|
||||
from beets import config
|
||||
from beets.autotag.hooks import AlbumInfo, TrackInfo, Distance
|
||||
from beets.autotag import APIAutotaggerPlugin, album_distance
|
||||
from beets.autotag.hooks import AlbumInfo, TrackInfo
|
||||
from beets.plugins import BeetsPlugin
|
||||
import confuse
|
||||
from discogs_client import Release, Master, Client
|
||||
|
|
@ -159,10 +160,11 @@ class DiscogsPlugin(BeetsPlugin):
|
|||
def album_distance(self, items, album_info, mapping):
|
||||
"""Returns the album distance.
|
||||
"""
|
||||
dist = Distance()
|
||||
if album_info.data_source == 'Discogs':
|
||||
dist.add('source', self.config['source_weight'].as_number())
|
||||
return dist
|
||||
return album_distance(
|
||||
data_source='Discogs',
|
||||
album_info=album_info,
|
||||
config=self.config
|
||||
)
|
||||
|
||||
def candidates(self, items, artist, album, va_likely):
|
||||
"""Returns a list of AlbumInfo objects for discogs search results
|
||||
|
|
@ -292,7 +294,7 @@ class DiscogsPlugin(BeetsPlugin):
|
|||
self._log.warning(u"Release does not contain the required fields")
|
||||
return None
|
||||
|
||||
artist, artist_id = self.get_artist([a.data for a in result.artists])
|
||||
artist, artist_id = APIAutotaggerPlugin.get_artist([a.data for a in result.artists])
|
||||
album = re.sub(r' +', ' ', result.title)
|
||||
album_id = result.data['id']
|
||||
# Use `.data` to access the tracklist directly instead of the
|
||||
|
|
@ -368,26 +370,6 @@ class DiscogsPlugin(BeetsPlugin):
|
|||
else:
|
||||
return None
|
||||
|
||||
def get_artist(self, artists):
|
||||
"""Returns an artist string (all artists) and an artist_id (the main
|
||||
artist) for a list of discogs album or track artists.
|
||||
"""
|
||||
artist_id = None
|
||||
bits = []
|
||||
for i, artist in enumerate(artists):
|
||||
if not artist_id:
|
||||
artist_id = artist['id']
|
||||
name = artist['name']
|
||||
# Strip disambiguation number.
|
||||
name = re.sub(r' \(\d+\)$', '', name)
|
||||
# Move articles to the front.
|
||||
name = re.sub(r'(?i)^(.*?), (a|an|the)$', r'\2 \1', name)
|
||||
bits.append(name)
|
||||
if artist['join'] and i < len(artists) - 1:
|
||||
bits.append(artist['join'])
|
||||
artist = ' '.join(bits).replace(' ,', ',') or None
|
||||
return artist, artist_id
|
||||
|
||||
def get_tracks(self, tracklist):
|
||||
"""Returns a list of TrackInfo objects for a discogs tracklist.
|
||||
"""
|
||||
|
|
@ -551,7 +533,7 @@ class DiscogsPlugin(BeetsPlugin):
|
|||
title = track['title']
|
||||
track_id = None
|
||||
medium, medium_index, _ = self.get_track_index(track['position'])
|
||||
artist, artist_id = self.get_artist(track.get('artists', []))
|
||||
artist, artist_id = APIAutotaggerPlugin.get_artist(track.get('artists', []))
|
||||
length = self.get_track_length(track['duration'])
|
||||
return TrackInfo(title, track_id, artist=artist, artist_id=artist_id,
|
||||
length=length, index=index,
|
||||
|
|
|
|||
|
|
@ -27,14 +27,14 @@ import collections
|
|||
import six
|
||||
import unidecode
|
||||
import requests
|
||||
import confuse
|
||||
|
||||
from beets import ui
|
||||
from beets.plugins import BeetsPlugin
|
||||
import confuse
|
||||
from beets.autotag.hooks import AlbumInfo, TrackInfo, Distance
|
||||
from beets.autotag import APIAutotaggerPlugin
|
||||
from beets.autotag.hooks import AlbumInfo, TrackInfo
|
||||
|
||||
|
||||
class SpotifyPlugin(BeetsPlugin):
|
||||
class SpotifyPlugin(APIAutotaggerPlugin):
|
||||
# Base URLs for the Spotify API
|
||||
# Documentation: https://developer.spotify.com/web-api
|
||||
oauth_token_url = 'https://accounts.spotify.com/api/token'
|
||||
|
|
@ -43,6 +43,14 @@ class SpotifyPlugin(BeetsPlugin):
|
|||
album_url = 'https://api.spotify.com/v1/albums/'
|
||||
track_url = 'https://api.spotify.com/v1/tracks/'
|
||||
playlist_partial = 'spotify:trackset:Playlist:'
|
||||
data_source = 'Spotify'
|
||||
|
||||
# Spotify IDs consist of 22 alphanumeric characters
|
||||
# (zero-left-padded base62 representation of randomly generated UUID4)
|
||||
id_regex = {
|
||||
'pattern': r'(^|open\.spotify\.com/{url_type}/)([0-9A-Za-z]{{22}})',
|
||||
'match_group': 2,
|
||||
}
|
||||
|
||||
def __init__(self):
|
||||
super(SpotifyPlugin, self).__init__()
|
||||
|
|
@ -59,7 +67,6 @@ class SpotifyPlugin(BeetsPlugin):
|
|||
'client_id': '4e414367a1d14c75a5c5129a627fcab8',
|
||||
'client_secret': 'f82bdc09b2254f1a8286815d02fd46dc',
|
||||
'tokenfile': 'spotify_token.json',
|
||||
'source_weight': 0.5,
|
||||
}
|
||||
)
|
||||
self.config['client_secret'].redact = True
|
||||
|
|
@ -140,26 +147,11 @@ class SpotifyPlugin(BeetsPlugin):
|
|||
self._authenticate()
|
||||
return self._handle_response(request_type, url, params=params)
|
||||
else:
|
||||
raise ui.UserError(u'Spotify API error:\n{}', response.text)
|
||||
raise ui.UserError(
|
||||
u'{} API error:\n{}', self.data_source, response.text
|
||||
)
|
||||
return response.json()
|
||||
|
||||
def _get_spotify_id(self, url_type, id_):
|
||||
"""Parse a Spotify ID from its URL if necessary.
|
||||
|
||||
:param url_type: Type of Spotify URL, either 'album' or 'track'.
|
||||
:type url_type: str
|
||||
:param id_: Spotify ID or URL.
|
||||
:type id_: str
|
||||
:return: Spotify ID.
|
||||
:rtype: str
|
||||
"""
|
||||
# Spotify IDs consist of 22 alphanumeric characters
|
||||
# (zero-left-padded base62 representation of randomly generated UUID4)
|
||||
id_regex = r'(^|open\.spotify\.com/{}/)([0-9A-Za-z]{{22}})'
|
||||
self._log.debug(u'Searching for {} {}', url_type, id_)
|
||||
match = re.search(id_regex.format(url_type), id_)
|
||||
return match.group(2) if match else None
|
||||
|
||||
def album_for_id(self, album_id):
|
||||
"""Fetch an album by its Spotify ID or URL and return an
|
||||
AlbumInfo object or None if the album is not found.
|
||||
|
|
@ -169,20 +161,20 @@ class SpotifyPlugin(BeetsPlugin):
|
|||
:return: AlbumInfo object for album
|
||||
:rtype: beets.autotag.hooks.AlbumInfo or None
|
||||
"""
|
||||
spotify_id = self._get_spotify_id('album', album_id)
|
||||
spotify_id = self._get_id('album', album_id)
|
||||
if spotify_id is None:
|
||||
return None
|
||||
|
||||
response_data = self._handle_response(
|
||||
album_data = self._handle_response(
|
||||
requests.get, self.album_url + spotify_id
|
||||
)
|
||||
artist, artist_id = self._get_artist(response_data['artists'])
|
||||
artist, artist_id = self.get_artist(album_data['artists'])
|
||||
|
||||
date_parts = [
|
||||
int(part) for part in response_data['release_date'].split('-')
|
||||
int(part) for part in album_data['release_date'].split('-')
|
||||
]
|
||||
|
||||
release_date_precision = response_data['release_date_precision']
|
||||
release_date_precision = album_data['release_date_precision']
|
||||
if release_date_precision == 'day':
|
||||
year, month, day = date_parts
|
||||
elif release_date_precision == 'month':
|
||||
|
|
@ -195,14 +187,14 @@ class SpotifyPlugin(BeetsPlugin):
|
|||
else:
|
||||
raise ui.UserError(
|
||||
u"Invalid `release_date_precision` returned "
|
||||
u"by Spotify API: '{}'".format(release_date_precision)
|
||||
u"by {} API: '{}'".format(
|
||||
self.data_source, release_date_precision
|
||||
)
|
||||
)
|
||||
|
||||
tracks = []
|
||||
medium_totals = collections.defaultdict(int)
|
||||
for i, track_data in enumerate(
|
||||
response_data['tracks']['items'], start=1
|
||||
):
|
||||
for i, track_data in enumerate(album_data['tracks']['items'], start=1):
|
||||
track = self._get_track(track_data)
|
||||
track.index = i
|
||||
medium_totals[track.medium] += 1
|
||||
|
|
@ -211,21 +203,21 @@ class SpotifyPlugin(BeetsPlugin):
|
|||
track.medium_total = medium_totals[track.medium]
|
||||
|
||||
return AlbumInfo(
|
||||
album=response_data['name'],
|
||||
album=album_data['name'],
|
||||
album_id=spotify_id,
|
||||
artist=artist,
|
||||
artist_id=artist_id,
|
||||
tracks=tracks,
|
||||
albumtype=response_data['album_type'],
|
||||
va=len(response_data['artists']) == 1
|
||||
albumtype=album_data['album_type'],
|
||||
va=len(album_data['artists']) == 1
|
||||
and artist.lower() == 'various artists',
|
||||
year=year,
|
||||
month=month,
|
||||
day=day,
|
||||
label=response_data['label'],
|
||||
label=album_data['label'],
|
||||
mediums=max(medium_totals.keys()),
|
||||
data_source='Spotify',
|
||||
data_url=response_data['external_urls']['spotify'],
|
||||
data_source=self.data_source,
|
||||
data_url=album_data['external_urls']['spotify'],
|
||||
)
|
||||
|
||||
def _get_track(self, track_data):
|
||||
|
|
@ -237,7 +229,7 @@ class SpotifyPlugin(BeetsPlugin):
|
|||
:return: TrackInfo object for track
|
||||
:rtype: beets.autotag.hooks.TrackInfo
|
||||
"""
|
||||
artist, artist_id = self._get_artist(track_data['artists'])
|
||||
artist, artist_id = self.get_artist(track_data['artists'])
|
||||
return TrackInfo(
|
||||
title=track_data['name'],
|
||||
track_id=track_data['id'],
|
||||
|
|
@ -247,7 +239,7 @@ class SpotifyPlugin(BeetsPlugin):
|
|||
index=track_data['track_number'],
|
||||
medium=track_data['disc_number'],
|
||||
medium_index=track_data['track_number'],
|
||||
data_source='Spotify',
|
||||
data_source=self.data_source,
|
||||
data_url=track_data['external_urls']['spotify'],
|
||||
)
|
||||
|
||||
|
|
@ -265,7 +257,7 @@ class SpotifyPlugin(BeetsPlugin):
|
|||
:rtype: beets.autotag.hooks.TrackInfo or None
|
||||
"""
|
||||
if track_data is None:
|
||||
spotify_id = self._get_spotify_id('track', track_id)
|
||||
spotify_id = self._get_id('track', track_id)
|
||||
if spotify_id is None:
|
||||
return None
|
||||
track_data = self._handle_response(
|
||||
|
|
@ -288,99 +280,6 @@ class SpotifyPlugin(BeetsPlugin):
|
|||
track.medium_total = medium_total
|
||||
return track
|
||||
|
||||
@staticmethod
|
||||
def _get_artist(artists):
|
||||
"""Returns an artist string (all artists) and an artist_id (the main
|
||||
artist) for a list of Spotify artist object dicts.
|
||||
|
||||
:param artists: Iterable of simplified Spotify artist objects
|
||||
(https://developer.spotify.com/documentation/web-api/reference/object-model/#artist-object-simplified)
|
||||
:type artists: list[dict]
|
||||
:return: Normalized artist string
|
||||
:rtype: str
|
||||
"""
|
||||
artist_id = None
|
||||
artist_names = []
|
||||
for artist in artists:
|
||||
if not artist_id:
|
||||
artist_id = artist['id']
|
||||
name = artist['name']
|
||||
# Move articles to the front.
|
||||
name = re.sub(r'^(.*?), (a|an|the)$', r'\2 \1', name, flags=re.I)
|
||||
artist_names.append(name)
|
||||
artist = ', '.join(artist_names).replace(' ,', ',') or None
|
||||
return artist, artist_id
|
||||
|
||||
def album_distance(self, items, album_info, mapping):
|
||||
"""Returns the Spotify source weight and the maximum source weight
|
||||
for albums.
|
||||
"""
|
||||
dist = Distance()
|
||||
if album_info.data_source == 'Spotify':
|
||||
dist.add('source', self.config['source_weight'].as_number())
|
||||
return dist
|
||||
|
||||
def track_distance(self, item, track_info):
|
||||
"""Returns the Spotify source weight and the maximum source weight
|
||||
for individual tracks.
|
||||
"""
|
||||
dist = Distance()
|
||||
if track_info.data_source == 'Spotify':
|
||||
dist.add('source', self.config['source_weight'].as_number())
|
||||
return dist
|
||||
|
||||
def candidates(self, items, artist, album, va_likely):
|
||||
"""Returns a list of AlbumInfo objects for Spotify Search API results
|
||||
matching an ``album`` and ``artist`` (if not various).
|
||||
|
||||
:param items: List of items comprised by an album to be matched.
|
||||
:type items: list[beets.library.Item]
|
||||
:param artist: The artist of the album to be matched.
|
||||
:type artist: str
|
||||
:param album: The name of the album to be matched.
|
||||
:type album: str
|
||||
:param va_likely: True if the album to be matched likely has
|
||||
Various Artists.
|
||||
:type va_likely: bool
|
||||
:return: Candidate AlbumInfo objects.
|
||||
:rtype: list[beets.autotag.hooks.AlbumInfo]
|
||||
"""
|
||||
query_filters = {'album': album}
|
||||
if not va_likely:
|
||||
query_filters['artist'] = artist
|
||||
response_data = self._search_spotify(
|
||||
query_type='album', filters=query_filters
|
||||
)
|
||||
if response_data is None:
|
||||
return []
|
||||
return [
|
||||
self.album_for_id(album_id=album_data['id'])
|
||||
for album_data in response_data['albums']['items']
|
||||
]
|
||||
|
||||
def item_candidates(self, item, artist, title):
|
||||
"""Returns a list of TrackInfo objects for Spotify Search API results
|
||||
matching ``title`` and ``artist``.
|
||||
|
||||
:param item: Singleton item to be matched.
|
||||
:type item: beets.library.Item
|
||||
:param artist: The artist of the track to be matched.
|
||||
:type artist: str
|
||||
:param title: The title of the track to be matched.
|
||||
:type title: str
|
||||
:return: Candidate TrackInfo objects.
|
||||
:rtype: list[beets.autotag.hooks.TrackInfo]
|
||||
"""
|
||||
response_data = self._search_spotify(
|
||||
query_type='track', keywords=title, filters={'artist': artist}
|
||||
)
|
||||
if response_data is None:
|
||||
return []
|
||||
return [
|
||||
self.track_for_id(track_data=track_data)
|
||||
for track_data in response_data['tracks']['items']
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def _construct_search_query(filters=None, keywords=''):
|
||||
"""Construct a query string with the specified filters and keywords to
|
||||
|
|
@ -403,14 +302,12 @@ class SpotifyPlugin(BeetsPlugin):
|
|||
query = query.decode('utf8')
|
||||
return unidecode.unidecode(query)
|
||||
|
||||
def _search_spotify(self, query_type, filters=None, keywords=''):
|
||||
def _search_api(self, query_type, filters=None, keywords=''):
|
||||
"""Query the Spotify Search API for the specified ``keywords``, applying
|
||||
the provided ``filters``.
|
||||
|
||||
:param query_type: A comma-separated list of item types to search
|
||||
across. Valid types are: 'album', 'artist', 'playlist', and
|
||||
'track'. Search results include hits from all the specified item
|
||||
types.
|
||||
:param query_type: Item type to search across. Valid types are: 'album',
|
||||
'artist', 'playlist', and 'track'.
|
||||
:type query_type: str
|
||||
:param filters: (Optional) Field filters to apply.
|
||||
:type filters: dict
|
||||
|
|
@ -425,19 +322,25 @@ class SpotifyPlugin(BeetsPlugin):
|
|||
)
|
||||
if not query:
|
||||
return None
|
||||
self._log.debug(u"Searching Spotify for '{}'".format(query))
|
||||
response_data = self._handle_response(
|
||||
requests.get,
|
||||
self.search_url,
|
||||
params={'q': query, 'type': query_type},
|
||||
)
|
||||
num_results = 0
|
||||
for result_type_data in response_data.values():
|
||||
num_results += len(result_type_data['items'])
|
||||
self._log.debug(
|
||||
u"Found {} results from Spotify for '{}'", num_results, query
|
||||
u"Searching {} for '{}'".format(self.data_source, query)
|
||||
)
|
||||
return response_data if num_results > 0 else None
|
||||
response_data = (
|
||||
self._handle_response(
|
||||
requests.get,
|
||||
self.search_url,
|
||||
params={'q': query, 'type': query_type},
|
||||
)
|
||||
.get(query_type + 's', {})
|
||||
.get('items', [])
|
||||
)
|
||||
self._log.debug(
|
||||
u"Found {} results from {} for '{}'",
|
||||
self.data_source,
|
||||
len(response_data),
|
||||
query,
|
||||
)
|
||||
return response_data
|
||||
|
||||
def commands(self):
|
||||
def queries(lib, opts, args):
|
||||
|
|
@ -529,7 +432,7 @@ class SpotifyPlugin(BeetsPlugin):
|
|||
|
||||
# Query the Web API for each track, look for the items' JSON data
|
||||
query_filters = {'artist': artist, 'album': album}
|
||||
response_data = self._search_spotify(
|
||||
response_data = self._search_api(
|
||||
query_type='track', keywords=keywords, filters=query_filters
|
||||
)
|
||||
if response_data is None:
|
||||
|
|
|
|||
Loading…
Reference in a new issue