<s> __author__ = '<STR_LIT>' <EOL> import struct <EOL> from scalyr_agent . json_lib import JsonArray , JsonObject , JsonParseException <EOL> class ByteScanner ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , string_input , start_pos = <NUM_LIT:0> , max_pos = None ) : <EOL> """<STR_LIT>""" <EOL> self . __buffer = [ ] <EOL> self . __pos = start_pos <EOL> self . __start_pos = start_pos <EOL> self . __max_pos = len ( string_input ) <EOL> if not max_pos is None : <EOL> self . __max_pos = max_pos <EOL> for i in range ( len ( string_input ) ) : <EOL> self . __buffer . append ( "<STR_LIT>" + chr ( ord ( string_input [ i ] ) & <NUM_LIT:255> ) ) <EOL> @ property <EOL> def at_end ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __pos >= self . __max_pos <EOL> @ property <EOL> def position ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __pos <EOL> @ property <EOL> def line_number ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . line_number_for_offset ( <NUM_LIT:0> ) <EOL> def line_number_for_offset ( self , offset ) : <EOL> """<STR_LIT>""" <EOL> line_num = <NUM_LIT:1> <EOL> target_pos = self . __pos + offset <EOL> x = self . __start_pos <EOL> self . __check_index ( target_pos ) <EOL> while x < target_pos : <EOL> b = self . __buffer [ x ] <EOL> x += <NUM_LIT:1> <EOL> if b == '<STR_LIT:\n>' : <EOL> line_num += <NUM_LIT:1> <EOL> elif b == '<STR_LIT:\r>' : <EOL> line_num += <NUM_LIT:1> <EOL> if x < target_pos and self . __buffer [ x ] == '<STR_LIT:\n>' : <EOL> x += <NUM_LIT:1> <EOL> return line_num <EOL> @ property <EOL> def bytes_remaining ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __max_pos - self . __pos <EOL> def read_ubyte ( self ) : <EOL> """<STR_LIT>""" <EOL> self . __check_read_size ( <NUM_LIT:1> ) <EOL> self . __pos += <NUM_LIT:1> <EOL> return self . __buffer [ self . __pos - <NUM_LIT:1> ] <EOL> def read_ubytes ( self , length ) : <EOL> """<STR_LIT>""" <EOL> self . __check_read_size ( length ) <EOL> my_slice = self . __buffer [ self . __pos : self . __pos + length ] <EOL> self . __pos += length <EOL> return "<STR_LIT>" . join ( my_slice ) <EOL> def peek_next_ubyte ( self , offset = <NUM_LIT:0> , none_if_bad_index = False ) : <EOL> """<STR_LIT>""" <EOL> index = self . __pos + offset <EOL> if not none_if_bad_index : <EOL> self . __check_index ( index ) <EOL> elif index < self . __start_pos or index >= self . __max_pos : <EOL> return None <EOL> return self . __buffer [ index ] <EOL> def __check_read_size ( self , read_length ) : <EOL> index = self . __pos + read_length - <NUM_LIT:1> <EOL> if index < self . __start_pos or index >= self . __max_pos : <EOL> raise IndexError ( <EOL> "<STR_LIT>" <EOL> % ( self . __pos , self . __max_pos , read_length ) ) <EOL> def __check_index ( self , index ) : <EOL> if index < self . __start_pos or index >= self . __max_pos : <EOL> raise IndexError ( <EOL> "<STR_LIT>" % <EOL> ( index , self . __start_pos , self . __max_pos ) ) <EOL> class JsonParser ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , input_scanner , allow_missing_commas = True , check_duplicate_keys = False ) : <EOL> """<STR_LIT>""" <EOL> self . __scanner = input_scanner <EOL> self . allow_missing_commas = allow_missing_commas <EOL> self . check_duplicate_keys = check_duplicate_keys <EOL> @ staticmethod <EOL> def parse ( input_bytes , check_duplicate_keys = False ) : <EOL> return JsonParser ( ByteScanner ( input_bytes ) , True , check_duplicate_keys ) . parse_value ( ) <EOL> def parse_value ( self ) : <EOL> """<STR_LIT>""" <EOL> start_pos = self . __scanner . position <EOL> try : <EOL> c = self . __peek_next_non_whitespace ( ) <EOL> if c == '<STR_LIT:{>' : <EOL> return self . __parse_object ( ) <EOL> elif c == '<STR_LIT:[>' : <EOL> return self . __parse_array ( ) <EOL> elif c == '<STR_LIT:">' : <EOL> return self . __parse_string_with_concatenation ( ) <EOL> elif c == '<STR_LIT:t>' : <EOL> self . __match ( "<STR_LIT:true>" , "<STR_LIT>" ) <EOL> return True <EOL> elif c == '<STR_LIT:f>' : <EOL> self . __match ( "<STR_LIT:false>" , "<STR_LIT>" ) <EOL> return False <EOL> elif c == '<STR_LIT:n>' : <EOL> self . __match ( "<STR_LIT:null>" , "<STR_LIT>" ) <EOL> return None <EOL> elif c == '<STR_LIT:->' or '<STR_LIT:0>' <= c <= '<STR_LIT>' : <EOL> return self . __parse_number ( ) <EOL> elif c == '<STR_LIT:}>' : <EOL> return self . __error ( "<STR_LIT>" ) <EOL> elif c == '<STR_LIT>' : <EOL> return self . __parse_length_prefixed_string ( ) <EOL> else : <EOL> if c is None : <EOL> if start_pos == <NUM_LIT:0> : <EOL> return self . __error ( "<STR_LIT>" ) <EOL> else : <EOL> return self . __error ( "<STR_LIT>" ) <EOL> else : <EOL> return self . __error ( "<STR_LIT>" % c ) <EOL> except IndexError : <EOL> raise JsonParseException ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , start_pos , <EOL> self . __scanner . line_number_for_offset ( start_pos ) ) <EOL> def __parse_object ( self ) : <EOL> """<STR_LIT>""" <EOL> object_start = self . __scanner . position <EOL> self . __scanner . read_ubyte ( ) <EOL> result_object = JsonObject ( ) <EOL> while True : <EOL> c = self . __peek_next_non_whitespace ( ) <EOL> if c is None : <EOL> return self . __error ( "<STR_LIT>" , object_start ) <EOL> elif c == '<STR_LIT:">' : <EOL> key = self . __parse_string ( ) <EOL> elif c == '<STR_LIT:_>' or '<STR_LIT:a>' <= c <= '<STR_LIT:z>' or '<STR_LIT:A>' <= c <= '<STR_LIT>' : <EOL> key = self . __parse_identifier ( ) <EOL> next_char = self . __scanner . peek_next_ubyte ( <EOL> none_if_bad_index = True ) <EOL> if next_char is None : <EOL> return self . __error ( "<STR_LIT>" , <EOL> object_start ) <EOL> if ord ( next_char ) > <NUM_LIT:32> and next_char != '<STR_LIT::>' : <EOL> self . __error ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % next_char ) <EOL> elif c == '<STR_LIT:}>' : <EOL> self . __scanner . read_ubyte ( ) <EOL> return result_object <EOL> else : <EOL> return self . __error ( <EOL> "<STR_LIT>" ) <EOL> self . __peek_next_non_whitespace ( ) <EOL> c = self . __scanner . read_ubyte ( ) <EOL> if c != '<STR_LIT::>' : <EOL> self . __error ( "<STR_LIT>" ) <EOL> self . __peek_next_non_whitespace ( ) <EOL> if self . check_duplicate_keys and result_object . __contains__ ( key ) : <EOL> self . __error ( "<STR_LIT>" + key + "<STR_LIT:]>" , object_start ) <EOL> result_object . put ( key , self . parse_value ( ) ) <EOL> c = self . __peek_next_non_whitespace ( ) <EOL> if c is None : <EOL> self . __error ( "<STR_LIT>" , object_start ) <EOL> elif c == '<STR_LIT:}>' : <EOL> continue <EOL> elif c == '<STR_LIT:U+002C>' : <EOL> self . __scanner . read_ubyte ( ) <EOL> else : <EOL> if self . __preceding_line_break ( ) and self . allow_missing_commas : <EOL> continue <EOL> else : <EOL> self . __error ( "<STR_LIT>" <EOL> "<STR_LIT>" % c ) <EOL> def __parse_array ( self ) : <EOL> """<STR_LIT>""" <EOL> array_start = self . __scanner . position <EOL> self . __scanner . read_ubyte ( ) <EOL> array = JsonArray ( ) <EOL> while True : <EOL> if self . __peek_next_non_whitespace ( ) == '<STR_LIT:]>' : <EOL> self . __scanner . read_ubyte ( ) <EOL> return array <EOL> self . __peek_next_non_whitespace ( ) <EOL> array . add ( self . parse_value ( ) ) <EOL> c = self . __peek_next_non_whitespace ( ) <EOL> if c is None : <EOL> self . __error ( "<STR_LIT>" , array_start ) <EOL> elif c == '<STR_LIT:]>' : <EOL> continue <EOL> elif c == '<STR_LIT:U+002C>' : <EOL> self . __scanner . read_ubyte ( ) <EOL> else : <EOL> if self . __preceding_line_break ( ) and self . allow_missing_commas : <EOL> continue <EOL> else : <EOL> self . __error ( "<STR_LIT>" <EOL> "<STR_LIT>" % c ) <EOL> def __parse_length_prefixed_string ( self ) : <EOL> """<STR_LIT>""" <EOL> self . __scanner . read_ubyte ( ) <EOL> if self . __scanner . peek_next_ubyte ( none_if_bad_index = True ) != '<STR_LIT:s>' : <EOL> self . __error ( "<STR_LIT>" ) <EOL> self . __scanner . read_ubyte ( ) <EOL> encoded_num_bytes = self . __scanner . read_ubytes ( <NUM_LIT:4> ) <EOL> num_bytes = struct . unpack ( '<STR_LIT>' , encoded_num_bytes ) [ <NUM_LIT:0> ] <EOL> return self . __scanner . read_ubytes ( num_bytes ) <EOL> def __parse_string_with_concatenation ( self ) : <EOL> """<STR_LIT>""" <EOL> value = self . __parse_string ( ) <EOL> c = self . __peek_next_non_whitespace ( ) <EOL> if c != '<STR_LIT:+>' : <EOL> return value <EOL> all_strings = [ value ] <EOL> while True : <EOL> assert ( self . __scanner . read_ubyte ( ) == '<STR_LIT:+>' ) , "<STR_LIT>" <EOL> c = self . __peek_next_non_whitespace ( ) <EOL> if c != '<STR_LIT:">' : <EOL> self . __error ( "<STR_LIT>" ) <EOL> all_strings . append ( self . __parse_string ( ) ) <EOL> if self . __peek_next_non_whitespace ( ) != '<STR_LIT:+>' : <EOL> break <EOL> return '<STR_LIT>' . join ( all_strings ) <EOL> def __parse_identifier ( self ) : <EOL> """<STR_LIT>""" <EOL> length = <NUM_LIT:0> <EOL> while True : <EOL> c = self . __scanner . peek_next_ubyte ( offset = length , none_if_bad_index = True ) <EOL> if c == '<STR_LIT:_>' or '<STR_LIT:a>' <= c <= '<STR_LIT:z>' or '<STR_LIT:A>' <= c <= '<STR_LIT>' or '<STR_LIT:0>' <= c <= '<STR_LIT>' : <EOL> length += <NUM_LIT:1> <EOL> else : <EOL> break <EOL> return self . __scanner . read_ubytes ( length ) <EOL> def __parse_string ( self ) : <EOL> """<STR_LIT>""" <EOL> start_pos = self . __scanner . position <EOL> c = self . __scanner . read_ubyte ( ) <EOL> if c != '<STR_LIT:">' : <EOL> return None <EOL> length = <NUM_LIT:0> <EOL> while True : <EOL> if length >= self . __scanner . bytes_remaining : <EOL> self . __error ( "<STR_LIT>" ) <EOL> c = self . __scanner . peek_next_ubyte ( offset = length ) <EOL> if c == '<STR_LIT:">' : <EOL> break <EOL> length += <NUM_LIT:1> <EOL> if c == '<STR_LIT:\\>' : <EOL> if length >= self . __scanner . bytes_remaining : <EOL> self . __error ( "<STR_LIT>" ) <EOL> length += <NUM_LIT:1> <EOL> elif c == '<STR_LIT:\r>' or c == '<STR_LIT:\n>' : <EOL> self . __error ( "<STR_LIT>" ) <EOL> result = self . __scanner . read_ubytes ( length ) <EOL> self . __scanner . read_ubyte ( ) <EOL> return self . __process_escapes ( result . decode ( "<STR_LIT:utf8>" ) , start_pos + <NUM_LIT:1> ) <EOL> def __process_escapes ( self , s , literal_start ) : <EOL> """<STR_LIT>""" <EOL> if s . find ( '<STR_LIT:\\>' ) < <NUM_LIT:0> : <EOL> return s <EOL> slen = len ( s ) <EOL> sb = [ ] <EOL> i = <NUM_LIT:0> <EOL> while i < slen : <EOL> c = s [ i ] <EOL> if c != '<STR_LIT:\\>' : <EOL> sb . append ( c ) <EOL> i += <NUM_LIT:1> <EOL> continue <EOL> if i + <NUM_LIT:1> >= slen : <EOL> self . __error ( "<STR_LIT>" , literal_start + i ) <EOL> i += <NUM_LIT:1> <EOL> c = s [ i ] <EOL> if c == '<STR_LIT:t>' : <EOL> sb . append ( '<STR_LIT:\t>' ) <EOL> elif c == '<STR_LIT:n>' : <EOL> sb . append ( '<STR_LIT:\n>' ) <EOL> elif c == '<STR_LIT:r>' : <EOL> sb . append ( '<STR_LIT:\r>' ) <EOL> elif c == '<STR_LIT:b>' : <EOL> sb . append ( '<STR_LIT>' ) <EOL> elif c == '<STR_LIT:f>' : <EOL> sb . append ( '<STR_LIT>' ) <EOL> elif c == '<STR_LIT:">' : <EOL> sb . append ( '<STR_LIT:">' ) <EOL> elif c == '<STR_LIT:\\>' : <EOL> sb . append ( '<STR_LIT:\\>' ) <EOL> elif c == '<STR_LIT:/>' : <EOL> sb . append ( '<STR_LIT:/>' ) <EOL> elif c == '<STR_LIT:u>' and i + <NUM_LIT:5> <= slen : <EOL> hex_string = s [ i + <NUM_LIT:1> : i + <NUM_LIT:5> ] <EOL> i += <NUM_LIT:4> <EOL> sb . append ( unichr ( int ( hex_string , <NUM_LIT:16> ) ) ) <EOL> else : <EOL> self . __error ( "<STR_LIT>" + c + "<STR_LIT:]>" , <EOL> literal_start + i ) <EOL> i += <NUM_LIT:1> <EOL> return "<STR_LIT>" . join ( sb ) <EOL> def __parse_number ( self ) : <EOL> """<STR_LIT>""" <EOL> literal_buffer = [ ] <EOL> all_digits = True <EOL> while not self . __scanner . at_end : <EOL> peek = self . __scanner . peek_next_ubyte ( ) <EOL> if ( ( peek != '<STR_LIT:+>' ) and ( peek != '<STR_LIT:->' ) and ( peek != '<STR_LIT:e>' ) and <EOL> ( peek != '<STR_LIT:E>' ) and ( peek != '<STR_LIT:.>' ) and <EOL> ( not '<STR_LIT:0>' <= peek <= '<STR_LIT>' ) ) : <EOL> break <EOL> if len ( literal_buffer ) >= <NUM_LIT:100> : <EOL> self . __error ( "<STR_LIT>" ) <EOL> next_char = self . __scanner . read_ubyte ( ) <EOL> all_digits = all_digits and '<STR_LIT:0>' <= next_char <= '<STR_LIT>' <EOL> literal_buffer . append ( next_char ) <EOL> if all_digits and len ( literal_buffer ) <= <NUM_LIT> : <EOL> value = <NUM_LIT:0> <EOL> for digit in literal_buffer : <EOL> value = ( value * <NUM_LIT:10> ) + ord ( digit ) - ord ( '<STR_LIT:0>' ) <EOL> return value <EOL> number_string = '<STR_LIT>' . join ( literal_buffer ) <EOL> if ( number_string . find ( '<STR_LIT:.>' ) < <NUM_LIT:0> and <EOL> number_string . find ( '<STR_LIT:e>' ) < <NUM_LIT:0> and <EOL> number_string . find ( '<STR_LIT:E>' ) < <NUM_LIT:0> ) : <EOL> try : <EOL> return long ( number_string ) <EOL> except ValueError : <EOL> self . __error ( <EOL> "<STR_LIT>" % number_string ) <EOL> else : <EOL> try : <EOL> return float ( number_string ) <EOL> except ValueError : <EOL> self . __error ( <EOL> "<STR_LIT>" % number_string ) <EOL> def __parse_comment ( self ) : <EOL> """<STR_LIT>""" <EOL> comment_start_pos = self . __scanner . position <EOL> self . __scanner . read_ubyte ( ) <EOL> if self . __scanner . at_end : <EOL> self . __error ( "<STR_LIT>" ) <EOL> c = self . __scanner . read_ubyte ( ) <EOL> if c == '<STR_LIT:/>' : <EOL> while not self . __scanner . at_end : <EOL> c = self . __scanner . read_ubyte ( ) <EOL> if c == '<STR_LIT:\n>' or c == '<STR_LIT:\r>' : <EOL> break <EOL> if c == '<STR_LIT:\r>' and self . __scanner . peek_next_ubyte ( none_if_bad_index = True ) == '<STR_LIT:\n>' : <EOL> self . __scanner . read_ubyte ( ) <EOL> elif c == '<STR_LIT:*>' : <EOL> while not self . __scanner . at_end : <EOL> c = self . __scanner . read_ubyte ( ) <EOL> if ( c == '<STR_LIT:*>' and <EOL> self . __scanner . peek_next_ubyte ( <EOL> none_if_bad_index = True ) == '<STR_LIT:/>' ) : <EOL> self . __scanner . read_ubyte ( ) <EOL> return <EOL> self . __error ( "<STR_LIT>" , comment_start_pos ) <EOL> else : <EOL> self . __error ( "<STR_LIT>" % c ) <EOL> def __match ( self , chars , error_message ) : <EOL> """<STR_LIT>""" <EOL> start_pos = self . __scanner . position <EOL> for i in range ( <NUM_LIT:0> , len ( chars ) ) : <EOL> expected = chars [ i ] <EOL> if self . __scanner . at_end : <EOL> actual = - <NUM_LIT:1> <EOL> else : <EOL> actual = self . __scanner . read_ubyte ( ) <EOL> if expected != actual : <EOL> if error_message is not None : <EOL> self . __error ( error_message , start_pos ) <EOL> else : <EOL> self . __error ( "<STR_LIT>" % chars , start_pos ) <EOL> def __error ( self , error_message , position = None ) : <EOL> """<STR_LIT>""" <EOL> if position is None : <EOL> position = max ( <NUM_LIT:0> , self . __scanner . position - <NUM_LIT:1> ) <EOL> raise JsonParseException ( <EOL> error_message , position , <EOL> self . __scanner . line_number_for_offset ( <EOL> position - self . __scanner . position ) ) <EOL> def __preceding_line_break ( self ) : <EOL> i = - <NUM_LIT:1> <EOL> b = self . __scanner . peek_next_ubyte ( offset = i , none_if_bad_index = True ) <EOL> while not b is None : <EOL> if b == '<STR_LIT:\r>' or b == '<STR_LIT:\n>' : <EOL> return True <EOL> elif b == '<STR_LIT:U+0020>' or b == '<STR_LIT:\t>' : <EOL> i += - <NUM_LIT:1> <EOL> b = self . __scanner . peek_next_ubyte ( offset = i , none_if_bad_index = True ) <EOL> else : <EOL> return False <EOL> return False <EOL> def __peek_next_non_whitespace ( self ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> raw_c = self . __scanner . peek_next_ubyte ( none_if_bad_index = True ) <EOL> if raw_c is None : <EOL> return raw_c <EOL> c = ord ( raw_c ) <EOL> if c == <NUM_LIT:32> or c == <NUM_LIT:9> or c == <NUM_LIT> or c == <NUM_LIT:10> : <EOL> self . __scanner . read_ubyte ( ) <EOL> continue <EOL> elif raw_c == '<STR_LIT:/>' : <EOL> self . __parse_comment ( ) <EOL> continue <EOL> return raw_c <EOL> def parse ( input_bytes , check_duplicate_keys = False ) : <EOL> """<STR_LIT>""" <EOL> return JsonParser . parse ( input_bytes , check_duplicate_keys ) </s>
