diff --git a/tests/files/transactions.ofx b/tests/files/transactions.ofx new file mode 100644 index 0000000..c146848 --- /dev/null +++ b/tests/files/transactions.ofx @@ -0,0 +1,11 @@ +OFXHEADER:100 +DATA:OFXSGML +VERSION:102 +SECURITY:NONE +ENCODING:USASCII +CHARSET:1252 +COMPRESSION:NONE +OLDFILEUID:NONE +NEWFILEUID:NONE + +0INFOThe operation succeeded.20160622111242.299[-6:MDT]ENGAmerica First Credit Union54324000000000INFOUSD324377516123456-0.9:CHKCHECKING20151231170000.000[-7:MST]20160622111242.327[-6:MDT]POS20151231120000.00020151231120000.000-50.190006547UT LEHI COSTCO WHSE #0733POINT OF SALE PURCHASE #0006547PAYMENT20151231120000.00020151231120000.000-79.640006548Payment to PACIFICORP ONLIN#0006548INT20151231120000.00020151231120000.0000.840006549DIVIDEND FOR 12/01/15 - 12/31/1ANNUAL PERCENTAGE YIELD EARNED IS .05% #000654924237.8520160622111242.315[-6:MDT]24237.8520160622111242.315[-6:MDT] diff --git a/tests/test_ofx.py b/tests/test_ofx.py new file mode 100644 index 0000000..1b808d0 --- /dev/null +++ b/tests/test_ofx.py @@ -0,0 +1,83 @@ +import datetime + +import vanth.ofx + + +def MST(): + return datetime.timezone(datetime.timedelta(hours=-7), 'MST') + +def MDT(): + return datetime.timezone(datetime.timedelta(hours=-6), 'MDT') + +def test_query_transactions(mocker): + institution = { + 'bankid' : "1234567", + 'fid' : "12345", + 'name' : "AFCU", + } + account = { + "account_id" : "123456-0.9:CHK", + "user_id" : "123456789", + "password" : "1234", + "type" : "checking", + } + with mocker.patch('vanth.ofx.now', return_value='20160102030405.000[-7:MST]'): + results = vanth.ofx.query_transactions(institution, account, start=datetime.date(2016, 1, 2)) + with open('tests/files/query_transactions.ofx', 'rb') as f: + expected = f.read().decode('utf-8') + assert results == expected + +def test_parse(): + with open('tests/files/transactions.ofx', 'rb') as f: + transactions = f.read().decode('utf-8') + document = vanth.ofx.parse(transactions) + assert document.header == { + 'CHARSET' : '1252', + 'COMPRESSION' : 'NONE', + 'DATA' : 'OFXSGML', + 'ENCODING' : 'USASCII', + 'NEWFILEUID' : 'NONE', + 'OFXHEADER' : '100', + 'OLDFILEUID' : 'NONE', + 'SECURITY' : 'NONE', + 'VERSION' : '102' + } + assert document.body.status.code == '0' + assert document.body.status.severity == 'INFO' + assert document.body.status.message == 'The operation succeeded.' + assert document.body.statement.status.code == '0' + assert document.body.statement.status.severity == 'INFO' + assert document.body.statement.status.message is None + assert document.body.statement.transactions.currency == 'USD' + assert document.body.statement.transactions.account.accountid == '123456-0.9:CHK' + assert document.body.statement.transactions.account.bankid == '324377516' + assert document.body.statement.transactions.account.type == 'CHECKING' + assert document.body.statement.transactions.start == datetime.datetime(2015, 12, 31, 17, 0, tzinfo=MST()) + assert document.body.statement.transactions.end == datetime.datetime(2016, 6, 22, 11, 12, 42, tzinfo=MDT()) + expected_items = [{ + 'amount' : -50.19, + 'available' : datetime.datetime(2015, 12, 31, 12), + 'id' : '0006547', + 'memo' : 'POINT OF SALE PURCHASE #0006547', + 'name' : 'UT LEHI COSTCO WHSE #0733', + 'posted' : datetime.datetime(2015, 12, 31, 12), + 'type' : 'POS', + },{ + 'amount' : -79.64, + 'available' : datetime.datetime(2015, 12, 31, 12), + 'id' : '0006548', + 'memo' : '#0006548', + 'name' : 'Payment to PACIFICORP ONLIN', + 'posted' : datetime.datetime(2015, 12, 31, 12), + 'type' : 'PAYMENT', + },{ + 'amount' : 0.84, + 'available' : datetime.datetime(2015, 12, 31, 12), + 'id' : '0006549', + 'memo' : 'ANNUAL PERCENTAGE YIELD EARNED IS .05% #0006549', + 'name' : 'DIVIDEND FOR 12/01/15 - 12/31/1', + 'posted' : datetime.datetime(2015, 12, 31, 12), + 'type' : 'INT', + }] + items = [dict(item) for item in document.body.statement.transactions.items] + assert items == expected_items diff --git a/vanth/ofx.py b/vanth/ofx.py new file mode 100644 index 0000000..45e392b --- /dev/null +++ b/vanth/ofx.py @@ -0,0 +1,147 @@ +import collections +import datetime +import re + +import vanth.sgml + +Document = collections.namedtuple('Document', ['header', 'body']) + +class Body(): # pylint:disable=too-few-public-methods + def __init__(self, sgml): + self.status = Status(sgml['SIGNONMSGSRSV1']['SONRS']['STATUS']) + self.statement = TransactionStatement(sgml['BANKMSGSRSV1']['STMTTRNRS']) + +class Status(): # pylint:disable=too-few-public-methods + def __init__(self, sgml): + self.code = sgml['CODE'].value + self.severity = sgml['SEVERITY'].value + self.message = sgml['MESSAGE'].value if sgml['MESSAGE'] else None + +class TransactionStatement(): # pylint:disable=too-few-public-methods + def __init__(self, sgml): + self.trnuid = sgml['TRNUID'].value + self.status = Status(sgml['STATUS']) + self.transactions = TransactionList(sgml['STMTRS']) + +class TransactionList(): # pylint:disable=too-few-public-methods + def __init__(self, sgml): + self.currency = sgml['CURDEF'].value + self.account = Account(sgml['BANKACCTFROM']) + self.start = _parse_date_with_tz(sgml['BANKTRANLIST']['DTSTART'].value) + self.end = _parse_date_with_tz(sgml['BANKTRANLIST']['DTEND'].value) + self.items = [Transaction(child) for child in sgml['BANKTRANLIST'].children if child.name == 'STMTTRN'] + +class Transaction(): # pylint:disable=too-few-public-methods + def __init__(self, sgml): + self.amount = float(sgml['TRNAMT'].value) + self.available = _parse_date(sgml['DTAVAIL'].value) + self.id = sgml['FITID'].value + self.memo = sgml['MEMO'].value + self.name = sgml['NAME'].value + self.posted = _parse_date(sgml['DTPOSTED'].value) + self.type = sgml['TRNTYPE'].value + + def __iter__(self): + return ((prop, getattr(self, prop)) for prop in ('amount', 'available', 'id', 'memo', 'name', 'posted', 'type')) + +class Account(): # pylint:disable=too-few-public-methods + def __init__(self, sgml): + self.bankid = sgml['BANKID'].value + self.accountid = sgml['ACCTID'].value + self.type = sgml['ACCTTYPE'].value + +def _fix_offset(offset): + result = int(offset) * 100 + return "{:04d}".format(result) if result > 0 else "{:05d}".format(result) + +def _parse_date(date): + return datetime.datetime.strptime(date, "%Y%m%d%H%M%S.000") + +def _parse_date_with_tz(date): + match = re.match(r'(?P\d+)\.\d+\[(?P[\d\-]+):(?P\w+)\]', date) + if not match: + raise ValueError("Unable to extract datetime from {}".format(date)) + formatted = "{datetime} {offset} {tzname}".format( + datetime = match.group('datetime'), + offset = _fix_offset(match.group('offset')), + tzname = match.group('tzname'), + ) + return datetime.datetime.strptime(formatted, "%Y%m%d%H%M%S %z %Z") + +def header(): + return "\r\n".join([ + "OFXHEADER:100", + "DATA:OFXSGML", + "VERSION:102", + "SECURITY:NONE", + "ENCODING:USASCII", + "CHARSET:1252", + "COMPRESSION:NONE", + "OLDFILEUID:NONE", + "NEWFILEUID:NONE", + ]) + +def now(): + return datetime.datetime.now().strftime("%Y%m%d%H%M%S.000[-7:MST]") + +def signonmsg(institution, account): + return "\r\n".join([ + "", + "", + "{}".format(now()), + "{}".format(account['user_id']), + "{}".format(account['password']), + "ENG", + "", + "{}".format(institution['name']), + "{}".format(institution['fid']), + "", + "QWIN", + "1200", + "", + "", + ]) + +def bankmsg(institution, account, start): + return "\r\n".join([ + "", + "", + "00000000", + "", + "", + "{}".format(institution['bankid']), + "{}".format(account['account_id']), + "{}".format(account['type'].upper()), + "", + "", + "{}".format(start.strftime("%Y%m%d")), + "Y", + "", + "", + "", + "", + ]) + +def body(institution, account, start): + return "\r\n" + signonmsg(institution, account) + "\r\n" + bankmsg(institution, account, start) + "\r\n" + +def query_transactions(institution, account, start=None): + return header() + (2*"\r\n") + body(institution, account, start) + "\r\n" + + +def _first_empty_line(lines): + for i, line in enumerate(lines): + if not line: + return i + +def _parse_header(header_lines): + splits = [line.partition(':') for line in header_lines] + return {k: v for k, _, v in splits} + +def parse(content): + lines = content.split('\r\n') + split = _first_empty_line(lines) + header_lines = lines[:split] + _header = _parse_header(header_lines) + _body = vanth.sgml.parse('\n'.join(lines[split+1:])) + return Document(_header, Body(_body))