<s> """<STR_LIT>""" <EOL> import os <EOL> import re <EOL> from . import log <EOL> from . compat import ConfigParser , is_py2 <EOL> from . PluginInfo import PluginInfo <EOL> from . import PLUGIN_NAME_FORBIDEN_STRING <EOL> from . IPluginLocator import IPluginLocator <EOL> class IPluginFileAnalyzer ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name ) : <EOL> self . name = name <EOL> def isValidPlugin ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" % self ) <EOL> def getInfosDictFromPlugin ( self , dirpath , filename ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" % self ) <EOL> class PluginFileAnalyzerWithInfoFile ( IPluginFileAnalyzer ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name , extensions = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> IPluginFileAnalyzer . __init__ ( self , name ) <EOL> self . setPluginInfoExtension ( extensions ) <EOL> def setPluginInfoExtension ( self , extensions ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( extensions , tuple ) : <EOL> extensions = ( extensions , ) <EOL> self . expectedExtensions = extensions <EOL> def isValidPlugin ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> res = False <EOL> for ext in self . expectedExtensions : <EOL> if filename . endswith ( "<STR_LIT>" % ext ) : <EOL> res = True <EOL> break <EOL> return res <EOL> def getPluginNameAndModuleFromStream ( self , infoFileObject , candidate_infofile = None ) : <EOL> """<STR_LIT>""" <EOL> config_parser = ConfigParser ( ) <EOL> try : <EOL> if is_py2 : <EOL> config_parser . readfp ( infoFileObject ) <EOL> else : <EOL> config_parser . read_file ( infoFileObject ) <EOL> except Exception as e : <EOL> log . debug ( "<STR_LIT>" % ( candidate_infofile , e ) ) <EOL> return ( None , None , None ) <EOL> if not config_parser . has_section ( "<STR_LIT>" ) : <EOL> log . debug ( "<STR_LIT>" % candidate_infofile ) <EOL> return ( None , None , None ) <EOL> if not config_parser . has_option ( "<STR_LIT>" , "<STR_LIT:Name>" ) or not config_parser . has_option ( "<STR_LIT>" , "<STR_LIT>" ) : <EOL> log . debug ( "<STR_LIT>" % candidate_infofile ) <EOL> return ( None , None , None ) <EOL> name = config_parser . get ( "<STR_LIT>" , "<STR_LIT:Name>" ) <EOL> name = name . strip ( ) <EOL> if PLUGIN_NAME_FORBIDEN_STRING in name : <EOL> log . debug ( "<STR_LIT>" % ( PLUGIN_NAME_FORBIDEN_STRING , <EOL> candidate_infofile ) ) <EOL> return ( None , None , None ) <EOL> return ( name , config_parser . get ( "<STR_LIT>" , "<STR_LIT>" ) , config_parser ) <EOL> def _extractCorePluginInfo ( self , directory , filename ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( filename , str ) : <EOL> name , moduleName , config_parser = self . getPluginNameAndModuleFromStream ( filename ) <EOL> else : <EOL> candidate_infofile_path = os . path . join ( directory , filename ) <EOL> with open ( candidate_infofile_path ) as candidate_infofile : <EOL> name , moduleName , config_parser = self . getPluginNameAndModuleFromStream ( candidate_infofile , candidate_infofile_path ) <EOL> if ( name , moduleName , config_parser ) == ( None , None , None ) : <EOL> return ( None , None ) <EOL> infos = { "<STR_LIT:name>" : name , "<STR_LIT:path>" : os . path . join ( directory , moduleName ) } <EOL> return infos , config_parser <EOL> def _extractBasicPluginInfo ( self , directory , filename ) : <EOL> """<STR_LIT>""" <EOL> infos , config_parser = self . _extractCorePluginInfo ( directory , filename ) <EOL> if infos and config_parser and config_parser . has_section ( "<STR_LIT>" ) : <EOL> if config_parser . has_option ( "<STR_LIT>" , "<STR_LIT>" ) : <EOL> infos [ "<STR_LIT>" ] = config_parser . get ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> if config_parser . has_option ( "<STR_LIT>" , "<STR_LIT>" ) : <EOL> infos [ "<STR_LIT:version>" ] = config_parser . get ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> if config_parser . has_option ( "<STR_LIT>" , "<STR_LIT>" ) : <EOL> infos [ "<STR_LIT>" ] = config_parser . get ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> if config_parser . has_option ( "<STR_LIT>" , "<STR_LIT>" ) : <EOL> infos [ "<STR_LIT>" ] = config_parser . get ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> if config_parser . has_option ( "<STR_LIT>" , "<STR_LIT>" ) : <EOL> infos [ "<STR_LIT:description>" ] = config_parser . get ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> return infos , config_parser <EOL> def getInfosDictFromPlugin ( self , dirpath , filename ) : <EOL> """<STR_LIT>""" <EOL> infos , config_parser = self . _extractBasicPluginInfo ( dirpath , filename ) <EOL> if not infos or infos . get ( "<STR_LIT:name>" , None ) is None : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if not infos or infos . get ( "<STR_LIT:path>" , None ) is None : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return infos , config_parser <EOL> class PluginFileAnalyzerMathingRegex ( IPluginFileAnalyzer ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name , regexp ) : <EOL> IPluginFileAnalyzer . __init__ ( self , name ) <EOL> self . regexp = regexp <EOL> def isValidPlugin ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> reg = re . compile ( self . regexp ) <EOL> if reg . match ( filename ) is not None : <EOL> return True <EOL> return False <EOL> def getInfosDictFromPlugin ( self , dirpath , filename ) : <EOL> """<STR_LIT>""" <EOL> infos = { } <EOL> module_name = os . path . splitext ( filename ) [ <NUM_LIT:0> ] <EOL> plugin_filename = os . path . join ( dirpath , filename ) <EOL> if module_name == "<STR_LIT>" : <EOL> module_name = os . path . basename ( dirpath ) <EOL> plugin_filename = dirpath <EOL> infos [ "<STR_LIT:name>" ] = "<STR_LIT:%s>" % module_name <EOL> infos [ "<STR_LIT:path>" ] = plugin_filename <EOL> cf_parser = ConfigParser ( ) <EOL> cf_parser . add_section ( "<STR_LIT>" ) <EOL> cf_parser . set ( "<STR_LIT>" , "<STR_LIT:Name>" , infos [ "<STR_LIT:name>" ] ) <EOL> cf_parser . set ( "<STR_LIT>" , "<STR_LIT>" , infos [ "<STR_LIT:path>" ] ) <EOL> return infos , cf_parser <EOL> class PluginFileLocator ( IPluginLocator ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , analyzers = None , plugin_info_cls = PluginInfo ) : <EOL> """<STR_LIT>""" <EOL> IPluginLocator . __init__ ( self ) <EOL> self . _discovered_plugins = { } <EOL> self . setPluginPlaces ( None ) <EOL> self . _analyzers = analyzers <EOL> if self . _analyzers is None : <EOL> self . _analyzers = [ PluginFileAnalyzerWithInfoFile ( "<STR_LIT>" ) ] <EOL> self . _default_plugin_info_cls = PluginInfo <EOL> self . _plugin_info_cls_map = { } <EOL> self . _max_size = <NUM_LIT> * <NUM_LIT> <EOL> self . recursive = True <EOL> def disableRecursiveScan ( self ) : <EOL> """<STR_LIT>""" <EOL> self . recursive = False <EOL> def setAnalyzers ( self , analyzers ) : <EOL> """<STR_LIT>""" <EOL> self . _analyzers = analyzers <EOL> def removeAnalyzers ( self , name ) : <EOL> """<STR_LIT>""" <EOL> analyzersListCopy = self . _analyzers [ : ] <EOL> foundAndRemoved = False <EOL> for obj in analyzersListCopy : <EOL> if obj . name == name : <EOL> self . _analyzers . remove ( obj ) <EOL> foundAndRemoved = True <EOL> if not foundAndRemoved : <EOL> log . debug ( "<STR_LIT>" % name ) <EOL> def removeAllAnalyzer ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _analyzers = [ ] <EOL> def appendAnalyzer ( self , analyzer ) : <EOL> """<STR_LIT>""" <EOL> self . _analyzers . append ( analyzer ) <EOL> def _getInfoForPluginFromAnalyzer ( self , analyzer , dirpath , filename ) : <EOL> """<STR_LIT>""" <EOL> plugin_info_dict , config_parser = analyzer . getInfosDictFromPlugin ( dirpath , filename ) <EOL> if plugin_info_dict is None : <EOL> return None <EOL> plugin_info_cls = self . _plugin_info_cls_map . get ( analyzer . name , self . _default_plugin_info_cls ) <EOL> plugin_info = plugin_info_cls ( plugin_info_dict [ "<STR_LIT:name>" ] , plugin_info_dict [ "<STR_LIT:path>" ] ) <EOL> plugin_info . details = config_parser <EOL> return plugin_info <EOL> def locatePlugins ( self ) : <EOL> """<STR_LIT>""" <EOL> _candidates = [ ] <EOL> _discovered = { } <EOL> for directory in map ( os . path . abspath , self . plugins_places ) : <EOL> if not os . path . isdir ( directory ) : <EOL> log . debug ( "<STR_LIT>" % ( self . __class__ . __name__ , directory ) ) <EOL> continue <EOL> if self . recursive : <EOL> debug_txt_mode = "<STR_LIT>" <EOL> walk_iter = os . walk ( directory , followlinks = True ) <EOL> else : <EOL> debug_txt_mode = "<STR_LIT>" <EOL> walk_iter = [ ( directory , [ ] , os . listdir ( directory ) ) ] <EOL> log . debug ( "<STR_LIT>" % ( self . __class__ . __name__ , debug_txt_mode , directory ) ) <EOL> for item in walk_iter : <EOL> dirpath = item [ <NUM_LIT:0> ] <EOL> for filename in item [ <NUM_LIT:2> ] : <EOL> for analyzer in self . _analyzers : <EOL> if not analyzer . isValidPlugin ( filename ) : <EOL> log . debug ( "<STR_LIT>" % ( filename , analyzer . name ) ) <EOL> continue <EOL> candidate_infofile = os . path . join ( dirpath , filename ) <EOL> if candidate_infofile in _discovered : <EOL> log . debug ( "<STR_LIT>" % ( candidate_infofile , analyzer . name ) ) <EOL> continue <EOL> log . debug ( "<STR_LIT>" % ( self . __class__ . __name__ , candidate_infofile ) ) <EOL> plugin_info = self . _getInfoForPluginFromAnalyzer ( analyzer , dirpath , filename ) <EOL> if plugin_info is None : <EOL> log . warning ( "<STR_LIT>" % ( candidate_infofile , analyzer . name ) ) <EOL> break <EOL> if os . path . isdir ( plugin_info . path ) : <EOL> candidate_filepath = os . path . join ( plugin_info . path , "<STR_LIT>" ) <EOL> for _file in os . listdir ( plugin_info . path ) : <EOL> if _file . endswith ( "<STR_LIT>" ) : <EOL> self . _discovered_plugins [ os . path . join ( plugin_info . path , _file ) ] = candidate_filepath <EOL> _discovered [ os . path . join ( plugin_info . path , _file ) ] = candidate_filepath <EOL> elif ( plugin_info . path . endswith ( "<STR_LIT>" ) and os . path . isfile ( plugin_info . path ) ) or os . path . isfile ( plugin_info . path + "<STR_LIT>" ) : <EOL> candidate_filepath = plugin_info . path <EOL> if candidate_filepath . endswith ( "<STR_LIT>" ) : <EOL> candidate_filepath = candidate_filepath [ : - <NUM_LIT:3> ] <EOL> self . _discovered_plugins [ "<STR_LIT:.>" . join ( ( plugin_info . path , "<STR_LIT>" ) ) ] = candidate_filepath <EOL> _discovered [ "<STR_LIT:.>" . join ( ( plugin_info . path , "<STR_LIT>" ) ) ] = candidate_filepath <EOL> else : <EOL> log . error ( "<STR_LIT>" % ( candidate_infofile ) ) <EOL> break <EOL> _candidates . append ( ( candidate_infofile , candidate_filepath , plugin_info ) ) <EOL> _discovered [ candidate_infofile ] = candidate_filepath <EOL> self . _discovered_plugins [ candidate_infofile ] = candidate_filepath <EOL> return _candidates , len ( _candidates ) <EOL> def gatherCorePluginInfo ( self , directory , filename ) : <EOL> """<STR_LIT>""" <EOL> for analyzer in self . _analyzers : <EOL> if not analyzer . isValidPlugin ( filename ) : <EOL> continue <EOL> plugin_info = self . _getInfoForPluginFromAnalyzer ( analyzer , directory , filename ) <EOL> return plugin_info , plugin_info . details <EOL> return None , None <EOL> def getPluginNameAndModuleFromStream ( self , infoFileObject , candidate_infofile = None ) : <EOL> for analyzer in self . _analyzers : <EOL> if analyzer . name == "<STR_LIT>" : <EOL> return analyzer . getPluginNameAndModuleFromStream ( infoFileObject ) <EOL> else : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> def setPluginInfoClass ( self , picls , name = None ) : <EOL> """<STR_LIT>""" <EOL> if name is None : <EOL> self . _default_plugin_info_cls = picls <EOL> self . _plugin_info_cls_map = { } <EOL> else : <EOL> self . _plugin_info_cls_map [ name ] = picls <EOL> def setPluginPlaces ( self , directories_list ) : <EOL> """<STR_LIT>""" <EOL> if directories_list is None : <EOL> directories_list = [ os . path . dirname ( __file__ ) ] <EOL> self . plugins_places = directories_list <EOL> def updatePluginPlaces ( self , directories_list ) : <EOL> """<STR_LIT>""" <EOL> self . plugins_places = list ( set . union ( set ( directories_list ) , set ( self . plugins_places ) ) ) <EOL> def setPluginInfoExtension ( self , ext ) : <EOL> """<STR_LIT>""" <EOL> for analyzer in self . _analyzers : <EOL> if analyzer . name == "<STR_LIT>" : <EOL> analyzer . setPluginInfoExtension ( ext ) </s>
