<s> """<STR_LIT>""" <EOL> from __future__ import division , absolute_import , print_function <EOL> import beets . ui <EOL> from beets import logging <EOL> from beets import config <EOL> from beets . autotag . hooks import AlbumInfo , TrackInfo , Distance <EOL> from beets . plugins import BeetsPlugin <EOL> from beets . util import confit <EOL> from discogs_client import Release , Client <EOL> from discogs_client . exceptions import DiscogsAPIError <EOL> from requests . exceptions import ConnectionError <EOL> import beets <EOL> import re <EOL> import time <EOL> import json <EOL> import socket <EOL> import httplib <EOL> import os <EOL> urllib3_logger = logging . getLogger ( '<STR_LIT>' ) <EOL> urllib3_logger . setLevel ( logging . CRITICAL ) <EOL> USER_AGENT = u'<STR_LIT>' . format ( beets . __version__ ) <EOL> CONNECTION_ERRORS = ( ConnectionError , socket . error , httplib . HTTPException , <EOL> ValueError , <EOL> DiscogsAPIError ) <EOL> class DiscogsPlugin ( BeetsPlugin ) : <EOL> def __init__ ( self ) : <EOL> super ( DiscogsPlugin , self ) . __init__ ( ) <EOL> self . config . add ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:0.5> , <EOL> } ) <EOL> self . config [ '<STR_LIT>' ] . redact = True <EOL> self . config [ '<STR_LIT>' ] . redact = True <EOL> self . discogs_client = None <EOL> self . register_listener ( '<STR_LIT>' , self . setup ) <EOL> def setup ( self , session = None ) : <EOL> """<STR_LIT>""" <EOL> c_key = self . config [ '<STR_LIT>' ] . get ( unicode ) <EOL> c_secret = self . config [ '<STR_LIT>' ] . get ( unicode ) <EOL> try : <EOL> with open ( self . _tokenfile ( ) ) as f : <EOL> tokendata = json . load ( f ) <EOL> except IOError : <EOL> token , secret = self . authenticate ( c_key , c_secret ) <EOL> else : <EOL> token = tokendata [ '<STR_LIT>' ] <EOL> secret = tokendata [ '<STR_LIT>' ] <EOL> self . discogs_client = Client ( USER_AGENT , c_key , c_secret , <EOL> token , secret ) <EOL> def reset_auth ( self ) : <EOL> """<STR_LIT>""" <EOL> os . remove ( self . _tokenfile ( ) ) <EOL> self . setup ( ) <EOL> def _tokenfile ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . config [ '<STR_LIT>' ] . get ( confit . Filename ( in_app_dir = True ) ) <EOL> def authenticate ( self , c_key , c_secret ) : <EOL> auth_client = Client ( USER_AGENT , c_key , c_secret ) <EOL> try : <EOL> _ , _ , url = auth_client . get_authorize_url ( ) <EOL> except CONNECTION_ERRORS as e : <EOL> self . _log . debug ( u'<STR_LIT>' , e ) <EOL> raise beets . ui . UserError ( u'<STR_LIT>' ) <EOL> beets . ui . print_ ( u"<STR_LIT>" ) <EOL> beets . ui . print_ ( url ) <EOL> code = beets . ui . input_ ( u"<STR_LIT>" ) <EOL> try : <EOL> token , secret = auth_client . get_access_token ( code ) <EOL> except DiscogsAPIError : <EOL> raise beets . ui . UserError ( u'<STR_LIT>' ) <EOL> except CONNECTION_ERRORS as e : <EOL> self . _log . debug ( u'<STR_LIT>' , e ) <EOL> raise beets . ui . UserError ( u'<STR_LIT>' ) <EOL> self . _log . debug ( u'<STR_LIT>' , token , secret ) <EOL> with open ( self . _tokenfile ( ) , '<STR_LIT:w>' ) as f : <EOL> json . dump ( { '<STR_LIT>' : token , '<STR_LIT>' : secret } , f ) <EOL> return token , secret <EOL> def album_distance ( self , items , album_info , mapping ) : <EOL> """<STR_LIT>""" <EOL> dist = Distance ( ) <EOL> if album_info . data_source == '<STR_LIT>' : <EOL> dist . add ( '<STR_LIT:source>' , self . config [ '<STR_LIT>' ] . as_number ( ) ) <EOL> return dist <EOL> def candidates ( self , items , artist , album , va_likely ) : <EOL> """<STR_LIT>""" <EOL> if not self . discogs_client : <EOL> return <EOL> if va_likely : <EOL> query = album <EOL> else : <EOL> query = '<STR_LIT>' % ( artist , album ) <EOL> try : <EOL> return self . get_albums ( query ) <EOL> except DiscogsAPIError as e : <EOL> self . _log . debug ( u'<STR_LIT>' , e , query ) <EOL> if e . status_code == <NUM_LIT> : <EOL> self . reset_auth ( ) <EOL> return self . candidates ( items , artist , album , va_likely ) <EOL> else : <EOL> return [ ] <EOL> except CONNECTION_ERRORS : <EOL> self . _log . debug ( u'<STR_LIT>' , exc_info = True ) <EOL> return [ ] <EOL> def album_for_id ( self , album_id ) : <EOL> """<STR_LIT>""" <EOL> if not self . discogs_client : <EOL> return <EOL> self . _log . debug ( u'<STR_LIT>' , album_id ) <EOL> match = re . search ( r'<STR_LIT>' , <EOL> album_id ) <EOL> if not match : <EOL> return None <EOL> result = Release ( self . discogs_client , { '<STR_LIT:id>' : int ( match . group ( <NUM_LIT:2> ) ) } ) <EOL> try : <EOL> getattr ( result , '<STR_LIT:title>' ) <EOL> except DiscogsAPIError as e : <EOL> if e . status_code != <NUM_LIT> : <EOL> self . _log . debug ( u'<STR_LIT>' , e , result . _uri ) <EOL> if e . status_code == <NUM_LIT> : <EOL> self . reset_auth ( ) <EOL> return self . album_for_id ( album_id ) <EOL> return None <EOL> except CONNECTION_ERRORS : <EOL> self . _log . debug ( u'<STR_LIT>' , exc_info = True ) <EOL> return None <EOL> return self . get_album_info ( result ) <EOL> def get_albums ( self , query ) : <EOL> """<STR_LIT>""" <EOL> query = re . sub ( r'<STR_LIT>' , '<STR_LIT:U+0020>' , query ) . encode ( '<STR_LIT:ascii>' , "<STR_LIT:replace>" ) <EOL> query = re . sub ( r'<STR_LIT>' , '<STR_LIT>' , query ) <EOL> try : <EOL> releases = self . discogs_client . search ( query , <EOL> type = '<STR_LIT>' ) . page ( <NUM_LIT:1> ) <EOL> except CONNECTION_ERRORS : <EOL> self . _log . debug ( u"<STR_LIT>" , <EOL> query , exc_info = True ) <EOL> return [ ] <EOL> return [ self . get_album_info ( release ) for release in releases [ : <NUM_LIT:5> ] ] <EOL> def get_album_info ( self , result ) : <EOL> """<STR_LIT>""" <EOL> artist , artist_id = self . get_artist ( [ a . data for a in result . artists ] ) <EOL> album = re . sub ( r'<STR_LIT>' , '<STR_LIT:U+0020>' , result . title ) <EOL> album_id = result . data [ '<STR_LIT:id>' ] <EOL> tracks = self . get_tracks ( result . data [ '<STR_LIT>' ] ) <EOL> albumtype = '<STR_LIT:U+002CU+0020>' . join ( <EOL> result . data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . get ( '<STR_LIT>' , [ ] ) ) or None <EOL> va = result . data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:name>' ] . lower ( ) == '<STR_LIT>' <EOL> if va : <EOL> artist = config [ '<STR_LIT>' ] . get ( unicode ) <EOL> year = result . data [ '<STR_LIT>' ] <EOL> label = result . data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:name>' ] <EOL> mediums = len ( set ( t . medium for t in tracks ) ) <EOL> catalogno = result . data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> if catalogno == '<STR_LIT:none>' : <EOL> catalogno = None <EOL> country = result . data . get ( '<STR_LIT>' ) <EOL> media = result . data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:name>' ] <EOL> data_url = result . data [ '<STR_LIT>' ] <EOL> return AlbumInfo ( album , album_id , artist , artist_id , tracks , asin = None , <EOL> albumtype = albumtype , va = va , year = year , month = None , <EOL> day = None , label = label , mediums = mediums , <EOL> artist_sort = None , releasegroup_id = None , <EOL> catalognum = catalogno , script = None , language = None , <EOL> country = country , albumstatus = None , media = media , <EOL> albumdisambig = None , artist_credit = None , <EOL> original_year = None , original_month = None , <EOL> original_day = None , data_source = '<STR_LIT>' , <EOL> data_url = data_url ) <EOL> def get_artist ( self , artists ) : <EOL> """<STR_LIT>""" <EOL> artist_id = None <EOL> bits = [ ] <EOL> for i , artist in enumerate ( artists ) : <EOL> if not artist_id : <EOL> artist_id = artist [ '<STR_LIT:id>' ] <EOL> name = artist [ '<STR_LIT:name>' ] <EOL> name = re . sub ( r'<STR_LIT>' , '<STR_LIT>' , name ) <EOL> name = re . sub ( r'<STR_LIT>' , r'<STR_LIT>' , name ) <EOL> bits . append ( name ) <EOL> if artist [ '<STR_LIT>' ] and i < len ( artists ) - <NUM_LIT:1> : <EOL> bits . append ( artist [ '<STR_LIT>' ] ) <EOL> artist = '<STR_LIT:U+0020>' . join ( bits ) . replace ( '<STR_LIT>' , '<STR_LIT:U+002C>' ) or None <EOL> return artist , artist_id <EOL> def get_tracks ( self , tracklist ) : <EOL> """<STR_LIT>""" <EOL> tracks = [ ] <EOL> index_tracks = { } <EOL> index = <NUM_LIT:0> <EOL> for track in tracklist : <EOL> if track [ '<STR_LIT>' ] : <EOL> index += <NUM_LIT:1> <EOL> tracks . append ( self . get_track_info ( track , index ) ) <EOL> else : <EOL> index_tracks [ index + <NUM_LIT:1> ] = track [ '<STR_LIT:title>' ] <EOL> medium = None <EOL> medium_count , index_count = <NUM_LIT:0> , <NUM_LIT:0> <EOL> for track in tracks : <EOL> medium_is_index = track . medium and not track . medium_index and ( <EOL> len ( track . medium ) != <NUM_LIT:1> or <EOL> ord ( track . medium ) - <NUM_LIT:64> != medium_count + <NUM_LIT:1> <EOL> ) <EOL> if not medium_is_index and medium != track . medium : <EOL> medium = track . medium <EOL> medium_count += <NUM_LIT:1> <EOL> index_count = <NUM_LIT:0> <EOL> index_count += <NUM_LIT:1> <EOL> track . medium , track . medium_index = medium_count , index_count <EOL> for track in tracks : <EOL> if track . medium_index == <NUM_LIT:1> : <EOL> if track . index in index_tracks : <EOL> disctitle = index_tracks [ track . index ] <EOL> else : <EOL> disctitle = None <EOL> track . disctitle = disctitle <EOL> return tracks <EOL> def get_track_info ( self , track , index ) : <EOL> """<STR_LIT>""" <EOL> title = track [ '<STR_LIT:title>' ] <EOL> track_id = None <EOL> medium , medium_index = self . get_track_index ( track [ '<STR_LIT>' ] ) <EOL> artist , artist_id = self . get_artist ( track . get ( '<STR_LIT>' , [ ] ) ) <EOL> length = self . get_track_length ( track [ '<STR_LIT>' ] ) <EOL> return TrackInfo ( title , track_id , artist , artist_id , length , index , <EOL> medium , medium_index , artist_sort = None , <EOL> disctitle = None , artist_credit = None ) <EOL> def get_track_index ( self , position ) : <EOL> """<STR_LIT>""" <EOL> match = re . match ( r'<STR_LIT>' , position . upper ( ) ) <EOL> if match : <EOL> medium , index = match . groups ( ) <EOL> else : <EOL> self . _log . debug ( u'<STR_LIT>' , position ) <EOL> medium = index = None <EOL> return medium or None , index or None <EOL> def get_track_length ( self , duration ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> length = time . strptime ( duration , '<STR_LIT>' ) <EOL> except ValueError : <EOL> return None <EOL> return length . tm_min * <NUM_LIT> + length . tm_sec </s>
