<s> """<STR_LIT>""" <EOL> import os <EOL> import sys <EOL> import datetime <EOL> import collections <EOL> from . ifiles import ifind <EOL> from . struct_parser import StructParser <EOL> from . field_parser import FieldParser <EOL> from . memo import find_memofile , open_memofile , FakeMemoFile , BinaryMemo <EOL> from . codepages import guess_encoding <EOL> from . dbversions import get_dbversion_string <EOL> from . exceptions import * <EOL> DBFHeader = StructParser ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] ) <EOL> DBFField = StructParser ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> [ '<STR_LIT:name>' , <EOL> '<STR_LIT:type>' , <EOL> '<STR_LIT:address>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] ) <EOL> def expand_year ( year ) : <EOL> """<STR_LIT>""" <EOL> if year < <NUM_LIT> : <EOL> return <NUM_LIT> + year <EOL> else : <EOL> return <NUM_LIT> + year <EOL> class RecordIterator ( object ) : <EOL> def __init__ ( self , table , record_type ) : <EOL> self . _record_type = record_type <EOL> self . _table = table <EOL> def __iter__ ( self ) : <EOL> return self . _table . _iter_records ( self . _record_type ) <EOL> def __len__ ( self ) : <EOL> return self . _table . _count_records ( self . _record_type ) <EOL> class DBF ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , filename , encoding = None , ignorecase = True , <EOL> lowernames = False , <EOL> parserclass = FieldParser , <EOL> recfactory = collections . OrderedDict , <EOL> load = False , <EOL> raw = False , <EOL> ignore_missing_memofile = False ) : <EOL> self . encoding = encoding <EOL> self . ignorecase = ignorecase <EOL> self . lowernames = lowernames <EOL> self . parserclass = parserclass <EOL> self . raw = raw <EOL> self . ignore_missing_memofile = ignore_missing_memofile <EOL> if recfactory is None : <EOL> self . recfactory = lambda items : items <EOL> else : <EOL> self . recfactory = recfactory <EOL> self . name = os . path . basename ( filename ) <EOL> self . name = os . path . splitext ( self . name ) [ <NUM_LIT:0> ] . lower ( ) <EOL> self . _records = None <EOL> self . _deleted = None <EOL> if ignorecase : <EOL> self . filename = ifind ( filename ) <EOL> if not self . filename : <EOL> raise DBFNotFound ( '<STR_LIT>' . format ( filename ) ) <EOL> else : <EOL> self . filename = filename <EOL> self . memofilename = None <EOL> self . header = None <EOL> self . fields = [ ] <EOL> self . field_names = [ ] <EOL> with open ( self . filename , mode = '<STR_LIT:rb>' ) as infile : <EOL> self . _read_header ( infile ) <EOL> self . _read_field_headers ( infile ) <EOL> self . _check_headers ( ) <EOL> try : <EOL> self . date = datetime . date ( expand_year ( self . header . year ) , <EOL> self . header . month , <EOL> self . header . day ) <EOL> except ValueError : <EOL> self . date = None <EOL> self . memofilename = self . _get_memofilename ( ) <EOL> if load : <EOL> self . load ( ) <EOL> @ property <EOL> def dbversion ( self ) : <EOL> return get_dbversion_string ( self . header . dbversion ) <EOL> def _get_memofilename ( self ) : <EOL> field_types = [ field . type for field in self . fields ] <EOL> if not set ( field_types ) & set ( '<STR_LIT>' ) : <EOL> return None <EOL> path = find_memofile ( self . filename ) <EOL> if path is None : <EOL> if self . ignore_missing_memofile : <EOL> return None <EOL> raise MissingMemoFile ( '<STR_LIT>' . format ( <EOL> self . filename ) ) <EOL> else : <EOL> return path <EOL> @ property <EOL> def loaded ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _records is not None <EOL> def load ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . loaded : <EOL> self . _records = list ( self . _iter_records ( b'<STR_LIT:U+0020>' ) ) <EOL> self . _deleted = list ( self . _iter_records ( b'<STR_LIT:*>' ) ) <EOL> def unload ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _records = None <EOL> self . _deleted = None <EOL> @ property <EOL> def records ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . loaded : <EOL> return self . _records <EOL> else : <EOL> return RecordIterator ( self , b'<STR_LIT:U+0020>' ) <EOL> @ property <EOL> def deleted ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . loaded : <EOL> return self . _deleted <EOL> else : <EOL> return RecordIterator ( self , b'<STR_LIT:*>' ) <EOL> def _read_header ( self , infile ) : <EOL> self . header = DBFHeader . read ( infile ) <EOL> if self . encoding is None : <EOL> try : <EOL> self . encoding = guess_encoding ( self . header . language_driver ) <EOL> except LookupError as err : <EOL> self . encoding = '<STR_LIT:ascii>' <EOL> def _read_field_headers ( self , infile ) : <EOL> while True : <EOL> sep = infile . read ( <NUM_LIT:1> ) <EOL> if sep in ( b'<STR_LIT:\r>' , b'<STR_LIT:\n>' , b'<STR_LIT>' ) : <EOL> break <EOL> field = DBFField . unpack ( sep + infile . read ( DBFField . size - <NUM_LIT:1> ) ) <EOL> field . type = chr ( ord ( field . type ) ) <EOL> field . name = field . name . split ( b'<STR_LIT>' ) [ <NUM_LIT:0> ] . decode ( self . encoding ) <EOL> if self . lowernames : <EOL> field . name = field . name . lower ( ) <EOL> self . field_names . append ( field . name ) <EOL> self . fields . append ( field ) <EOL> def _open_memofile ( self ) : <EOL> if self . memofilename and not self . raw : <EOL> return open_memofile ( self . memofilename , self . header . dbversion ) <EOL> else : <EOL> return FakeMemoFile ( self . memofilename ) <EOL> def _check_headers ( self ) : <EOL> field_parser = self . parserclass ( self ) <EOL> """<STR_LIT>""" <EOL> for field in self . fields : <EOL> if field . type == '<STR_LIT:I>' and field . length != <NUM_LIT:4> : <EOL> message = '<STR_LIT>' <EOL> raise ValueError ( message . format ( field . length ) ) <EOL> elif field . type == '<STR_LIT:L>' and field . length != <NUM_LIT:1> : <EOL> message = '<STR_LIT>' <EOL> raise ValueError ( message . format ( field . length ) ) <EOL> elif not field_parser . field_type_supported ( field . type ) : <EOL> raise ValueError ( '<STR_LIT>' . format ( field . type ) ) <EOL> def _skip_record ( self , infile ) : <EOL> infile . seek ( self . header . recordlen - <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> def _count_records ( self , record_type = b'<STR_LIT:U+0020>' ) : <EOL> count = <NUM_LIT:0> <EOL> with open ( self . filename , '<STR_LIT:rb>' ) as infile : <EOL> infile . seek ( self . header . headerlen , <NUM_LIT:0> ) <EOL> while True : <EOL> sep = infile . read ( <NUM_LIT:1> ) <EOL> if sep == record_type : <EOL> count += <NUM_LIT:1> <EOL> self . _skip_record ( infile ) <EOL> elif sep in ( b'<STR_LIT>' , b'<STR_LIT>' ) : <EOL> break <EOL> else : <EOL> self . _skip_record ( infile ) <EOL> return count <EOL> def _iter_records ( self , record_type = b'<STR_LIT:U+0020>' ) : <EOL> with open ( self . filename , '<STR_LIT:rb>' ) as infile , self . _open_memofile ( ) as memofile : <EOL> infile . seek ( self . header . headerlen , <NUM_LIT:0> ) <EOL> if not self . raw : <EOL> field_parser = self . parserclass ( self , memofile ) <EOL> parse = field_parser . parse <EOL> skip_record = self . _skip_record <EOL> read = infile . read <EOL> while True : <EOL> sep = read ( <NUM_LIT:1> ) <EOL> if sep == record_type : <EOL> if self . raw : <EOL> items = [ ( field . name , read ( field . length ) ) for field in self . fields ] <EOL> else : <EOL> items = [ ( field . name , <EOL> parse ( field , read ( field . length ) ) ) for field in self . fields ] <EOL> yield self . recfactory ( items ) <EOL> elif sep in ( b'<STR_LIT>' , b'<STR_LIT>' ) : <EOL> break <EOL> else : <EOL> skip_record ( infile ) <EOL> def __iter__ ( self ) : <EOL> if self . loaded : <EOL> return list . __iter__ ( self . _records ) <EOL> else : <EOL> return self . _iter_records ( ) <EOL> def __len__ ( self ) : <EOL> return len ( self . records ) <EOL> def __repr__ ( self ) : <EOL> if self . loaded : <EOL> status = '<STR_LIT>' <EOL> else : <EOL> status = '<STR_LIT>' <EOL> return '<STR_LIT>' . format ( status , self . filename ) <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , type , value , traceback ) : <EOL> self . unload ( ) <EOL> return False </s>
