1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
|
from __future__ import print_function
import re
import webbrowser
import requests
import logging
from beets.plugins import BeetsPlugin
from beets.ui import decargs
from beets import ui
from requests.exceptions import HTTPError
log = logging.getLogger('beets')
class SpotifyPlugin(BeetsPlugin):
# URL for the Web API of Spotify
# Documentation here: https://developer.spotify.com/web-api/search-item/
base_url = "https://api.spotify.com/v1/search"
open_url = "http://open.spotify.com/track/"
playlist_partial = "spotify:trackset:Playlist:"
def __init__(self):
super(SpotifyPlugin, self).__init__()
self.config.add({
'mode': 'list',
'tiebreak': 'popularity',
'show_failures': False,
'artist_field': 'albumartist',
'album_field': 'album',
'track_field': 'title',
'region_filter': None,
'regex': []
})
def commands(self):
def queries(lib, opts, args):
success = self.parse_opts(opts)
if success:
results = self.query_spotify(lib, decargs(args))
self.output_results(results)
spotify_cmd = ui.Subcommand(
'spotify',
help='build a Spotify playlist'
)
spotify_cmd.parser.add_option(
'-m', '--mode', action='store',
help='"open" to open Spotify with playlist, '
'"list" to print (default)'
)
spotify_cmd.parser.add_option(
'-f', '--show-failures', action='store_true',
help='list tracks that did not match a Sptoify ID',
dest='show_failures',
)
spotify_cmd.func = queries
return [spotify_cmd]
def parse_opts(self, opts):
if opts.mode:
self.config['mode'].set(opts.mode)
if opts.show_failures:
self.config['show_failures'].set(True)
if self.config['mode'].get() not in ['list', 'open']:
log.warn(u'{0} is not a valid mode'
.format(self.config['mode'].get()))
return False
self.opts = opts
return True
def query_spotify(self, lib, query):
results = []
failures = []
items = lib.items(query)
if not items:
log.debug(u'Your beets query returned no items, skipping spotify')
return
log.info(u'Processing {0} tracks...'.format(len(items)))
for item in items:
# Apply regex transformations if provided
for regex in self.config['regex'].get():
if (
not regex['field'] or
not regex['search'] or
not regex['replace']
):
continue
value = item[regex['field']]
item[regex['field']] = re.sub(
regex['search'], regex['replace'], value
)
# Custom values can be passed in the config (just in case)
artist = item[self.config['artist_field'].get()]
album = item[self.config['album_field'].get()]
query = item[self.config['track_field'].get()]
search_url = query + " album:" + album + " artist:" + artist
# Query the Web API for each track, look for the items' JSON data
r = requests.get(self.base_url, params={
"q": search_url, "type": "track"
})
log.debug(r.url)
try:
r.raise_for_status()
except HTTPError as e:
log.debug(u'URL returned a {0} error'
.format(e.response.status_code))
failures.append(search_url)
continue
r_data = r.json()['tracks']['items']
# Apply market filter if requested
region_filter = self.config['region_filter'].get()
if region_filter:
r_data = filter(
lambda x: region_filter in x['available_markets'], r_data
)
# Simplest, take the first result
chosen_result = None
if len(r_data) == 1 or self.config['tiebreak'].get() == "first":
log.debug(u'Spotify track(s) found, count: {0}'
.format(len(r_data)))
chosen_result = r_data[0]
elif len(r_data) > 1:
# Use the popularity filter
log.debug(u'Most popular track chosen, count: {0}'
.format(len(r_data)))
chosen_result = max(r_data, key=lambda x: x['popularity'])
if chosen_result:
results.append(chosen_result)
else:
log.debug(u'No spotify track found: {0}'.format(search_url))
failures.append(search_url)
failure_count = len(failures)
if failure_count > 0:
if self.config['show_failures'].get():
log.info(u'{0} track(s) did not match a Spotify ID:'
.format(failure_count))
for track in failures:
log.info(u'track: {0}'.format(track))
log.info(u'')
else:
log.warn(u'{0} track(s) did not match a Spotify ID;\n'
u'use --show-failures to display'
.format(failure_count))
return results
def output_results(self, results):
if results:
ids = map(lambda x: x['id'], results)
if self.config['mode'].get() == "open":
log.info(u'Attempting to open Spotify with playlist')
spotify_url = self.playlist_partial + ",".join(ids)
webbrowser.open(spotify_url)
else:
for item in ids:
print(unicode.encode(self.open_url + item))
else:
log.warn(u'No Spotify tracks found from beets query')
|
