<s> """<STR_LIT>""" <EOL> import codecs <EOL> import csv <EOL> import six <EOL> from agate . exceptions import FieldSizeLimitError <EOL> EIGHT_BIT_ENCODINGS = [ <EOL> '<STR_LIT:utf-8>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:utf8>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> ] <EOL> POSSIBLE_DELIMITERS = [ '<STR_LIT:U+002C>' , '<STR_LIT:\t>' , '<STR_LIT:;>' , '<STR_LIT:U+0020>' , '<STR_LIT::>' , '<STR_LIT:|>' ] <EOL> class UTF8Recoder ( six . Iterator ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , f , encoding ) : <EOL> self . reader = codecs . getreader ( encoding ) ( f ) <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def __next__ ( self ) : <EOL> return next ( self . reader ) . encode ( '<STR_LIT:utf-8>' ) <EOL> class UnicodeReader ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , f , encoding = '<STR_LIT:utf-8>' , field_size_limit = None , line_numbers = False , header = True , ** kwargs ) : <EOL> self . line_numbers = line_numbers <EOL> self . header = header <EOL> f = UTF8Recoder ( f , encoding ) <EOL> self . reader = csv . reader ( f , ** kwargs ) <EOL> if field_size_limit : <EOL> csv . field_size_limit ( field_size_limit ) <EOL> def next ( self ) : <EOL> try : <EOL> row = next ( self . reader ) <EOL> except csv . Error as e : <EOL> if '<STR_LIT>' in str ( e ) : <EOL> raise FieldSizeLimitError ( csv . field_size_limit ( ) ) <EOL> else : <EOL> raise e <EOL> if self . line_numbers : <EOL> if self . header and self . line_num == <NUM_LIT:1> : <EOL> row . insert ( <NUM_LIT:0> , '<STR_LIT>' ) <EOL> else : <EOL> row . insert ( <NUM_LIT:0> , str ( self . line_num - <NUM_LIT:1> if self . header else self . line_num ) ) <EOL> return [ six . text_type ( s , '<STR_LIT:utf-8>' ) for s in row ] <EOL> def __iter__ ( self ) : <EOL> return self <EOL> @ property <EOL> def dialect ( self ) : <EOL> return self . reader . dialect <EOL> @ property <EOL> def line_num ( self ) : <EOL> return self . reader . line_num <EOL> class UnicodeWriter ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , f , encoding = '<STR_LIT:utf-8>' , ** kwargs ) : <EOL> self . encoding = encoding <EOL> self . _eight_bit = ( self . encoding . lower ( ) . replace ( '<STR_LIT:_>' , '<STR_LIT:->' ) in EIGHT_BIT_ENCODINGS ) <EOL> if self . _eight_bit : <EOL> self . writer = csv . writer ( f , ** kwargs ) <EOL> else : <EOL> self . queue = six . StringIO ( ) <EOL> self . writer = csv . writer ( self . queue , ** kwargs ) <EOL> self . stream = f <EOL> self . encoder = codecs . getincrementalencoder ( encoding ) ( ) <EOL> def writerow ( self , row ) : <EOL> if self . _eight_bit : <EOL> self . writer . writerow ( [ six . text_type ( s if s is not None else '<STR_LIT>' ) . encode ( self . encoding ) for s in row ] ) <EOL> else : <EOL> self . writer . writerow ( [ six . text_type ( s if s is not None else '<STR_LIT>' ) . encode ( '<STR_LIT:utf-8>' ) for s in row ] ) <EOL> data = self . queue . getvalue ( ) <EOL> data = data . decode ( '<STR_LIT:utf-8>' ) <EOL> data = self . encoder . encode ( data ) <EOL> self . stream . write ( data ) <EOL> self . queue . truncate ( <NUM_LIT:0> ) <EOL> def writerows ( self , rows ) : <EOL> for row in rows : <EOL> self . writerow ( row ) <EOL> class UnicodeDictReader ( csv . DictReader ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , f , fieldnames = None , restkey = None , restval = None , * args , ** kwargs ) : <EOL> reader = UnicodeReader ( f , * args , ** kwargs ) <EOL> if '<STR_LIT>' in kwargs : <EOL> kwargs . pop ( '<STR_LIT>' ) <EOL> csv . DictReader . __init__ ( self , f , fieldnames , restkey , restval , * args , ** kwargs ) <EOL> self . reader = reader <EOL> class UnicodeDictWriter ( csv . DictWriter ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , f , fieldnames , restval = '<STR_LIT>' , extrasaction = '<STR_LIT>' , * args , ** kwds ) : <EOL> self . fieldnames = fieldnames <EOL> self . restval = restval <EOL> if extrasaction . lower ( ) not in ( '<STR_LIT>' , '<STR_LIT:ignore>' ) : <EOL> raise ValueError ( '<STR_LIT>' % extrasaction ) <EOL> self . extrasaction = extrasaction <EOL> self . writer = UnicodeWriter ( f , * args , ** kwds ) <EOL> class Reader ( UnicodeReader ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class Writer ( UnicodeWriter ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , f , encoding = '<STR_LIT:utf-8>' , line_numbers = False , ** kwargs ) : <EOL> self . row_count = <NUM_LIT:0> <EOL> self . line_numbers = line_numbers <EOL> if '<STR_LIT>' not in kwargs : <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT:\n>' <EOL> UnicodeWriter . __init__ ( self , f , encoding , ** kwargs ) <EOL> def _append_line_number ( self , row ) : <EOL> if self . row_count == <NUM_LIT:0> : <EOL> row . insert ( <NUM_LIT:0> , '<STR_LIT>' ) <EOL> else : <EOL> row . insert ( <NUM_LIT:0> , self . row_count ) <EOL> self . row_count += <NUM_LIT:1> <EOL> def writerow ( self , row ) : <EOL> if self . line_numbers : <EOL> row = list ( row ) <EOL> self . _append_line_number ( row ) <EOL> row = [ i . replace ( '<STR_LIT:\r>' , '<STR_LIT:\n>' ) if isinstance ( i , six . string_types ) else i for i in row ] <EOL> UnicodeWriter . writerow ( self , row ) <EOL> def writerows ( self , rows ) : <EOL> for row in rows : <EOL> self . writerow ( row ) <EOL> class DictReader ( UnicodeDictReader ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class DictWriter ( UnicodeDictWriter ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , f , fieldnames , encoding = '<STR_LIT:utf-8>' , line_numbers = False , ** kwargs ) : <EOL> self . row_count = <NUM_LIT:0> <EOL> self . line_numbers = line_numbers <EOL> if '<STR_LIT>' not in kwargs : <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT:\n>' <EOL> UnicodeDictWriter . __init__ ( self , f , fieldnames , encoding = encoding , ** kwargs ) <EOL> def _append_line_number ( self , row ) : <EOL> if self . row_count == <NUM_LIT:0> : <EOL> row [ '<STR_LIT>' ] = <NUM_LIT:0> <EOL> else : <EOL> row [ '<STR_LIT>' ] = self . row_count <EOL> self . row_count += <NUM_LIT:1> <EOL> def writerow ( self , row ) : <EOL> if self . line_numbers : <EOL> row = list ( row ) <EOL> self . _append_line_number ( row ) <EOL> row = dict ( [ ( k , v . replace ( '<STR_LIT:\r>' , '<STR_LIT:\n>' ) ) if isinstance ( v , basestring ) else ( k , v ) for k , v in row . items ( ) ] ) <EOL> UnicodeDictWriter . writerow ( self , row ) <EOL> def writerows ( self , rows ) : <EOL> for row in rows : <EOL> self . writerow ( row ) <EOL> class Sniffer ( object ) : <EOL> """<STR_LIT>""" <EOL> def sniff ( self , sample ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> dialect = csv . Sniffer ( ) . sniff ( sample , POSSIBLE_DELIMITERS ) <EOL> except : <EOL> dialect = None <EOL> return dialect <EOL> def reader ( * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return Reader ( * args , ** kwargs ) <EOL> def writer ( * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return Writer ( * args , ** kwargs ) </s>
