<s> """<STR_LIT>""" <EOL> import sqlite3 <EOL> import os <EOL> from robot . libdocpkg import LibraryDocumentation <EOL> import robot . libraries <EOL> import logging <EOL> import json <EOL> import re <EOL> import sys <EOL> from watchdog . observers import Observer <EOL> from watchdog . observers . polling import PollingObserver <EOL> from watchdog . events import PatternMatchingEventHandler <EOL> """<STR_LIT>""" <EOL> class WatchdogHandler ( PatternMatchingEventHandler ) : <EOL> patterns = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> def __init__ ( self , kwdb , path ) : <EOL> PatternMatchingEventHandler . __init__ ( self ) <EOL> self . kwdb = kwdb <EOL> self . path = path <EOL> def on_created ( self , event ) : <EOL> self . kwdb . add ( event . src_path , monitor = False ) <EOL> def on_deleted ( self , event ) : <EOL> pass <EOL> def on_modified ( self , event ) : <EOL> self . kwdb . on_change ( event . src_path , event . event_type ) <EOL> class KeywordTable ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , dbfile = "<STR_LIT>" , poll = False ) : <EOL> self . db = sqlite3 . connect ( dbfile , check_same_thread = False ) <EOL> self . log = logging . getLogger ( __name__ ) <EOL> self . _create_db ( ) <EOL> self . observer = PollingObserver ( ) if poll else Observer ( ) <EOL> self . observer . start ( ) <EOL> def add ( self , name , monitor = True ) : <EOL> """<STR_LIT>""" <EOL> if os . path . isdir ( name ) : <EOL> if ( not os . path . basename ( name ) . startswith ( "<STR_LIT:.>" ) ) : <EOL> self . add_folder ( name ) <EOL> elif os . path . isfile ( name ) : <EOL> if ( ( self . _looks_like_resource_file ( name ) ) or <EOL> ( self . _looks_like_libdoc_file ( name ) ) or <EOL> ( self . _looks_like_library_file ( name ) ) ) : <EOL> self . add_file ( name ) <EOL> else : <EOL> self . add_library ( name ) <EOL> def on_change ( self , path , event_type ) : <EOL> """<STR_LIT>""" <EOL> sql = """<STR_LIT>""" <EOL> cursor = self . _execute ( sql , ( path , ) ) <EOL> results = cursor . fetchall ( ) <EOL> for result in results : <EOL> collection_id = result [ <NUM_LIT:0> ] <EOL> sql = """<STR_LIT>""" <EOL> cursor = self . _execute ( sql , ( collection_id , ) ) <EOL> self . _load_keywords ( collection_id , path = path ) <EOL> def _load_keywords ( self , collection_id , path = None , libdoc = None ) : <EOL> """<STR_LIT>""" <EOL> if libdoc is None and path is None : <EOL> raise ( Exception ( "<STR_LIT>" ) ) <EOL> if libdoc is None : <EOL> libdoc = LibraryDocumentation ( path ) <EOL> if len ( libdoc . keywords ) > <NUM_LIT:0> : <EOL> for keyword in libdoc . keywords : <EOL> self . _add_keyword ( collection_id , keyword . name , keyword . doc , keyword . args ) <EOL> def add_file ( self , path ) : <EOL> """<STR_LIT>""" <EOL> libdoc = LibraryDocumentation ( path ) <EOL> if len ( libdoc . keywords ) > <NUM_LIT:0> : <EOL> if libdoc . doc . startswith ( "<STR_LIT>" ) : <EOL> libdoc . doc = "<STR_LIT>" <EOL> collection_id = self . add_collection ( path , libdoc . name , libdoc . type , <EOL> libdoc . doc , libdoc . version , <EOL> libdoc . scope , libdoc . named_args , <EOL> libdoc . doc_format ) <EOL> self . _load_keywords ( collection_id , libdoc = libdoc ) <EOL> def add_library ( self , name ) : <EOL> """<STR_LIT>""" <EOL> libdoc = LibraryDocumentation ( name ) <EOL> if len ( libdoc . keywords ) > <NUM_LIT:0> : <EOL> collection_id = self . add_collection ( None , libdoc . name , libdoc . type , <EOL> libdoc . doc , libdoc . version , <EOL> libdoc . scope , libdoc . named_args , <EOL> libdoc . doc_format ) <EOL> self . _load_keywords ( collection_id , libdoc = libdoc ) <EOL> def add_folder ( self , dirname , watch = True ) : <EOL> """<STR_LIT>""" <EOL> ignore_file = os . path . join ( dirname , "<STR_LIT>" ) <EOL> exclude_patterns = [ ] <EOL> try : <EOL> with open ( ignore_file , "<STR_LIT:r>" ) as f : <EOL> exclude_patterns = [ ] <EOL> for line in f . readlines ( ) : <EOL> line = line . strip ( ) <EOL> if ( re . match ( r'<STR_LIT>' , line ) ) : continue <EOL> if len ( line . strip ( ) ) > <NUM_LIT:0> : <EOL> exclude_patterns . append ( line ) <EOL> except : <EOL> pass <EOL> for filename in os . listdir ( dirname ) : <EOL> path = os . path . join ( dirname , filename ) <EOL> ( basename , ext ) = os . path . splitext ( filename . lower ( ) ) <EOL> try : <EOL> if ( os . path . isdir ( path ) ) : <EOL> if ( not basename . startswith ( "<STR_LIT:.>" ) ) : <EOL> if os . access ( path , os . R_OK ) : <EOL> self . add_folder ( path , watch = False ) <EOL> else : <EOL> if ( ext in ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) ) : <EOL> if os . access ( path , os . R_OK ) : <EOL> self . add ( path ) <EOL> except Exception , e : <EOL> print "<STR_LIT>" , str ( e ) <EOL> if watch : <EOL> dirname = os . path . abspath ( dirname ) <EOL> event_handler = WatchdogHandler ( self , dirname ) <EOL> self . observer . schedule ( event_handler , dirname , recursive = True ) <EOL> def add_collection ( self , path , c_name , c_type , c_doc , c_version = "<STR_LIT>" , <EOL> c_scope = "<STR_LIT>" , c_namedargs = "<STR_LIT:yes>" , c_doc_format = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> if path is not None : <EOL> path = os . path . abspath ( path ) <EOL> cursor = self . db . cursor ( ) <EOL> cursor . execute ( """<STR_LIT>""" , ( c_name , c_type , c_version , c_scope , c_namedargs , path , c_doc , c_doc_format ) ) <EOL> collection_id = cursor . lastrowid <EOL> return collection_id <EOL> def add_installed_libraries ( self , extra_libs = [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) : <EOL> """<STR_LIT>""" <EOL> libdir = os . path . dirname ( robot . libraries . __file__ ) <EOL> loaded = [ ] <EOL> for filename in os . listdir ( libdir ) : <EOL> if filename . endswith ( "<STR_LIT>" ) or filename . endswith ( "<STR_LIT>" ) : <EOL> libname , ext = os . path . splitext ( filename ) <EOL> if ( libname . lower ( ) not in loaded and <EOL> not self . _should_ignore ( libname ) ) : <EOL> try : <EOL> self . add ( libname ) <EOL> loaded . append ( libname . lower ( ) ) <EOL> except Exception , e : <EOL> self . log . debug ( "<STR_LIT>" + str ( e ) ) <EOL> for library in extra_libs : <EOL> if ( library . lower ( ) not in loaded and <EOL> not self . _should_ignore ( library ) ) : <EOL> try : <EOL> self . add ( library ) <EOL> loaded . append ( library . lower ( ) ) <EOL> except Exception , e : <EOL> self . log . debug ( "<STR_LIT>" % ( library , str ( e ) ) ) <EOL> def get_collection ( self , collection_id ) : <EOL> """<STR_LIT>""" <EOL> sql = """<STR_LIT>""" <EOL> cursor = self . _execute ( sql , ( collection_id , collection_id ) ) <EOL> sql_result = cursor . fetchone ( ) <EOL> return { <EOL> "<STR_LIT>" : sql_result [ <NUM_LIT:0> ] , <EOL> "<STR_LIT:type>" : sql_result [ <NUM_LIT:1> ] , <EOL> "<STR_LIT:name>" : sql_result [ <NUM_LIT:2> ] , <EOL> "<STR_LIT:path>" : sql_result [ <NUM_LIT:3> ] , <EOL> "<STR_LIT>" : sql_result [ <NUM_LIT:4> ] , <EOL> "<STR_LIT:version>" : sql_result [ <NUM_LIT:5> ] , <EOL> "<STR_LIT>" : sql_result [ <NUM_LIT:6> ] , <EOL> "<STR_LIT>" : sql_result [ <NUM_LIT:7> ] , <EOL> "<STR_LIT>" : sql_result [ <NUM_LIT:8> ] <EOL> } <EOL> return sql_result <EOL> def get_collections ( self , pattern = "<STR_LIT:*>" , libtype = "<STR_LIT:*>" ) : <EOL> """<STR_LIT>""" <EOL> sql = """<STR_LIT>""" <EOL> cursor = self . _execute ( sql , ( self . _glob_to_sql ( pattern ) , <EOL> self . _glob_to_sql ( libtype ) ) ) <EOL> sql_result = cursor . fetchall ( ) <EOL> return [ { "<STR_LIT>" : result [ <NUM_LIT:0> ] , <EOL> "<STR_LIT:name>" : result [ <NUM_LIT:1> ] , <EOL> "<STR_LIT>" : result [ <NUM_LIT:2> ] . split ( "<STR_LIT:\n>" ) [ <NUM_LIT:0> ] , <EOL> "<STR_LIT:type>" : result [ <NUM_LIT:3> ] , <EOL> "<STR_LIT:path>" : result [ <NUM_LIT:4> ] <EOL> } for result in sql_result ] <EOL> def get_keyword_data ( self , collection_id ) : <EOL> sql = """<STR_LIT>""" <EOL> cursor = self . _execute ( sql , ( collection_id , ) ) <EOL> return cursor . fetchall ( ) <EOL> def get_keyword ( self , collection_id , name ) : <EOL> """<STR_LIT>""" <EOL> sql = """<STR_LIT>""" <EOL> cursor = self . _execute ( sql , ( collection_id , name ) ) <EOL> row = cursor . fetchone ( ) <EOL> if row is not None : <EOL> return { "<STR_LIT:name>" : row [ <NUM_LIT:0> ] , <EOL> "<STR_LIT:args>" : json . loads ( row [ <NUM_LIT:1> ] ) , <EOL> "<STR_LIT>" : row [ <NUM_LIT:2> ] , <EOL> "<STR_LIT>" : collection_id <EOL> } <EOL> return { } <EOL> def get_keyword_hierarchy ( self , pattern = "<STR_LIT:*>" ) : <EOL> """<STR_LIT>""" <EOL> sql = """<STR_LIT>""" <EOL> cursor = self . _execute ( sql , ( self . _glob_to_sql ( pattern ) , ) ) <EOL> libraries = [ ] <EOL> current_library = None <EOL> for row in cursor . fetchall ( ) : <EOL> ( c_id , c_name , c_path , k_name , k_doc ) = row <EOL> if c_id != current_library : <EOL> current_library = c_id <EOL> libraries . append ( { "<STR_LIT:name>" : c_name , "<STR_LIT>" : c_id , "<STR_LIT>" : [ ] , "<STR_LIT:path>" : c_path } ) <EOL> libraries [ - <NUM_LIT:1> ] [ "<STR_LIT>" ] . append ( { "<STR_LIT:name>" : k_name , "<STR_LIT>" : k_doc } ) <EOL> return libraries <EOL> def search ( self , pattern = "<STR_LIT:*>" , mode = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> pattern = self . _glob_to_sql ( pattern ) <EOL> COND = "<STR_LIT>" <EOL> args = [ pattern , pattern ] <EOL> if mode == "<STR_LIT:name>" : <EOL> COND = "<STR_LIT>" <EOL> args = [ pattern , ] <EOL> sql = """<STR_LIT>""" % COND <EOL> cursor = self . _execute ( sql , args ) <EOL> result = [ ( row [ <NUM_LIT:0> ] , row [ <NUM_LIT:1> ] , row [ <NUM_LIT:2> ] , row [ <NUM_LIT:3> ] . strip ( ) . split ( "<STR_LIT:\n>" ) [ <NUM_LIT:0> ] ) <EOL> for row in cursor . fetchall ( ) ] <EOL> return list ( set ( result ) ) <EOL> def get_keywords ( self , pattern = "<STR_LIT:*>" ) : <EOL> """<STR_LIT>""" <EOL> sql = """<STR_LIT>""" <EOL> pattern = self . _glob_to_sql ( pattern ) <EOL> cursor = self . _execute ( sql , ( pattern , ) ) <EOL> result = [ ( row [ <NUM_LIT:0> ] , row [ <NUM_LIT:1> ] , row [ <NUM_LIT:2> ] , row [ <NUM_LIT:3> ] , row [ <NUM_LIT:4> ] ) <EOL> for row in cursor . fetchall ( ) ] <EOL> return list ( set ( result ) ) <EOL> def reset ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _execute ( "<STR_LIT>" ) <EOL> self . _execute ( "<STR_LIT>" ) <EOL> def _looks_like_library_file ( self , name ) : <EOL> return name . endswith ( "<STR_LIT>" ) <EOL> def _looks_like_libdoc_file ( self , name ) : <EOL> """<STR_LIT>""" <EOL> if name . lower ( ) . endswith ( "<STR_LIT>" ) : <EOL> with open ( name , "<STR_LIT:r>" ) as f : <EOL> data = f . read ( <NUM_LIT:200> ) <EOL> index = data . lower ( ) . find ( "<STR_LIT>" ) <EOL> if index > <NUM_LIT:0> : <EOL> return True <EOL> return False <EOL> def _looks_like_resource_file ( self , name ) : <EOL> """<STR_LIT>""" <EOL> if ( re . search ( r'<STR_LIT>' , name ) ) : <EOL> return False <EOL> found_keyword_table = False <EOL> if ( name . lower ( ) . endswith ( "<STR_LIT>" ) or <EOL> name . lower ( ) . endswith ( "<STR_LIT>" ) or <EOL> name . lower ( ) . endswith ( "<STR_LIT>" ) ) : <EOL> with open ( name , "<STR_LIT:r>" ) as f : <EOL> data = f . read ( ) <EOL> for match in re . finditer ( r'<STR_LIT>' , <EOL> data , re . MULTILINE | re . IGNORECASE ) : <EOL> if ( re . match ( r'<STR_LIT>' , match . group ( <NUM_LIT:1> ) , re . IGNORECASE ) ) : <EOL> return False <EOL> if ( not found_keyword_table and <EOL> re . match ( r'<STR_LIT>' , match . group ( <NUM_LIT:1> ) , re . IGNORECASE ) ) : <EOL> found_keyword_table = True <EOL> return found_keyword_table <EOL> def _should_ignore ( self , name ) : <EOL> """<STR_LIT>""" <EOL> _name = name . lower ( ) <EOL> return ( _name . startswith ( "<STR_LIT>" ) or <EOL> _name . startswith ( "<STR_LIT:_>" ) or <EOL> _name in ( "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> def _execute ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> cursor = self . db . cursor ( ) <EOL> cursor . execute ( * args ) <EOL> return cursor <EOL> def _add_keyword ( self , collection_id , name , doc , args ) : <EOL> """<STR_LIT>""" <EOL> argstring = json . dumps ( args ) <EOL> self . db . execute ( """<STR_LIT>""" , ( collection_id , name , doc , argstring ) ) <EOL> def _create_db ( self ) : <EOL> if not self . _table_exists ( "<STR_LIT>" ) : <EOL> self . db . execute ( """<STR_LIT>""" ) <EOL> self . db . execute ( """<STR_LIT>""" ) <EOL> if not self . _table_exists ( "<STR_LIT>" ) : <EOL> self . db . execute ( """<STR_LIT>""" ) <EOL> self . db . execute ( """<STR_LIT>""" ) <EOL> def _glob_to_sql ( self , string ) : <EOL> """<STR_LIT>""" <EOL> table = ( ( r'<STR_LIT:\\>' , chr ( <NUM_LIT:1> ) ) , ( r'<STR_LIT>' , chr ( <NUM_LIT:2> ) ) , ( r'<STR_LIT>' , chr ( <NUM_LIT:3> ) ) , <EOL> ( r'<STR_LIT:%>' , r'<STR_LIT>' ) , ( r'<STR_LIT:?>' , '<STR_LIT:_>' ) , ( r'<STR_LIT:*>' , '<STR_LIT:%>' ) , <EOL> ( chr ( <NUM_LIT:1> ) , r'<STR_LIT:\\>' ) , ( chr ( <NUM_LIT:2> ) , r'<STR_LIT>' ) , ( chr ( <NUM_LIT:3> ) , r'<STR_LIT>' ) ) <EOL> for ( a , b ) in table : <EOL> string = string . replace ( a , b ) <EOL> string = string [ <NUM_LIT:1> : ] if string . startswith ( "<STR_LIT>" ) else "<STR_LIT:%>" + string <EOL> string = string [ : - <NUM_LIT:1> ] if string . endswith ( "<STR_LIT:$>" ) else string + "<STR_LIT:%>" <EOL> return string <EOL> def _table_exists ( self , name ) : <EOL> cursor = self . db . execute ( """<STR_LIT>""" % name ) <EOL> return len ( cursor . fetchall ( ) ) > <NUM_LIT:0> </s>
