<s> import csv <EOL> import logging <EOL> import os <EOL> import pytz <EOL> import re <EOL> import shutil <EOL> import StringIO <EOL> import tempfile <EOL> import uuid <EOL> from dateutil . parser import parse <EOL> from django . conf import settings <EOL> from django . utils . translation import ugettext as _ <EOL> from desktop . lib . i18n import force_unicode , smart_str <EOL> from indexer import conf <EOL> from indexer . models import DATE_FIELD_TYPES , TEXT_FIELD_TYPES , INTEGER_FIELD_TYPES , DECIMAL_FIELD_TYPES , BOOLEAN_FIELD_TYPES <EOL> LOG = logging . getLogger ( __name__ ) <EOL> TIMESTAMP_PATTERN = '<STR_LIT>' <EOL> FIELD_XML_TEMPLATE = '<STR_LIT>' <EOL> DEFAULT_FIELD = { <EOL> '<STR_LIT:name>' : None , <EOL> '<STR_LIT:type>' : '<STR_LIT:text>' , <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : '<STR_LIT:true>' <EOL> } <EOL> def get_config_template_path ( solr_cloud_mode ) : <EOL> if solr_cloud_mode : <EOL> return os . path . join ( conf . CONFIG_TEMPLATE_PATH . get ( ) , '<STR_LIT>' ) <EOL> else : <EOL> return os . path . join ( conf . CONFIG_TEMPLATE_PATH . get ( ) , '<STR_LIT>' ) <EOL> class SchemaXml ( object ) : <EOL> def __init__ ( self , xml ) : <EOL> self . xml = xml <EOL> def uniqueKeyField ( self , unique_key_field ) : <EOL> self . xml = force_unicode ( force_unicode ( self . xml ) . replace ( u'<STR_LIT>' , force_unicode ( unique_key_field ) ) ) <EOL> def fields ( self , fields ) : <EOL> fields_xml = '<STR_LIT>' <EOL> for field in fields : <EOL> field_dict = DEFAULT_FIELD . copy ( ) <EOL> field_dict . update ( field ) <EOL> fields_xml += FIELD_XML_TEMPLATE % field_dict + '<STR_LIT:\n>' <EOL> self . xml = force_unicode ( force_unicode ( self . xml ) . replace ( u'<STR_LIT>' , force_unicode ( fields_xml ) ) ) <EOL> class SolrConfigXml ( object ) : <EOL> def __init__ ( self , xml ) : <EOL> self . xml = xml <EOL> def defaultField ( self , df ) : <EOL> self . xml = force_unicode ( force_unicode ( self . xml ) . replace ( u'<STR_LIT>' , u'<STR_LIT>' % force_unicode ( df ) ) ) <EOL> def copy_configs ( fields , unique_key_field , df , solr_cloud_mode = True ) : <EOL> tmp_path = tempfile . mkdtemp ( ) <EOL> try : <EOL> config_template_path = get_config_template_path ( solr_cloud_mode ) <EOL> solr_config_path = os . path . join ( tmp_path , '<STR_LIT>' ) <EOL> shutil . copytree ( config_template_path , solr_config_path ) <EOL> if fields or unique_key_field : <EOL> with open ( os . path . join ( config_template_path , '<STR_LIT>' ) ) as f : <EOL> schemaxml = SchemaXml ( f . read ( ) ) <EOL> schemaxml . uniqueKeyField ( unique_key_field ) <EOL> schemaxml . fields ( fields ) <EOL> with open ( os . path . join ( solr_config_path , '<STR_LIT>' ) , '<STR_LIT:w>' ) as f : <EOL> f . write ( smart_str ( schemaxml . xml ) ) <EOL> if df : <EOL> with open ( os . path . join ( config_template_path , '<STR_LIT>' ) ) as f : <EOL> solrconfigxml = SolrConfigXml ( f . read ( ) ) <EOL> solrconfigxml . defaultField ( df ) <EOL> with open ( os . path . join ( solr_config_path , '<STR_LIT>' ) , '<STR_LIT:w>' ) as f : <EOL> f . write ( smart_str ( solrconfigxml . xml ) ) <EOL> return tmp_path , solr_config_path <EOL> except Exception : <EOL> shutil . rmtree ( tmp_path ) <EOL> raise <EOL> def get_field_types ( field_list , iterations = <NUM_LIT:3> ) : <EOL> assert iterations > <NUM_LIT:0> , "<STR_LIT>" <EOL> def test_boolean ( value ) : <EOL> if value . lower ( ) not in ( '<STR_LIT:false>' , '<STR_LIT:true>' ) : <EOL> raise ValueError ( _ ( "<STR_LIT>" ) % value ) <EOL> def test_timestamp ( value ) : <EOL> if not value : <EOL> raise ValueError ( ) <EOL> if len ( value ) > <NUM_LIT:50> or len ( value ) < <NUM_LIT:3> : <EOL> raise ValueError ( ) <EOL> if value . startswith ( '<STR_LIT:[>' ) and value . endswith ( '<STR_LIT:]>' ) : <EOL> value = value [ <NUM_LIT:1> : - <NUM_LIT:1> ] <EOL> try : <EOL> parse ( value ) <EOL> except OverflowError : <EOL> raise ValueError ( ) <EOL> def test_int ( value ) : <EOL> if len ( bin ( int ( value ) ) ) - <NUM_LIT:2> > <NUM_LIT:32> : <EOL> raise ValueError ( ) <EOL> def test_string ( value ) : <EOL> if len ( smart_str ( value ) . split ( '<STR_LIT:U+0020>' ) ) > <NUM_LIT:4> : <EOL> raise ValueError ( ) <EOL> test_fns = [ ( '<STR_LIT>' , test_boolean ) , <EOL> ( '<STR_LIT>' , test_int ) , <EOL> ( '<STR_LIT>' , int ) , <EOL> ( '<STR_LIT>' , float ) , <EOL> ( '<STR_LIT>' , test_timestamp ) , <EOL> ( '<STR_LIT:string>' , test_string ) , <EOL> ( '<STR_LIT>' , any ) ] <EOL> all_field_types = [ ] <EOL> for row in field_list : <EOL> if iterations == <NUM_LIT:0> : <EOL> break <EOL> iterations -= <NUM_LIT:1> <EOL> row_field_types = [ ] <EOL> for field in row : <EOL> field_type_index = None <EOL> for index in range ( <NUM_LIT:0> , len ( test_fns ) ) : <EOL> try : <EOL> test_fns [ index ] [ <NUM_LIT:1> ] ( field ) <EOL> field_type_index = index <EOL> break <EOL> except ValueError : <EOL> pass <EOL> row_field_types . append ( field_type_index ) <EOL> all_field_types . append ( row_field_types ) <EOL> if not all_field_types : <EOL> return [ ] <EOL> final_field_types = all_field_types [ <NUM_LIT:0> ] <EOL> for row_field_types in all_field_types : <EOL> for index in range ( <NUM_LIT:0> , len ( row_field_types ) ) : <EOL> if row_field_types [ index ] > final_field_types [ index ] : <EOL> final_field_types [ index ] = row_field_types [ index ] <EOL> return [ test_fns [ index ] [ <NUM_LIT:0> ] for index in final_field_types ] <EOL> def get_type_from_morphline_type ( morphline_type ) : <EOL> if morphline_type in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> return '<STR_LIT>' <EOL> else : <EOL> return '<STR_LIT:string>' <EOL> def field_values_from_separated_file ( fh , delimiter , quote_character , fields = None ) : <EOL> if fields is None : <EOL> field_names = None <EOL> else : <EOL> field_names = [ field [ '<STR_LIT:name>' ] . strip ( ) for field in fields ] <EOL> if fields is None : <EOL> timestamp_fields = None <EOL> else : <EOL> timestamp_fields = [ field [ '<STR_LIT:name>' ] . strip ( ) for field in fields if field [ '<STR_LIT:type>' ] in DATE_FIELD_TYPES ] <EOL> if fields is None : <EOL> integer_fields = None <EOL> else : <EOL> integer_fields = [ field [ '<STR_LIT:name>' ] . strip ( ) for field in fields if field [ '<STR_LIT:type>' ] in INTEGER_FIELD_TYPES ] <EOL> if fields is None : <EOL> decimal_fields = None <EOL> else : <EOL> decimal_fields = [ field [ '<STR_LIT:name>' ] . strip ( ) for field in fields if field [ '<STR_LIT:type>' ] in DECIMAL_FIELD_TYPES ] <EOL> if fields is None : <EOL> boolean_fields = None <EOL> else : <EOL> boolean_fields = [ field [ '<STR_LIT:name>' ] . strip ( ) for field in fields if field [ '<STR_LIT:type>' ] in BOOLEAN_FIELD_TYPES ] <EOL> content = fh . read ( ) <EOL> headers = None <EOL> while content : <EOL> last_newline = content . rfind ( '<STR_LIT:\n>' ) <EOL> if last_newline > - <NUM_LIT:1> : <EOL> next_chunk = fh . read ( ) <EOL> if content [ last_newline - <NUM_LIT:1> ] == '<STR_LIT:">' and next_chunk : <EOL> content += next_chunk <EOL> continue <EOL> else : <EOL> if headers is None : <EOL> csvfile = StringIO . StringIO ( content [ : last_newline ] ) <EOL> else : <EOL> csvfile = StringIO . StringIO ( '<STR_LIT:\n>' + content [ : last_newline ] ) <EOL> content = content [ last_newline + <NUM_LIT:1> : ] + next_chunk <EOL> else : <EOL> if headers is None : <EOL> csvfile = StringIO . StringIO ( content ) <EOL> else : <EOL> csvfile = StringIO . StringIO ( '<STR_LIT:\n>' + content ) <EOL> content = fh . read ( ) <EOL> if headers is None : <EOL> headers = next ( csv . reader ( csvfile , delimiter = smart_str ( delimiter ) , quotechar = smart_str ( quote_character ) ) ) <EOL> headers = [ name . strip ( ) for name in headers ] <EOL> reader = csv . DictReader ( csvfile , fieldnames = headers , delimiter = smart_str ( delimiter ) , quotechar = smart_str ( quote_character ) ) <EOL> remove_keys = None <EOL> for row in reader : <EOL> row = dict ( [ ( force_unicode ( k ) , force_unicode ( v , errors = '<STR_LIT:ignore>' ) ) for k , v in row . iteritems ( ) ] ) <EOL> if remove_keys is None : <EOL> if field_names is None : <EOL> remove_keys = [ ] <EOL> else : <EOL> remove_keys = set ( row . keys ( ) ) - set ( field_names ) <EOL> if remove_keys : <EOL> for key in remove_keys : <EOL> del row [ key ] <EOL> if timestamp_fields : <EOL> tzinfo = pytz . timezone ( settings . TIME_ZONE ) <EOL> for key in timestamp_fields : <EOL> if key in row : <EOL> dt = parse ( row [ key ] ) <EOL> if not dt . tzinfo : <EOL> dt = tzinfo . localize ( dt ) <EOL> row [ key ] = dt . astimezone ( pytz . utc ) . strftime ( '<STR_LIT>' ) <EOL> if decimal_fields : <EOL> for key in decimal_fields : <EOL> if key in row : <EOL> row [ key ] = float ( row [ key ] ) <EOL> if integer_fields : <EOL> for key in integer_fields : <EOL> if key in row : <EOL> row [ key ] = int ( row [ key ] ) <EOL> if boolean_fields : <EOL> for key in boolean_fields : <EOL> if key in row : <EOL> row [ key ] = str ( row [ key ] ) . lower ( ) == "<STR_LIT:true>" <EOL> if '<STR_LIT:id>' not in row : <EOL> row [ '<STR_LIT:id>' ] = str ( uuid . uuid4 ( ) ) <EOL> yield row <EOL> def field_values_from_log ( fh , fields = [ { '<STR_LIT:name>' : '<STR_LIT:message>' , '<STR_LIT:type>' : '<STR_LIT>' } , { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT:type>' : '<STR_LIT>' } ] ) : <EOL> """<STR_LIT>""" <EOL> buf = "<STR_LIT>" <EOL> prev = content = fh . read ( ) <EOL> if fields is None : <EOL> timestamp_key = '<STR_LIT>' <EOL> message_key = '<STR_LIT:message>' <EOL> else : <EOL> try : <EOL> timestamp_key = next ( iter ( filter ( lambda field : field [ '<STR_LIT:type>' ] in DATE_FIELD_TYPES , fields ) ) ) [ '<STR_LIT:name>' ] <EOL> except : <EOL> LOG . exception ( '<STR_LIT>' ) <EOL> timestamp_key = None <EOL> try : <EOL> message_key = next ( iter ( filter ( lambda field : field [ '<STR_LIT:type>' ] in TEXT_FIELD_TYPES , fields ) ) ) [ '<STR_LIT:name>' ] <EOL> except : <EOL> LOG . exception ( '<STR_LIT>' ) <EOL> message_key = None <EOL> def value_generator ( buf ) : <EOL> rows = buf . split ( '<STR_LIT:\n>' ) <EOL> for row in rows : <EOL> if row : <EOL> data = { } <EOL> matches = re . search ( TIMESTAMP_PATTERN , row ) <EOL> if matches and timestamp_key : <EOL> data [ timestamp_key ] = parse ( matches . groups ( ) [ <NUM_LIT:0> ] ) . astimezone ( pytz . utc ) . strftime ( '<STR_LIT>' ) <EOL> if message_key : <EOL> data [ message_key ] = row <EOL> yield data <EOL> while prev : <EOL> last_newline = content . rfind ( '<STR_LIT:\n>' ) <EOL> if last_newline > - <NUM_LIT:1> : <EOL> buf = content [ : last_newline ] <EOL> content = content [ last_newline + <NUM_LIT:1> : ] <EOL> for row in value_generator ( buf ) : <EOL> yield row <EOL> prev = fh . read ( ) <EOL> content += prev <EOL> if content : <EOL> for row in value_generator ( content ) : <EOL> yield row <EOL> def fields_from_log ( fh ) : <EOL> """<STR_LIT>""" <EOL> rows = fh . read ( ) <EOL> row = rows . split ( '<STR_LIT:\n>' ) [ <NUM_LIT:0> ] <EOL> fields = [ ] <EOL> matches = re . search ( TIMESTAMP_PATTERN , row ) <EOL> if matches : <EOL> fields . append ( ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> fields . append ( ( '<STR_LIT:message>' , '<STR_LIT>' ) ) <EOL> return fields <EOL> def get_default_fields ( ) : <EOL> """<STR_LIT>""" <EOL> default_field = DEFAULT_FIELD <EOL> default_field . update ( { '<STR_LIT:name>' : '<STR_LIT:id>' , '<STR_LIT:type>' : '<STR_LIT:string>' , '<STR_LIT>' : '<STR_LIT:false>' } ) <EOL> return [ default_field ] </s>
