<s> """<STR_LIT>""" <EOL> import dis <EOL> import opcode <EOL> import marshal <EOL> import time <EOL> import struct <EOL> import types <EOL> import operator <EOL> import traceback <EOL> import imp <EOL> from dis import findlinestarts , findlabels <EOL> from . . utils . log import logger <EOL> from . utils import show_bytecode <EOL> from . decl import ModuleDeclaration , TypeDeclaration , MethodDeclaration , FieldDeclaration , ImportDeclaration <EOL> from . . visitors . bytecode import BytecodeVisitor <EOL> from . . visitors . classes import ClassVisitor <EOL> from . . visitors . methods import MethodVisitor <EOL> from . . visitors . modules import ModuleVisitor <EOL> CO_OPTIMIZED = <NUM_LIT> <EOL> CO_NEWLOCALS = <NUM_LIT> <EOL> CO_VARARGS = <NUM_LIT> <EOL> CO_VARKEYWORDS = <NUM_LIT> <EOL> CO_NESTED = <NUM_LIT> <EOL> CO_GENERATOR = <NUM_LIT> <EOL> CO_NOFREE = <NUM_LIT> <EOL> CO_GENERATOR_ALLOWED = <NUM_LIT> <EOL> CO_FUTURE_DIVISION = <NUM_LIT> <EOL> CO_FUTURE_ABSOLUTE_IMPORT = <NUM_LIT> <EOL> CO_FUTURE_WITH_STATEMENT = <NUM_LIT> <EOL> POP_TOP = <NUM_LIT:1> <EOL> IMPORT_STAR = <NUM_LIT> <EOL> BUILD_CLASS = <NUM_LIT> <EOL> STORE_NAME = <NUM_LIT> <EOL> LOAD_CONST = <NUM_LIT:100> <EOL> LOAD_ATTR = <NUM_LIT> <EOL> IMPORT_NAME = <NUM_LIT> <EOL> IMPORT_FROM = <NUM_LIT> <EOL> STORE_FAST = <NUM_LIT> <EOL> CALL_FUNCTION = <NUM_LIT> <EOL> MAKE_FUNCTION = <NUM_LIT> <EOL> MAKE_CLOSURE = <NUM_LIT> <EOL> class BytecodeObject ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , pyc_file , lazy_load = True ) : <EOL> """<STR_LIT>""" <EOL> self . code = None <EOL> self . magic = None <EOL> self . moddate = None <EOL> self . modif_date = None <EOL> self . pyc_file = pyc_file <EOL> if not lazy_load : <EOL> self . parse ( ) <EOL> self . main_module = None <EOL> self . bytecode = [ ] <EOL> self . all_decls = set ( ) <EOL> def parse ( self ) : <EOL> """<STR_LIT>""" <EOL> fd = open ( self . pyc_file , '<STR_LIT:rb>' ) <EOL> self . magic = fd . read ( <NUM_LIT:4> ) <EOL> self . moddate = fd . read ( <NUM_LIT:4> ) <EOL> self . modif_date = long ( struct . unpack ( '<STR_LIT>' , self . moddate ) [ <NUM_LIT:0> ] ) <EOL> co = marshal . load ( fd ) <EOL> self . parse_code ( co ) <EOL> def parse_code ( self , co ) : <EOL> """<STR_LIT>""" <EOL> self . code = co <EOL> self . bytecode = [ ] <EOL> try : <EOL> self . load_bytecode ( self . code ) <EOL> self . build_representation ( ) <EOL> except Exception , ex : <EOL> logger . error ( "<STR_LIT>" , repr ( ex ) , exc_info = ex ) <EOL> def get_module ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . main_module <EOL> def get_bytecode ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . bytecode <EOL> @ property <EOL> def declarations ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . all_decls <EOL> @ property <EOL> def has_changes ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . main_module . has_changes <EOL> def accept ( self , visitor ) : <EOL> """<STR_LIT>""" <EOL> if not self . code : <EOL> self . parse ( ) <EOL> if isinstance ( visitor , BytecodeVisitor ) : <EOL> for i in xrange ( len ( self . bytecode ) ) : <EOL> index , lineno , op , arg , cflow_in , _ = self . bytecode [ i ] <EOL> visitor . visit ( index , op , arg = arg , lineno = lineno , cflow_in = cflow_in ) <EOL> elif isinstance ( visitor , ModuleVisitor ) : <EOL> visitor . visit ( self . main_module ) <EOL> elif isinstance ( visitor , ClassVisitor ) or isinstance ( visitor , MethodVisitor ) : <EOL> if not self . main_module : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> return <EOL> self . __depth_visitor_run ( visitor ) <EOL> def __depth_visitor_run ( self , visitor ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( "<STR_LIT>" , visitor , self . main_module ) <EOL> cache = set ( ) <EOL> is_class_visitor = isinstance ( visitor , ClassVisitor ) <EOL> stack = [ self . main_module ] <EOL> while stack : <EOL> current = stack . pop ( ) <EOL> children = current . children <EOL> logger . debug ( "<STR_LIT>" , children ) <EOL> for child in children : <EOL> if not child : <EOL> continue <EOL> if ( is_class_visitor and isinstance ( child , TypeDeclaration ) ) or ( not is_class_visitor and isinstance ( child , MethodDeclaration ) ) : <EOL> if child in cache : <EOL> continue <EOL> visitor . visit ( child ) <EOL> cache . add ( child ) <EOL> stack . insert ( <NUM_LIT:0> , child ) <EOL> def add_enter_code ( self , python_code , import_code = None ) : <EOL> """<STR_LIT>""" <EOL> from . . rewriter . simple import SimpleRewriter <EOL> rewriter = SimpleRewriter ( self . main_module ) <EOL> rewriter . insert_enter_code ( python_code , import_code ) <EOL> def add_exit_code ( self , python_code , import_code = None ) : <EOL> """<STR_LIT>""" <EOL> from . . rewriter . simple import SimpleRewriter <EOL> rewriter = SimpleRewriter ( self . main_module ) <EOL> rewriter . insert_exit_code ( python_code , import_code ) <EOL> def write ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . has_changes : <EOL> logger . debug ( "<STR_LIT>" , self . main_module . module_path ) <EOL> return <EOL> try : <EOL> new_co = self . main_module . code_object <EOL> source_file = self . main_module . module_path . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> timestamp = self . modif_date <EOL> try : <EOL> timestamp = long ( os . stat ( source_file ) . st_mtime ) <EOL> except : <EOL> pass <EOL> fd = open ( self . main_module . module_path , '<STR_LIT:wb>' ) <EOL> fd . write ( '<STR_LIT>' ) <EOL> fd . write ( struct . pack ( '<STR_LIT>' , timestamp ) ) <EOL> marshal . dump ( new_co , fd ) <EOL> fd . flush ( ) <EOL> fd . seek ( <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> fd . write ( imp . get_magic ( ) ) <EOL> fd . close ( ) <EOL> logger . debug ( "<STR_LIT>" , self . main_module . module_path ) <EOL> return True <EOL> except Exception , ex : <EOL> logger . error ( "<STR_LIT>" , str ( ex ) ) <EOL> logger . error ( "<STR_LIT>" , traceback . format_exc ( ) ) <EOL> return False <EOL> def build_representation ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . bytecode : <EOL> return <EOL> decl_counter = <NUM_LIT:0> <EOL> self . all_decls = set ( ) <EOL> decl_map = { } <EOL> class_decl_indices , method_decl_indices = BytecodeObject . find_classes_methods ( self . bytecode ) <EOL> interest_indices = class_decl_indices . union ( method_decl_indices ) <EOL> current_co = self . bytecode [ <NUM_LIT:0> ] [ <NUM_LIT:5> ] <EOL> self . main_module = ModuleDeclaration ( self . pyc_file , current_co ) <EOL> self . main_module . bytecode = self . bytecode <EOL> self . main_module . bytecode_object = self <EOL> module_lines = ( <NUM_LIT:1> , max ( [ self . bytecode [ i ] [ <NUM_LIT:1> ] for i in xrange ( len ( self . bytecode ) ) ] ) , decl_counter ) <EOL> self . main_module . lines = module_lines <EOL> BytecodeObject . parse_imports ( self . main_module , self . bytecode ) <EOL> decl_map [ module_lines ] = self . main_module <EOL> decl_counter += <NUM_LIT:1> <EOL> self . all_decls . add ( self . main_module ) <EOL> for tpl_indices in interest_indices : <EOL> decl = None <EOL> start_index , end_index = tpl_indices <EOL> index , lineno , op , arg , _ , co = self . bytecode [ start_index ] <EOL> end_lineno = max ( [ self . bytecode [ i ] [ <NUM_LIT:1> ] for i in xrange ( start_index , min ( end_index + <NUM_LIT:1> , len ( self . bytecode ) ) ) ] ) <EOL> decl_co = BytecodeObject . next_code_object ( self . bytecode , start_index ) <EOL> lines_tuple = ( lineno , end_lineno , decl_counter ) <EOL> i = start_index <EOL> if tpl_indices in class_decl_indices : <EOL> decl = TypeDeclaration ( decl_co . co_name , decl_co ) <EOL> else : <EOL> decl = MethodDeclaration ( decl_co . co_name , decl_co ) <EOL> decl . formal_params = BytecodeObject . get_formal_params ( decl_co ) <EOL> decl . bytecode = self . bytecode [ start_index : end_index ] <EOL> decl . lines = lines_tuple [ : <NUM_LIT:2> ] <EOL> self . all_decls . add ( decl ) <EOL> decl_map [ lines_tuple ] = decl <EOL> decl_counter += <NUM_LIT:1> <EOL> co_deps_graph = { <EOL> self . main_module . code_object : { } <EOL> } <EOL> i , length = <NUM_LIT:0> , len ( self . bytecode ) <EOL> while i < length : <EOL> op , arg , co = self . bytecode [ i ] [ <NUM_LIT:2> ] , self . bytecode [ i ] [ <NUM_LIT:3> ] , self . bytecode [ i ] [ <NUM_LIT:5> ] <EOL> if op == LOAD_CONST and isinstance ( arg , types . CodeType ) : <EOL> if co not in co_deps_graph : <EOL> co_deps_graph [ co ] = dict ( ) <EOL> if arg not in co_deps_graph [ co ] : <EOL> co_deps_graph [ co ] [ arg ] = dict ( ) <EOL> decl_parent = self . get_decl ( code_object = co ) <EOL> decl_child = self . get_decl ( code_object = arg ) <EOL> decl_child . parent = decl_parent <EOL> i += <NUM_LIT:1> <EOL> logger . debug ( '<STR_LIT:\n>' + BytecodeObject . build_tree ( self . main_module ) ) <EOL> def get_decl ( self , code_object = None , method_name = None , type_name = None ) : <EOL> """<STR_LIT>""" <EOL> if code_object is not None : <EOL> for decl in self . declarations : <EOL> if decl . code_object == code_object : <EOL> return decl <EOL> elif method_name is not None : <EOL> results = [ ] <EOL> for decl in self . declarations : <EOL> if isinstance ( decl , MethodDeclaration ) and decl . method_name == method_name : <EOL> results . append ( decl ) <EOL> if not results : <EOL> return None <EOL> return results [ <NUM_LIT:0> ] if len ( results ) == <NUM_LIT:1> else results <EOL> elif type_name is not None : <EOL> results = [ ] <EOL> for decl in self . declarations : <EOL> if isinstance ( decl , TypeDeclaration ) and decl . type_name == type_name : <EOL> results . append ( decl ) <EOL> if not results : <EOL> return None <EOL> return results [ <NUM_LIT:0> ] if len ( results ) == <NUM_LIT:1> else results <EOL> return None <EOL> @ staticmethod <EOL> def build_tree ( root , indent = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> buffer = '<STR_LIT>' <EOL> buffer += indent + repr ( root ) + '<STR_LIT:\n>' <EOL> for c in root . children : <EOL> buffer += BytecodeObject . build_tree ( c , indent + '<STR_LIT:U+0020>' ) <EOL> return buffer <EOL> @ staticmethod <EOL> def finalize_decl_object ( kind , acc_data ) : <EOL> pass <EOL> @ staticmethod <EOL> def next_code_object ( bytecode , index ) : <EOL> i = index <EOL> while i < len ( bytecode ) : <EOL> co = bytecode [ i ] [ <NUM_LIT:5> ] <EOL> if isinstance ( co , types . CodeType ) : <EOL> return co <EOL> i += <NUM_LIT:1> <EOL> return None <EOL> @ staticmethod <EOL> def prev_code_object ( bytecode , index ) : <EOL> i = index <EOL> while i <= <NUM_LIT:0> : <EOL> co = bytecode [ i ] [ <NUM_LIT:5> ] <EOL> if isinstance ( co , types . CodeType ) : <EOL> return co <EOL> i -= <NUM_LIT:1> <EOL> return None <EOL> @ staticmethod <EOL> def find_classes_methods ( bytecode ) : <EOL> """<STR_LIT>""" <EOL> class_indices = set ( ) <EOL> method_indices = set ( ) <EOL> class_lines = set ( ) <EOL> method_lines = set ( ) <EOL> co_lines = { } <EOL> classes_co = set ( ) <EOL> methods_co = set ( ) <EOL> i , length = <NUM_LIT:0> , len ( bytecode ) <EOL> while i < length : <EOL> tpl = bytecode [ i ] <EOL> lineno , op_code , co = tpl [ <NUM_LIT:1> ] , tpl [ <NUM_LIT:2> ] , tpl [ <NUM_LIT:5> ] <EOL> if op_code == BUILD_CLASS : <EOL> class_co = bytecode [ i - <NUM_LIT:3> ] [ <NUM_LIT:5> ] <EOL> co_lines [ class_co ] = lineno <EOL> classes_co . add ( class_co ) <EOL> class_lines . add ( lineno ) <EOL> elif op_code in ( MAKE_FUNCTION , MAKE_CLOSURE ) : <EOL> if i < length - <NUM_LIT:3> and bytecode [ i + <NUM_LIT:1> ] [ <NUM_LIT:2> ] == CALL_FUNCTION and bytecode [ i + <NUM_LIT:2> ] [ <NUM_LIT:2> ] == BUILD_CLASS : <EOL> i += <NUM_LIT:1> <EOL> continue <EOL> prev_co = bytecode [ i - <NUM_LIT:1> ] [ <NUM_LIT:5> ] <EOL> method_lines . add ( lineno ) <EOL> methods_co . add ( prev_co ) <EOL> co_lines [ prev_co ] = lineno <EOL> i += <NUM_LIT:1> <EOL> for decl_co in co_lines : <EOL> if decl_co not in methods_co and decl_co not in classes_co : <EOL> continue <EOL> dest_set = method_indices if decl_co in methods_co else class_indices <EOL> i , length = <NUM_LIT:0> , len ( bytecode ) <EOL> start_index , end_index = <NUM_LIT:0> , <NUM_LIT:0> <EOL> while i < length : <EOL> co = bytecode [ i ] [ <NUM_LIT:5> ] <EOL> if co == decl_co : <EOL> start_index = i <EOL> j = i + <NUM_LIT:1> <EOL> while j < length : <EOL> if bytecode [ j ] [ <NUM_LIT:5> ] == decl_co : <EOL> end_index = j <EOL> j += <NUM_LIT:1> <EOL> break <EOL> i += <NUM_LIT:1> <EOL> dest_set . add ( ( start_index , end_index ) ) <EOL> return class_indices , method_indices <EOL> @ staticmethod <EOL> def get_formal_params ( code_object ) : <EOL> """<STR_LIT>""" <EOL> varargs = bool ( code_object . co_flags & CO_VARARGS ) <EOL> varkwargs = bool ( code_object . co_flags & CO_VARKEYWORDS ) <EOL> args = code_object . co_varnames [ : code_object . co_argcount + varargs + varkwargs ] <EOL> return args <EOL> def load_bytecode ( self , code_object ) : <EOL> BytecodeObject . parse_code_object ( code_object , self . bytecode ) <EOL> @ staticmethod <EOL> def get_parsed_code ( code_object ) : <EOL> bytecode = [ ] <EOL> BytecodeObject . parse_code_object ( code_object , bytecode ) <EOL> return bytecode <EOL> @ staticmethod <EOL> def parse_code_object ( code_object , bytecode ) : <EOL> """<STR_LIT>""" <EOL> if not code_object : <EOL> return <EOL> code = code_object . co_code <EOL> labels = findlabels ( code_object . co_code ) <EOL> linestarts = dict ( findlinestarts ( code_object ) ) <EOL> global_free = code_object . co_cellvars + code_object . co_freevars <EOL> newlocals = bool ( code_object . co_flags & CO_NEWLOCALS ) <EOL> length = len ( code ) <EOL> i = <NUM_LIT:0> <EOL> lineno = - <NUM_LIT:1> <EOL> extended_arg = <NUM_LIT:0> <EOL> while i < length : <EOL> c = code [ i ] <EOL> op = ord ( c ) <EOL> if i in linestarts : <EOL> lineno = linestarts [ i ] <EOL> arg1 = None <EOL> cflow_in = i in labels <EOL> current_index = i <EOL> i += <NUM_LIT:1> <EOL> if op >= opcode . HAVE_ARGUMENT : <EOL> oparg = ord ( code [ i ] ) + ( ord ( code [ i + <NUM_LIT:1> ] ) << <NUM_LIT:8> ) + extended_arg <EOL> i += <NUM_LIT:2> <EOL> label = - <NUM_LIT:1> <EOL> extended_arg = <NUM_LIT:0> <EOL> if op == opcode . EXTENDED_ARG : <EOL> extended_arg = ( oparg << <NUM_LIT:16> ) <EOL> if op in opcode . hasconst : <EOL> arg1 = code_object . co_consts [ oparg ] <EOL> elif op in opcode . hasname : <EOL> arg1 = code_object . co_names [ oparg ] <EOL> elif op in opcode . haslocal : <EOL> arg1 = code_object . co_varnames [ oparg ] <EOL> elif op in opcode . hascompare : <EOL> arg1 = opcode . cmp_op [ oparg ] <EOL> elif op in opcode . hasfree : <EOL> arg1 = global_free [ oparg ] <EOL> elif op in opcode . hasjrel : <EOL> arg1 = oparg <EOL> elif op in opcode . hasjabs : <EOL> arg1 = oparg <EOL> else : <EOL> arg1 = oparg <EOL> bytecode . append ( ( current_index , lineno , op , arg1 , cflow_in , code_object ) ) <EOL> if arg1 and isinstance ( arg1 , types . CodeType ) : <EOL> BytecodeObject . parse_code_object ( arg1 , bytecode ) <EOL> @ staticmethod <EOL> def parse_imports ( main_module , bytecode ) : <EOL> """<STR_LIT>""" <EOL> import_stmts = BytecodeObject . get_imports_from_bytecode ( main_module . code_object , <EOL> bytecode ) <EOL> for imp_stmt in import_stmts : <EOL> main_module . add_import ( imp_stmt ) <EOL> @ staticmethod <EOL> def get_imports_from_bytecode ( code_object , bytecode ) : <EOL> """<STR_LIT>""" <EOL> last_import_ref_idx = BytecodeObject . get_last_import_ref ( bytecode , <EOL> code_object ) <EOL> imports_bc = bytecode [ : last_import_ref_idx + <NUM_LIT:1> ] <EOL> import_stmts = [ ] <EOL> final_import_stmts = [ ] <EOL> started = False <EOL> buffer = [ ] <EOL> for tpl in imports_bc : <EOL> if tpl [ <NUM_LIT:2> ] == LOAD_CONST and isinstance ( tpl [ <NUM_LIT:3> ] , int ) : <EOL> started = True <EOL> if len ( buffer ) > <NUM_LIT:0> : <EOL> import_stmts . append ( buffer ) <EOL> buffer = [ ] <EOL> if started : <EOL> buffer . append ( tpl ) <EOL> if len ( buffer ) > <NUM_LIT:0> : <EOL> import_stmts . append ( buffer ) <EOL> logger . debug ( "<STR_LIT>" , len ( import_stmts ) ) <EOL> for import_stmt in import_stmts : <EOL> arg = import_stmt [ <NUM_LIT:0> ] [ <NUM_LIT:3> ] <EOL> import_star = False <EOL> import_root = None <EOL> aliased_names = [ ] <EOL> num_dots = arg if arg >= <NUM_LIT:0> else <NUM_LIT:0> <EOL> i , length = <NUM_LIT:0> , len ( import_stmt ) <EOL> while i < length : <EOL> op , arg = import_stmt [ i ] [ <NUM_LIT:2> ] , import_stmt [ i ] [ <NUM_LIT:3> ] <EOL> next_op = import_stmt [ i + <NUM_LIT:1> ] [ <NUM_LIT:2> ] if i < length - <NUM_LIT:1> else - <NUM_LIT:1> <EOL> next_next_op = import_stmt [ i + <NUM_LIT:2> ] [ <NUM_LIT:2> ] if i < length - <NUM_LIT:2> else - <NUM_LIT:1> <EOL> if op in ( IMPORT_NAME , IMPORT_FROM ) and next_op in ( STORE_NAME , STORE_FAST ) : <EOL> next_arg = import_stmt [ i + <NUM_LIT:1> ] [ <NUM_LIT:3> ] <EOL> if next_arg == arg or arg . startswith ( next_arg + '<STR_LIT:.>' ) : <EOL> aliased_names . append ( ( arg , None ) ) <EOL> else : <EOL> aliased_names . append ( ( arg , next_arg ) ) <EOL> elif op == IMPORT_NAME and next_op == LOAD_ATTR and next_next_op == STORE_NAME : <EOL> next_arg = import_stmt [ i + <NUM_LIT:2> ] [ <NUM_LIT:3> ] <EOL> aliased_names . append ( ( arg , next_arg ) ) <EOL> elif op == IMPORT_NAME : <EOL> if import_root is not None : <EOL> logger . error ( "<STR_LIT>" , import_root , arg ) <EOL> if arg : <EOL> import_root = arg <EOL> if next_op == IMPORT_STAR : <EOL> import_star = True <EOL> i += <NUM_LIT:1> <EOL> imp_stmt = ImportDeclaration ( code_object ) <EOL> imp_stmt . root = import_root <EOL> imp_stmt . star = import_star <EOL> imp_stmt . dots = num_dots <EOL> imp_stmt . aliases = aliased_names <EOL> final_import_stmts . append ( imp_stmt ) <EOL> return final_import_stmts <EOL> @ staticmethod <EOL> def get_last_import_ref ( bytecode , code_object ) : <EOL> """<STR_LIT>""" <EOL> i , length = <NUM_LIT:0> , len ( bytecode ) <EOL> last_index = - <NUM_LIT:1> <EOL> while i < length : <EOL> op , co = bytecode [ i ] [ <NUM_LIT:2> ] , bytecode [ i ] [ <NUM_LIT:5> ] <EOL> if op in ( IMPORT_FROM , IMPORT_NAME ) and co == code_object : <EOL> last_index = i <EOL> i += <NUM_LIT:1> <EOL> if last_index > - <NUM_LIT:1> : <EOL> while last_index < length : <EOL> op = bytecode [ last_index ] [ <NUM_LIT:2> ] <EOL> if op == STORE_NAME : <EOL> break <EOL> last_index += <NUM_LIT:1> <EOL> return last_index </s>
