-
-
Save whistler/e7c21c70d1cbb9c4b15d to your computer and use it in GitHub Desktop.
from csv import DictWriter | |
from glob import glob | |
from ofxparse import OfxParser | |
DATE_FORMAT = "%m/%d/%Y" | |
def write_csv(statement, out_file): | |
print "Writing: " + out_file | |
fields = ['date', 'payee', 'debit', 'credit', 'balance'] | |
with open(out_file, 'w') as f: | |
writer = DictWriter(f, fieldnames=fields) | |
for line in statement: | |
writer.writerow(line) | |
def get_statement_from_qfx(qfx): | |
balance = qfx.account.statement.balance | |
statement = [] | |
for transaction in qfx.account.statement.transactions: | |
credit = "" | |
debit = "" | |
balance = balance + transaction.amount | |
if transaction.type == 'credit': | |
credit = transaction.amount | |
elif transaction.type == 'debit': | |
debit = -transaction.amount | |
else: | |
raise Error("Unknown transaction type") | |
line = { | |
'date': transaction.date.strftime(DATE_FORMAT), | |
'payee': transaction.payee, | |
'debit': debit, | |
'credit': credit, | |
'balance': balance | |
} | |
statement.append(line) | |
return statement | |
files = glob("*.qfx") | |
for qfx_file in files: | |
qfx = OfxParser.parse(file(qfx_file)) | |
statement = get_statement_from_qfx(qfx) | |
out_file = "converted_" + qfx_file.replace(".qfx",".csv") | |
write_csv(statement, out_file) | |
Need to install some non-standard libraries -- any help there?
I ended up using this code
Thanks for this.
Here's a fork that turns this into a CLI with various input and output formats: https://gist.github.com/abirchall/54b0f2eb47d02eb3e36e33b3077298a5
[@oravecz] Need to install some non-standard libraries -- any help there?
You can run: pip3 install ofxparse
.
You can find info about this library here: https://pypi.org/project/ofxparse/
The underlying code is simple enough, just a bunch of parsing logic specific to the OFX spec.
Nice! I added a -i input argument to @schulzebittar 's script so I can select specific files to convert, if I want. Hope that helps.
#!/usr/bin/env` python3
# -*- coding: utf-8 -*-
from csv import DictWriter
from glob import glob
from ofxparse import OfxParser
import json
import urllib.request
import datetime
import argparse
argparser = argparse.ArgumentParser()
argparser.add_argument("-o", "--outputtype", help = "csv or json", default="csv")
argparser.add_argument("-i", "--input", nargs='+', help = "input file(s)", default="*.ofx")
args = argparser.parse_args()
DATE_FORMAT = "%m/%d/%Y"
jsonBody = {}
outputtype = args.outputtype
jsonBody["data"] = []
allStatements = []
def write_csv(statement, out_file):
print("Writing: " + out_file)
fields = ['Date', 'Description (payee)', 'Transaction Type (type)', 'UID', 'Amount',
'sic', 'mcc', 'Notes (memo)', 'Debit', 'Credit', 'Balance', 'FID', 'Organization']
with open(out_file, 'w', newline='') as f:
writer = DictWriter(f, fieldnames=fields)
writer.writeheader()
for line in statement:
writer.writerow(line)
def get_statement_from_qfx(qfx):
#print(qfx.account.account_id)
#print(qfx.account.institution.organization)
#print(qfx.account.institution.fid)
balance = qfx.account.statement.balance
statement = []
credit_transactions = ['credit', 'dep', 'int', 'directdep']
debit_transactions = ['debit', 'atm', 'pos',
'xfer', 'check', 'fee', 'payment']
other_transactions = ['other']
for transaction in qfx.account.statement.transactions:
#print(transaction.type)
credit = ""
debit = ""
balance = balance + transaction.amount
if transaction.type in credit_transactions:
credit = transaction.amount
elif transaction.type in debit_transactions:
debit = -transaction.amount
elif transaction.type in other_transactions:
if transaction.amount < 0:
debit = -transaction.amount
else:
credit = transaction.amount
else:
raise ValueError("Unknown transaction type:" + transaction.type)
line = {
'Date': transaction.date.strftime(DATE_FORMAT),
'Description (payee)': transaction.payee,
'Transaction Type (type)': transaction.type,
'Notes (memo)': transaction.memo,
'UID': transaction.id,
'Amount': str(transaction.amount),
'sic': transaction.sic,
'mcc': transaction.mcc,
'Debit': str(debit),
'Credit': str(credit),
'Balance': str(balance),
'FID': qfx.account.institution.fid,
'Organization': qfx.account.institution.organization}
statement.append(line)
jsonBody["data"].append(line)
return statement
files = glob(args.input)
for qfx_file in files:
qfx = OfxParser.parse(open(qfx_file, encoding="latin-1"))
statement = get_statement_from_qfx(qfx)
allStatements = allStatements + statement
#print(statement)
if outputtype == 'csv' and (len(args.input) > 1 or args.input == '*.ofx'):
out_file = "convert ed_" + qfx_file.replace(".ofx", ".csv")
write_csv(statement, out_file)
if outputtype == 'singlecsv':
out_file = "qfx-transactions.csv"
write_csv(allStatements, out_file)
elif outputtype == 'json' and (len(args.input) > 1 or args.input == '*.ofx'):
print('json output supports one file only')
else:
with open('qfx-transactions.json', 'w') as outfile:
json.dump(jsonBody, outfile)
I make in more generic to extract as much data from the QFX:
here the code: https://github.com/nhatkhai/ofx2csv/blob/main/ofx2csv.py
I make in more generic to extract as much data from the QFX: here the code: https://github.com/nhatkhai/ofx2csv/blob/main/ofx2csv.py
This looks good. I opened a PR to fix a few little bugs nhatkhai/ofx2csv#1
Need to install some non-standard libraries -- any help there?