<s> """<STR_LIT>""" <EOL> from __future__ import division , absolute_import , print_function <EOL> import os <EOL> import threading <EOL> import subprocess <EOL> import tempfile <EOL> import shlex <EOL> from string import Template <EOL> from beets import ui , util , plugins , config <EOL> from beets . plugins import BeetsPlugin <EOL> from beets . util . confit import ConfigTypeError <EOL> from beets import art <EOL> from beets . util . artresizer import ArtResizer <EOL> _fs_lock = threading . Lock ( ) <EOL> _temp_files = [ ] <EOL> ALIASES = { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> } <EOL> LOSSLESS_FORMATS = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> def replace_ext ( path , ext ) : <EOL> """<STR_LIT>""" <EOL> return os . path . splitext ( path ) [ <NUM_LIT:0> ] + b'<STR_LIT:.>' + ext <EOL> def get_format ( fmt = None ) : <EOL> """<STR_LIT>""" <EOL> if not fmt : <EOL> fmt = config [ '<STR_LIT>' ] [ '<STR_LIT>' ] . get ( unicode ) . lower ( ) <EOL> fmt = ALIASES . get ( fmt , fmt ) <EOL> try : <EOL> format_info = config [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ fmt ] . get ( dict ) <EOL> command = format_info [ '<STR_LIT>' ] <EOL> extension = format_info [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> raise ui . UserError ( <EOL> u'<STR_LIT>' <EOL> . format ( fmt ) <EOL> ) <EOL> except ConfigTypeError : <EOL> command = config [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ fmt ] . get ( bytes ) <EOL> extension = fmt <EOL> keys = config [ '<STR_LIT>' ] . keys ( ) <EOL> if '<STR_LIT>' in keys : <EOL> command = config [ '<STR_LIT>' ] [ '<STR_LIT>' ] . get ( unicode ) <EOL> elif '<STR_LIT>' in keys : <EOL> command = u'<STR_LIT>' . format ( <EOL> config [ '<STR_LIT>' ] [ '<STR_LIT>' ] . get ( unicode ) <EOL> ) <EOL> if '<STR_LIT>' in keys : <EOL> extension = config [ '<STR_LIT>' ] [ '<STR_LIT>' ] . get ( unicode ) <EOL> return ( command . encode ( '<STR_LIT:utf8>' ) , extension . encode ( '<STR_LIT:utf8>' ) ) <EOL> def should_transcode ( item , fmt ) : <EOL> """<STR_LIT>""" <EOL> if config [ '<STR_LIT>' ] [ '<STR_LIT>' ] and not ( item . format . lower ( ) in LOSSLESS_FORMATS ) : <EOL> return False <EOL> maxbr = config [ '<STR_LIT>' ] [ '<STR_LIT>' ] . get ( int ) <EOL> return fmt . lower ( ) != item . format . lower ( ) or item . bitrate >= <NUM_LIT:1000> * maxbr <EOL> class ConvertPlugin ( BeetsPlugin ) : <EOL> def __init__ ( self ) : <EOL> super ( ConvertPlugin , self ) . __init__ ( ) <EOL> self . config . add ( { <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : False , <EOL> u'<STR_LIT>' : util . cpu_count ( ) , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : { <EOL> u'<STR_LIT>' : { <EOL> u'<STR_LIT>' : u'<STR_LIT>' <EOL> u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> } , <EOL> u'<STR_LIT>' : { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> } , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : <EOL> u'<STR_LIT>' , <EOL> u'<STR_LIT>' : <EOL> u'<STR_LIT>' , <EOL> u'<STR_LIT>' : <EOL> u'<STR_LIT>' , <EOL> } , <EOL> u'<STR_LIT>' : <NUM_LIT> , <EOL> u'<STR_LIT>' : False , <EOL> u'<STR_LIT>' : None , <EOL> u'<STR_LIT>' : False , <EOL> u'<STR_LIT>' : True , <EOL> u'<STR_LIT>' : { } , <EOL> u'<STR_LIT>' : False , <EOL> u'<STR_LIT>' : False , <EOL> u'<STR_LIT>' : <NUM_LIT:0> , <EOL> } ) <EOL> self . import_stages = [ self . auto_convert ] <EOL> self . register_listener ( '<STR_LIT>' , self . _cleanup ) <EOL> def commands ( self ) : <EOL> cmd = ui . Subcommand ( '<STR_LIT>' , help = u'<STR_LIT>' ) <EOL> cmd . parser . add_option ( '<STR_LIT>' , '<STR_LIT>' , action = '<STR_LIT:store_true>' , <EOL> help = u'<STR_LIT>' ) <EOL> cmd . parser . add_option ( '<STR_LIT>' , '<STR_LIT>' , action = '<STR_LIT:store>' , type = '<STR_LIT:int>' , <EOL> help = u'<STR_LIT>' ) <EOL> cmd . parser . add_option ( '<STR_LIT>' , '<STR_LIT>' , action = '<STR_LIT:store_true>' , <EOL> dest = '<STR_LIT>' , help = u'<STR_LIT>' ) <EOL> cmd . parser . add_option ( '<STR_LIT>' , '<STR_LIT>' , action = '<STR_LIT:store>' , <EOL> help = u'<STR_LIT>' ) <EOL> cmd . parser . add_option ( '<STR_LIT>' , '<STR_LIT>' , action = '<STR_LIT:store>' , dest = '<STR_LIT>' , <EOL> help = u'<STR_LIT>' ) <EOL> cmd . parser . add_option ( '<STR_LIT>' , '<STR_LIT>' , action = '<STR_LIT:store_true>' , dest = '<STR_LIT:yes>' , <EOL> help = u'<STR_LIT>' ) <EOL> cmd . parser . add_album_option ( ) <EOL> cmd . func = self . convert_func <EOL> return [ cmd ] <EOL> def auto_convert ( self , config , task ) : <EOL> if self . config [ '<STR_LIT>' ] : <EOL> for item in task . imported_items ( ) : <EOL> self . convert_on_import ( config . lib , item ) <EOL> def encode ( self , command , source , dest , pretend = False ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( command , bytes ) <EOL> assert isinstance ( source , bytes ) <EOL> assert isinstance ( dest , bytes ) <EOL> quiet = self . config [ '<STR_LIT>' ] . get ( bool ) <EOL> if not quiet and not pretend : <EOL> self . _log . info ( u'<STR_LIT>' , util . displayable_path ( source ) ) <EOL> args = shlex . split ( command ) <EOL> for i , arg in enumerate ( args ) : <EOL> args [ i ] = Template ( arg ) . safe_substitute ( { <EOL> b'<STR_LIT:source>' : source , <EOL> b'<STR_LIT>' : dest , <EOL> } ) <EOL> if pretend : <EOL> self . _log . info ( '<STR_LIT:U+0020>' . join ( ui . decargs ( args ) ) ) <EOL> return <EOL> try : <EOL> util . command_output ( args ) <EOL> except subprocess . CalledProcessError as exc : <EOL> self . _log . info ( u'<STR_LIT>' , <EOL> util . displayable_path ( source ) ) <EOL> self . _log . debug ( u'<STR_LIT>' , <EOL> exc . cmd . decode ( '<STR_LIT:utf8>' , '<STR_LIT:ignore>' ) , <EOL> exc . returncode ) <EOL> util . remove ( dest ) <EOL> util . prune_dirs ( os . path . dirname ( dest ) ) <EOL> raise <EOL> except OSError as exc : <EOL> raise ui . UserError ( <EOL> u"<STR_LIT>" . format ( <EOL> '<STR_LIT:U+0020>' . join ( args ) , exc <EOL> ) <EOL> ) <EOL> if not quiet and not pretend : <EOL> self . _log . info ( u'<STR_LIT>' , <EOL> util . displayable_path ( source ) ) <EOL> def convert_item ( self , dest_dir , keep_new , path_formats , fmt , <EOL> pretend = False ) : <EOL> command , ext = get_format ( fmt ) <EOL> item , original , converted = None , None , None <EOL> while True : <EOL> item = yield ( item , original , converted ) <EOL> dest = item . destination ( basedir = dest_dir , <EOL> path_formats = path_formats ) <EOL> if keep_new : <EOL> original = dest <EOL> converted = item . path <EOL> if should_transcode ( item , fmt ) : <EOL> converted = replace_ext ( converted , ext ) <EOL> else : <EOL> original = item . path <EOL> if should_transcode ( item , fmt ) : <EOL> dest = replace_ext ( dest , ext ) <EOL> converted = dest <EOL> if not pretend : <EOL> with _fs_lock : <EOL> util . mkdirall ( dest ) <EOL> if os . path . exists ( util . syspath ( dest ) ) : <EOL> self . _log . info ( u'<STR_LIT>' , <EOL> util . displayable_path ( item . path ) ) <EOL> continue <EOL> if keep_new : <EOL> if pretend : <EOL> self . _log . info ( u'<STR_LIT>' , <EOL> util . displayable_path ( item . path ) , <EOL> util . displayable_path ( original ) ) <EOL> else : <EOL> self . _log . info ( u'<STR_LIT>' , <EOL> util . displayable_path ( original ) ) <EOL> util . move ( item . path , original ) <EOL> if should_transcode ( item , fmt ) : <EOL> try : <EOL> self . encode ( command , original , converted , pretend ) <EOL> except subprocess . CalledProcessError : <EOL> continue <EOL> else : <EOL> if pretend : <EOL> self . _log . info ( u'<STR_LIT>' , <EOL> util . displayable_path ( original ) , <EOL> util . displayable_path ( converted ) ) <EOL> else : <EOL> self . _log . info ( u'<STR_LIT>' , <EOL> util . displayable_path ( item . path ) ) <EOL> util . copy ( original , converted ) <EOL> if pretend : <EOL> continue <EOL> item . try_write ( path = converted ) <EOL> if keep_new : <EOL> item . path = converted <EOL> item . read ( ) <EOL> item . store ( ) <EOL> if self . config [ '<STR_LIT>' ] : <EOL> album = item . get_album ( ) <EOL> if album and album . artpath : <EOL> self . _log . debug ( u'<STR_LIT>' , <EOL> util . displayable_path ( album . artpath ) ) <EOL> art . embed_item ( self . _log , item , album . artpath , <EOL> itempath = converted ) <EOL> if keep_new : <EOL> plugins . send ( '<STR_LIT>' , item = item , <EOL> dest = dest , keepnew = True ) <EOL> else : <EOL> plugins . send ( '<STR_LIT>' , item = item , <EOL> dest = converted , keepnew = False ) <EOL> def copy_album_art ( self , album , dest_dir , path_formats , pretend = False ) : <EOL> """<STR_LIT>""" <EOL> if not album or not album . artpath : <EOL> return <EOL> album_item = album . items ( ) . get ( ) <EOL> if not album_item : <EOL> return <EOL> dest = album_item . destination ( basedir = dest_dir , <EOL> path_formats = path_formats ) <EOL> dest = os . path . join ( * util . components ( dest ) [ : - <NUM_LIT:1> ] ) <EOL> dest = album . art_destination ( album . artpath , item_dir = dest ) <EOL> if album . artpath == dest : <EOL> return <EOL> if not pretend : <EOL> util . mkdirall ( dest ) <EOL> if os . path . exists ( util . syspath ( dest ) ) : <EOL> self . _log . info ( u'<STR_LIT>' , <EOL> util . displayable_path ( album . artpath ) ) <EOL> return <EOL> resize = False <EOL> maxwidth = None <EOL> if self . config [ '<STR_LIT>' ] : <EOL> maxwidth = self . config [ '<STR_LIT>' ] . get ( int ) <EOL> size = ArtResizer . shared . get_size ( album . artpath ) <EOL> self . _log . debug ( '<STR_LIT>' , size ) <EOL> if size : <EOL> resize = size [ <NUM_LIT:0> ] > maxwidth <EOL> else : <EOL> self . _log . warning ( u'<STR_LIT>' <EOL> u'<STR_LIT>' ) <EOL> if resize : <EOL> self . _log . info ( u'<STR_LIT>' , <EOL> util . displayable_path ( album . artpath ) , <EOL> util . displayable_path ( dest ) ) <EOL> if not pretend : <EOL> ArtResizer . shared . resize ( maxwidth , album . artpath , dest ) <EOL> else : <EOL> if pretend : <EOL> self . _log . info ( u'<STR_LIT>' , <EOL> util . displayable_path ( album . artpath ) , <EOL> util . displayable_path ( dest ) ) <EOL> else : <EOL> self . _log . info ( u'<STR_LIT>' , <EOL> util . displayable_path ( album . artpath ) , <EOL> util . displayable_path ( dest ) ) <EOL> util . copy ( album . artpath , dest ) <EOL> def convert_func ( self , lib , opts , args ) : <EOL> if not opts . dest : <EOL> opts . dest = self . config [ '<STR_LIT>' ] . get ( ) <EOL> if not opts . dest : <EOL> raise ui . UserError ( u'<STR_LIT>' ) <EOL> opts . dest = util . bytestring_path ( opts . dest ) <EOL> if not opts . threads : <EOL> opts . threads = self . config [ '<STR_LIT>' ] . get ( int ) <EOL> if self . config [ '<STR_LIT>' ] : <EOL> path_formats = ui . get_path_formats ( self . config [ '<STR_LIT>' ] ) <EOL> else : <EOL> path_formats = ui . get_path_formats ( ) <EOL> if not opts . format : <EOL> opts . format = self . config [ '<STR_LIT>' ] . get ( unicode ) . lower ( ) <EOL> pretend = opts . pretend if opts . pretend is not None else self . config [ '<STR_LIT>' ] . get ( bool ) <EOL> if not pretend : <EOL> ui . commands . list_items ( lib , ui . decargs ( args ) , opts . album ) <EOL> if not ( opts . yes or ui . input_yn ( u"<STR_LIT>" ) ) : <EOL> return <EOL> if opts . album : <EOL> albums = lib . albums ( ui . decargs ( args ) ) <EOL> items = ( i for a in albums for i in a . items ( ) ) <EOL> if self . config [ '<STR_LIT>' ] : <EOL> for album in albums : <EOL> self . copy_album_art ( album , opts . dest , path_formats , <EOL> pretend ) <EOL> else : <EOL> items = iter ( lib . items ( ui . decargs ( args ) ) ) <EOL> convert = [ self . convert_item ( opts . dest , <EOL> opts . keep_new , <EOL> path_formats , <EOL> opts . format , <EOL> pretend ) <EOL> for _ in range ( opts . threads ) ] <EOL> pipe = util . pipeline . Pipeline ( [ items , convert ] ) <EOL> pipe . run_parallel ( ) <EOL> def convert_on_import ( self , lib , item ) : <EOL> """<STR_LIT>""" <EOL> fmt = self . config [ '<STR_LIT>' ] . get ( unicode ) . lower ( ) <EOL> if should_transcode ( item , fmt ) : <EOL> command , ext = get_format ( ) <EOL> tmpdir = self . config [ '<STR_LIT>' ] . get ( ) <EOL> fd , dest = tempfile . mkstemp ( '<STR_LIT:.>' + ext , dir = tmpdir ) <EOL> os . close ( fd ) <EOL> dest = util . bytestring_path ( dest ) <EOL> _temp_files . append ( dest ) <EOL> try : <EOL> self . encode ( command , item . path , dest ) <EOL> except subprocess . CalledProcessError : <EOL> return <EOL> item . path = dest <EOL> item . write ( ) <EOL> item . read ( ) <EOL> item . store ( ) <EOL> def _cleanup ( self , task , session ) : <EOL> for path in task . old_paths : <EOL> if path in _temp_files : <EOL> if os . path . isfile ( path ) : <EOL> util . remove ( path ) <EOL> _temp_files . remove ( path ) </s>
