From e2821457f5c26309de750c9bd7bc62b9ab2d9fa1 Mon Sep 17 00:00:00 2001 From: Eli Ribble Date: Wed, 22 Jun 2016 17:15:34 -0600 Subject: [PATCH] Add translating parsed SGML document into OFX structures This gives the structures some semantic meaning beyond just being raw SGML and adds some niceties in like parsing the datetimes and their timezones. --- tests/files/transactions.ofx | 11 +++ tests/test_ofx.py | 83 ++++++++++++++++++++ vanth/ofx.py | 147 +++++++++++++++++++++++++++++++++++ 3 files changed, 241 insertions(+) create mode 100644 tests/files/transactions.ofx create mode 100644 tests/test_ofx.py create mode 100644 vanth/ofx.py 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))