Instantly share code, notes, and snippets.

Embed
What would you like to do?
#!/home/fanqiuwen/ledger/venv/bin/python
import os
import re
import subprocess
import tempfile
from ofxclient import Institution, CreditCardAccount
DOWNLOAD_DAYS = 5
LEDGER_FILENAME = '/home/fanqiuwen/ledger/journal.ledger'
def get_pass(pass_name):
return subprocess.check_output(['pass', pass_name]).splitlines()[0].decode()
def ofx2csv(account, account_name, days, fout):
try:
trans = account.transactions(days)
except Exception as e:
print("Failed to download for %s: %s" % (account_name, e))
return False
if not trans:
print("No transactions for %s" % account_name)
return False
for tran in sorted(trans, key=lambda x: x.date):
line = ("%s,%s,\"%s\"" % (tran.date.strftime('%Y/%m/%d'),
tran.amount, tran.payee))
if tran.amount == 0:
print("Skipping " + line)
else:
fout.write(line + "\n")
fout.flush()
return True
def csv2ledger(account_name, csv_filename, fout):
with tempfile.NamedTemporaryFile('r') as tmp_ledger:
subprocess.run(['reckon', '-f', csv_filename, '-a', account_name,
'-l', LEDGER_FILENAME, '-o', tmp_ledger.name])
for line in tmp_ledger.readlines():
line = re.sub(r'^\t(\S+)\s+(\$.*)', r' \1\t\2', line)
line = re.sub(r'^\t(\S+)\s+-\$.*', r' \1', line)
fout.write(line)
chase_args = {
'id': '10898',
'org': 'B1',
'url': 'https://ofx.chase.com',
'username': 'REDACTED',
'password': get_pass('sites/chase.com'),
'client_args': {
'id': 'REDACTED', # 32-char UUID that identifies this client
'app_id': 'QWIN',
'app_version': '2200',
'ofx_version': '103'
}
}
discover_args = {
'id': '7101',
'org': 'Discover Financial Services',
'url': 'https://ofx.discovercard.com',
'username': 'REDACTED',
'password': get_pass('sites/discover.com'),
'client_args': {
'app_id': 'QWIN',
'app_version': '2200',
'ofx_version': '102'
}
}
chase_sr = CreditCardAccount(institution=Institution(**chase_args),
number='REDACTED')
discover = CreditCardAccount(institution=Institution(**discover_args),
number='REDACTED')
if __name__ == '__main__':
accounts = [
(chase_sr, 'Liability:ChaseSR'),
(discover, 'Liability:Discover')
]
with open(LEDGER_FILENAME, 'a') as file_ledger:
for account, account_name in accounts:
with tempfile.NamedTemporaryFile('w') as file_csv:
if ofx2csv(account, account_name, DOWNLOAD_DAYS, file_csv):
csv2ledger(account_name, file_csv.name, file_ledger)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment