Adding MusicBrainz OAuth2 support

beetsplug/mbcollection.py

@@ -23,10 +23,12 @@
 
 import re
 
+from rauth import OAuth2Service
+import json
+
 SUBMISSION_CHUNK_SIZE = 200
 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.
     """
@@ -53,16 +55,27 @@ def submit_albums(collection_id, release_ids):
 
 
 class MusicBrainzCollectionPlugin(BeetsPlugin):
-    def __init__(self):
+    def __init__(self, client_id, client_secret, token=None, refresh_token=None):
         super(MusicBrainzCollectionPlugin, self).__init__()
-        config['musicbrainz']['pass'].redact = True
-        musicbrainzngs.auth(
-            config['musicbrainz']['user'].as_str(),
-            config['musicbrainz']['pass'].as_str(),
+        self.service = OAuth2Service(
+            client_id=client_id,
+            client_secret=client_secret,
+            authorize_url="https://musicbrainz.org/oauth2/authorize",
+            access_token_url="https://musicbrainz.org/oauth2/token",
+            base_url="https://musicbrainz.org/"
         )
-        self.config.add({'auto': False})
-        if self.config['auto']:
-            self.import_stages = [self.imported]
+        self.redirect_uri = "urn:ietf:wg:oauth:2.0:oob"
+        self.params = {'response_type': 'code',
+                       'redirect_uri': self.redirect_uri}
+
+    def authorize_url(self):
+        return self.service.get_authorize_url(**self.params)
+
+    def get_token(self, code):
+        session = service.get_raw_access_token(data={'code': code, 'redirect_uri': self.redirect_uri,
+                                                     'grant_type': 'authorization_code', 'token_type': 'bearer'})
+        t = session.json()
+        return t['access_token'], t['refresh_token']
 
     def commands(self):
         mbupdate = Subcommand('mbupdate',
@@ -112,3 +125,61 @@ def update_album_list(self, album_list):
         )
         submit_albums(collection_id, album_ids)
         self._log.info(u'...MusicBrainz collection updated.')
+
+
+class OAuth(BeetsPlugin):
+    def __init__(self):
+        super(OAuth, self).__init__()
+        self.config.add({
+            'client_id': 'qLYLK0LlrhifkgXq0N-J4w',  # Needs to be generated
+            'client_secret': 'XPiJdmSkh6uJDoWguo-J7g',
+            'tokenfile': 'oauth2_token.json'
+        })
+        self.config['client_id'].redact = True
+        self.config['client_secret'].redact = True
+        self.client = None
+        self.register_listener('import_begin', self.setup)
+
+    def setup(self, session=None):
+        client_id = self.config['client_id'].as_str()
+        secret = self.config['client_secret'].as_str()
+        try:
+            with open(self.tokenfile()) as f:
+                tokendata = json.load(f)
+        except IOError:
+            token, refresh_token = self.authenticate(client_id, secret)
+        else:
+            token = tokendata['token']
+            refresh_token = tokendata['refresh_token']
+
+        self.client = MusicBrainzCollectionPlugin(client_id, secret, token, refresh_token)
+
+    def authenticate(self, client_id, secret):
+        auth_client = MusicBrainzCollectionPlugin(client_id, secret)
+        try:
+            url = auth_client.authorize_url()
+        except Exception as e:
+            self._log.debug(u'authentication error: {0}', e)
+            raise beets.ui.UserError(u'Token request failed')
+
+        beets.ui.print_(u"To authenticate, visit:")
+        beets.ui.print_(url)
+
+        data = beets.ui.input_(u"Enter the string displayed in your browser:")
+
+        try:
+            token, refresh_token = auth_client.get_token(data)
+        except AUTH_ERRORS as e:
+            self._log.debug(u'authentication error: {0}', e)
+            raise beets.ui.UserError(u'Token request failed')
+
+        with open(tokenfile(), 'w') as f:
+            json.dump({'token': token, 'refresh_token': refresh_token}, f)
+
+        return token
+
+    @staticmethod
+    def tokenfile():
+        from os import path
+        basedir = path.abspath(path.dirname(__file__))
+        return "".join([basedir, '/token.json'])