eliribble
/
vanth
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

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.
This commit is contained in:
Eli Ribble 2016-06-22 17:15:34 -06:00
parent 95244d2974
commit e2821457f5
3 changed files with 241 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
tests/files/transactions.ofx Normal file
View File

@ -0,0 +1,11 @@
OFXHEADER:100
DATA:OFXSGML
VERSION:102
SECURITY:NONE
ENCODING:USASCII
CHARSET:1252
COMPRESSION:NONE
OLDFILEUID:NONE
NEWFILEUID:NONE
<OFX><SIGNONMSGSRSV1><SONRS><STATUS><CODE>0<SEVERITY>INFO<MESSAGE>The operation succeeded.</STATUS><DTSERVER>20160622111242.299[-6:MDT]<LANGUAGE>ENG<FI><ORG>America First Credit Union<FID>54324</FI></SONRS></SIGNONMSGSRSV1><BANKMSGSRSV1><STMTTRNRS><TRNUID>00000000<STATUS><CODE>0<SEVERITY>INFO</STATUS><STMTRS><CURDEF>USD<BANKACCTFROM><BANKID>324377516<ACCTID>123456-0.9:CHK<ACCTTYPE>CHECKING</BANKACCTFROM><BANKTRANLIST><DTSTART>20151231170000.000[-7:MST]<DTEND>20160622111242.327[-6:MDT]<STMTTRN><TRNTYPE>POS<DTPOSTED>20151231120000.000<DTAVAIL>20151231120000.000<TRNAMT>-50.19<FITID>0006547<NAME>UT LEHI COSTCO WHSE #0733<MEMO>POINT OF SALE PURCHASE #0006547</STMTTRN><STMTTRN><TRNTYPE>PAYMENT<DTPOSTED>20151231120000.000<DTAVAIL>20151231120000.000<TRNAMT>-79.64<FITID>0006548<NAME>Payment to PACIFICORP ONLIN<MEMO>#0006548</STMTTRN><STMTTRN><TRNTYPE>INT<DTPOSTED>20151231120000.000<DTAVAIL>20151231120000.000<TRNAMT>0.84<FITID>0006549<NAME>DIVIDEND FOR 12/01/15 - 12/31/1<MEMO>ANNUAL PERCENTAGE YIELD EARNED IS .05% #0006549</STMTTRN></BANKTRANLIST><LEDGERBAL><BALAMT>24237.85<DTASOF>20160622111242.315[-6:MDT]</LEDGERBAL><AVAILBAL><BALAMT>24237.85<DTASOF>20160622111242.315[-6:MDT]</AVAILBAL></STMTRS></STMTTRNRS></BANKMSGSRSV1></OFX>

83
tests/test_ofx.py Normal file
View File

@ -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

147
vanth/ofx.py Normal file
View File

@ -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<datetime>\d+)\.\d+\[(?P<offset>[\d\-]+):(?P<tzname>\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([
"<SIGNONMSGSRQV1>",
"<SONRQ>",
"<DTCLIENT>{}".format(now()),
"<USERID>{}".format(account['user_id']),
"<USERPASS>{}".format(account['password']),
"<LANGUAGE>ENG",
"<FI>",
"<ORG>{}".format(institution['name']),
"<FID>{}".format(institution['fid']),
"</FI>",
"<APPID>QWIN",
"<APPVER>1200",
"</SONRQ>",
"</SIGNONMSGSRQV1>",
])
def bankmsg(institution, account, start):
return "\r\n".join([
"<BANKMSGSRQV1>",
"<STMTTRNRQ>",
"<TRNUID>00000000",
"<STMTRQ>",
"<BANKACCTFROM>",
"<BANKID>{}".format(institution['bankid']),
"<ACCTID>{}".format(account['account_id']),
"<ACCTTYPE>{}".format(account['type'].upper()),
"</BANKACCTFROM>",
"<INCTRAN>",
"<DTSTART>{}".format(start.strftime("%Y%m%d")),
"<INCLUDE>Y",
"</INCTRAN>",
"</STMTRQ>",
"</STMTTRNRQ>",
"</BANKMSGSRQV1>",
])
def body(institution, account, start):
return "<OFX>\r\n" + signonmsg(institution, account) + "\r\n" + bankmsg(institution, account, start) + "\r\n</OFX>"
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))