<s> """<STR_LIT>""" <EOL> from __future__ import division <EOL> import fileinput <EOL> import collections <EOL> import csv <EOL> import os <EOL> from pprint import pprint <EOL> import common as comm <EOL> def get_quote_number ( quote_name ) : <EOL> """<STR_LIT>""" <EOL> return csv . __dict__ [ quote_name . upper ( ) ] <EOL> def get_quote_name ( quote_number ) : <EOL> """<STR_LIT>""" <EOL> for key , value in csv . __dict__ . items ( ) : <EOL> if value == quote_number : <EOL> return key <EOL> class FileTyper ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> fqfn , <EOL> delimiter = None , <EOL> rec_delimiter = None , <EOL> has_header = None , <EOL> quoting = None ) : <EOL> """<STR_LIT>""" <EOL> self . _delimiter = delimiter <EOL> self . _has_header = has_header <EOL> self . _quoting = quoting <EOL> self . fqfn = fqfn <EOL> self . format_type = None <EOL> self . fixed_length = None <EOL> self . field_cnt = None <EOL> self . record_cnt = None <EOL> self . csv_quoting = None <EOL> self . dialect = None <EOL> def analyze_file ( self ) : <EOL> """<STR_LIT>""" <EOL> if os . path . getsize ( self . fqfn ) == <NUM_LIT:0> : <EOL> raise IOErrorEmptyFile , "<STR_LIT>" <EOL> if self . _delimiter : <EOL> self . dialect = csv . Dialect <EOL> self . dialect . delimiter = self . _delimiter <EOL> self . dialect . skipinitialspace = False <EOL> if self . _quoting is None : <EOL> self . _quoting = csv . QUOTE_MINIMAL <EOL> self . dialect . quoting = self . _quoting <EOL> self . dialect . quotechar = '<STR_LIT:">' <EOL> self . dialect . lineterminator = '<STR_LIT:\n>' <EOL> else : <EOL> self . dialect = self . _get_dialect ( ) <EOL> self . dialect . lineterminator = '<STR_LIT:\n>' <EOL> self . _quoting = self . dialect . quoting <EOL> self . _delimiter = self . dialect . delimiter <EOL> if self . dialect . quoting == csv . QUOTE_NONE : <EOL> self . csv_quoting = False <EOL> else : <EOL> self . csv_quoting = True <EOL> self . format_type = self . _get_format_type ( ) <EOL> self . dialect . has_header = self . _get_has_header ( self . _has_header ) <EOL> self . _has_header = self . dialect . has_header <EOL> self . field_cnt = self . _get_field_cnt ( ) <EOL> self . record_cnt = self . _count_records ( ) <EOL> return self . dialect <EOL> def _get_dialect ( self ) : <EOL> """<STR_LIT>""" <EOL> csvfile = open ( self . fqfn , "<STR_LIT:rb>" ) <EOL> try : <EOL> dialect = csv . Sniffer ( ) . sniff ( csvfile . read ( <NUM_LIT> ) ) <EOL> except : <EOL> raise IOError , '<STR_LIT>' <EOL> csvfile . close ( ) <EOL> dialect . quoting = self . _get_dialect_quoting ( dialect ) <EOL> return dialect <EOL> def _get_dialect_quoting ( self , dialect ) : <EOL> """<STR_LIT>""" <EOL> total_field_cnt = collections . defaultdict ( int ) <EOL> quoted_field_cnt = collections . defaultdict ( int ) <EOL> rec_cnt = <NUM_LIT:0> <EOL> for rec in fileinput . input ( self . fqfn ) : <EOL> fields = rec [ : - <NUM_LIT:1> ] . split ( dialect . delimiter ) <EOL> total_field_cnt [ len ( fields ) ] += <NUM_LIT:1> <EOL> quoted_cnt = <NUM_LIT:0> <EOL> for field in fields : <EOL> if len ( field ) >= <NUM_LIT:2> : <EOL> if field [ <NUM_LIT:0> ] == '<STR_LIT:">' and field [ - <NUM_LIT:1> ] == '<STR_LIT:">' : <EOL> quoted_cnt += <NUM_LIT:1> <EOL> quoted_field_cnt [ quoted_cnt ] += <NUM_LIT:1> <EOL> if rec_cnt > <NUM_LIT:1000> : <EOL> break <EOL> fileinput . close ( ) <EOL> if ( len ( total_field_cnt ) == <NUM_LIT:1> <EOL> and len ( quoted_field_cnt ) == <NUM_LIT:1> ) : <EOL> if total_field_cnt [ <NUM_LIT:0> ] == quoted_field_cnt [ <NUM_LIT:0> ] : <EOL> return csv . QUOTE_ALL <EOL> elif quoted_field_cnt [ <NUM_LIT:0> ] == <NUM_LIT:0> : <EOL> return csv . QUOTE_NONE <EOL> common_field_key , common_field_pct = comm . get_common_key ( total_field_cnt ) <EOL> common_quoted_field_key , common_quoted_field_pct = comm . get_common_key ( quoted_field_cnt ) <EOL> if ( common_field_pct > <NUM_LIT> <EOL> and common_quoted_field_pct > <NUM_LIT> ) : <EOL> if common_field_key == common_quoted_field_key : <EOL> return csv . QUOTE_ALL <EOL> else : <EOL> return csv . QUOTE_NONE <EOL> return dialect . quoting <EOL> def _get_has_header ( self , has_header ) : <EOL> """<STR_LIT>""" <EOL> if has_header is None : <EOL> sample = open ( self . fqfn , '<STR_LIT:r>' ) . read ( <NUM_LIT> ) <EOL> try : <EOL> return csv . Sniffer ( ) . has_header ( sample ) <EOL> except : <EOL> raise IOError , '<STR_LIT>' <EOL> else : <EOL> return has_header <EOL> def _get_field_cnt ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . _delimiter or len ( self . _delimiter ) == <NUM_LIT:1> : <EOL> csvfile = open ( self . fqfn , "<STR_LIT:r>" ) <EOL> for row in csv . reader ( csvfile , self . dialect ) : <EOL> row_len = len ( row ) <EOL> csvfile . close ( ) <EOL> break <EOL> else : <EOL> print '<STR_LIT>' <EOL> for rec in fileinput . input ( self . fqfn ) : <EOL> fields = rec [ : - <NUM_LIT:1> ] . split ( self . _delimiter ) <EOL> fileinput . close ( ) <EOL> row_len = len ( fields ) <EOL> break <EOL> return row_len <EOL> def _get_format_type ( self ) : <EOL> """<STR_LIT>""" <EOL> rec_length = collections . defaultdict ( int ) <EOL> rec_cnt = <NUM_LIT:0> <EOL> for rec in fileinput . input ( self . fqfn ) : <EOL> rec_length [ len ( rec ) ] += <NUM_LIT:1> <EOL> rec_cnt += <NUM_LIT:1> <EOL> if rec_cnt > <NUM_LIT:1000> : <EOL> break <EOL> fileinput . close ( ) <EOL> if len ( rec_length ) == <NUM_LIT:1> : <EOL> return '<STR_LIT>' <EOL> else : <EOL> return '<STR_LIT>' <EOL> def _count_records ( self ) : <EOL> """<STR_LIT>""" <EOL> rec_cnt = <NUM_LIT:0> <EOL> for dummy in fileinput . input ( self . fqfn ) : <EOL> rec_cnt += <NUM_LIT:1> <EOL> fileinput . close ( ) <EOL> return rec_cnt <EOL> class IOErrorEmptyFile ( IOError ) : <EOL> """<STR_LIT>""" <EOL> pass </s>
