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
|
# This file is part of beets.
# Copyright (c) 2011, Jeffrey Aylesworth <mail@jeffrey.red>
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
import re
import musicbrainzngs
from beets import config, ui
from beets.plugins import BeetsPlugin
from beets.ui import Subcommand
SUBMISSION_CHUNK_SIZE = 200
FETCH_CHUNK_SIZE = 100
UUID_REGEX = r"^[a-f0-9]{8}(-[a-f0-9]{4}){3}-[a-f0-9]{12}$"
def mb_call(func, *args, **kwargs):
"""Call a MusicBrainz API function and catch exceptions."""
try:
return func(*args, **kwargs)
except musicbrainzngs.AuthenticationError:
raise ui.UserError("authentication with MusicBrainz failed")
except (musicbrainzngs.ResponseError, musicbrainzngs.NetworkError) as exc:
raise ui.UserError(f"MusicBrainz API error: {exc}")
except musicbrainzngs.UsageError:
raise ui.UserError("MusicBrainz credentials missing")
def submit_albums(collection_id, release_ids):
"""Add all of the release IDs to the indicated collection. Multiple
requests are made if there are many release IDs to submit.
"""
for i in range(0, len(release_ids), SUBMISSION_CHUNK_SIZE):
chunk = release_ids[i : i + SUBMISSION_CHUNK_SIZE]
mb_call(musicbrainzngs.add_releases_to_collection, collection_id, chunk)
class MusicBrainzCollectionPlugin(BeetsPlugin):
def __init__(self):
super().__init__()
config["musicbrainz"]["pass"].redact = True
musicbrainzngs.auth(
config["musicbrainz"]["user"].as_str(),
config["musicbrainz"]["pass"].as_str(),
)
self.config.add(
{
"auto": False,
"collection": "",
"remove": False,
}
)
if self.config["auto"]:
self.import_stages = [self.imported]
def _get_collection(self):
collections = mb_call(musicbrainzngs.get_collections)
if not collections["collection-list"]:
raise ui.UserError("no collections exist for user")
# Get all collection IDs, avoiding event collections
collection_ids = [x["id"] for x in collections["collection-list"]]
if not collection_ids:
raise ui.UserError("No collection found.")
# Check that the collection exists so we can present a nice error
collection = self.config["collection"].as_str()
if collection:
if collection not in collection_ids:
raise ui.UserError(
"invalid collection ID: {}".format(collection)
)
return collection
# No specified collection. Just return the first collection ID
return collection_ids[0]
def _get_albums_in_collection(self, id):
def _fetch(offset):
res = mb_call(
musicbrainzngs.get_releases_in_collection,
id,
limit=FETCH_CHUNK_SIZE,
offset=offset,
)["collection"]
return [x["id"] for x in res["release-list"]], res["release-count"]
offset = 0
albums_in_collection, release_count = _fetch(offset)
for i in range(0, release_count, FETCH_CHUNK_SIZE):
albums_in_collection += _fetch(offset)[0]
offset += FETCH_CHUNK_SIZE
return albums_in_collection
def commands(self):
mbupdate = Subcommand("mbupdate", help="Update MusicBrainz collection")
mbupdate.parser.add_option(
"-r",
"--remove",
action="store_true",
default=None,
dest="remove",
help="Remove albums not in beets library",
)
mbupdate.func = self.update_collection
return [mbupdate]
def remove_missing(self, collection_id, lib_albums):
lib_ids = {x.mb_albumid for x in lib_albums}
albums_in_collection = self._get_albums_in_collection(collection_id)
remove_me = list(set(albums_in_collection) - lib_ids)
for i in range(0, len(remove_me), FETCH_CHUNK_SIZE):
chunk = remove_me[i : i + FETCH_CHUNK_SIZE]
mb_call(
musicbrainzngs.remove_releases_from_collection,
collection_id,
chunk,
)
def update_collection(self, lib, opts, args):
self.config.set_args(opts)
remove_missing = self.config["remove"].get(bool)
self.update_album_list(lib, lib.albums(), remove_missing)
def imported(self, session, task):
"""Add each imported album to the collection."""
if task.is_album:
self.update_album_list(session.lib, [task.album])
def update_album_list(self, lib, album_list, remove_missing=False):
"""Update the MusicBrainz collection from a list of Beets albums"""
collection_id = self._get_collection()
# Get a list of all the album IDs.
album_ids = []
for album in album_list:
aid = album.mb_albumid
if aid:
if re.match(UUID_REGEX, aid):
album_ids.append(aid)
else:
self._log.info("skipping invalid MBID: {0}", aid)
# Submit to MusicBrainz.
self._log.info("Updating MusicBrainz collection {0}...", collection_id)
submit_albums(collection_id, album_ids)
if remove_missing:
self.remove_missing(collection_id, lib.albums())
self._log.info("...MusicBrainz collection updated.")
|
