toldg now uses Python build-in templating, not jinja2\ngetofx uses ofx library for extracting transactions
This commit is contained in:
177
getofx.py
177
getofx.py
@@ -1,94 +1,137 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import datetime
|
||||
import ofxtools
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import csv
|
||||
import xml.etree.ElementTree as ET
|
||||
import json
|
||||
import logging
|
||||
import datetime
|
||||
import ofxtools
|
||||
|
||||
from typing import List
|
||||
from dataclasses import dataclass
|
||||
from ofxtools import OFXClient
|
||||
from ofxtools.Client import StmtRq, CcStmtEndRq, CcStmtRq
|
||||
from functools import namedtuple
|
||||
from ofxtools.Parser import OFXTree
|
||||
|
||||
|
||||
def get_transactions(data):
|
||||
Transaction = namedtuple("Transaction",
|
||||
["details", "date", "description",
|
||||
"amount", "type", "balance", "slip"])
|
||||
root = ET.fromstring(data)
|
||||
ts = []
|
||||
for statement in root.iter("STMTTRN"):
|
||||
description, date, amount = "", "", ""
|
||||
for child in statement:
|
||||
if child.tag == "TRNAMT":
|
||||
amount = child.text
|
||||
elif child.tag == "DTPOSTED":
|
||||
d = datetime.datetime.strptime(child.text[:8], "%Y%m%d")
|
||||
date = d.strftime("%m/%d/%Y")
|
||||
elif child.tag == "NAME":
|
||||
if description:
|
||||
description = child.text + " " + description
|
||||
else:
|
||||
description = child.text
|
||||
elif child.tag == "MEMO":
|
||||
if description:
|
||||
description = description + " " + child.text
|
||||
else:
|
||||
description = child.text
|
||||
t = Transaction("-", date, description, amount, "-", "-", "-")
|
||||
ts.append(t)
|
||||
@dataclass
|
||||
class ClientConfig:
|
||||
url: str
|
||||
userid: str
|
||||
org: str
|
||||
clientuid: str
|
||||
fid: str
|
||||
bankid: str
|
||||
version: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class AccountConfig:
|
||||
name: str
|
||||
accttype: str
|
||||
acctid: str
|
||||
csv_file: str
|
||||
fields: List[str]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Config:
|
||||
"""
|
||||
Basic class for abstracting the configuration.
|
||||
"""
|
||||
secret: str
|
||||
client: ClientConfig
|
||||
accounts: List[AccountConfig]
|
||||
|
||||
@dataclass
|
||||
class Transaction:
|
||||
date: str
|
||||
description: str
|
||||
amount: str
|
||||
|
||||
|
||||
def get_transactions(client: OFXClient, secret: str, account: AccountConfig):
|
||||
dtstart = datetime.datetime(2020, 1, 1, tzinfo=ofxtools.utils.UTC)
|
||||
dtend = datetime.datetime(2020, 12, 31, tzinfo=ofxtools.utils.UTC)
|
||||
|
||||
if account.accttype.upper() in ("CHECKING", "SAVINGS"):
|
||||
rq = StmtRq(acctid=account.acctid, accttype=account.accttype.upper(),
|
||||
dtstart=dtstart, dtend=dtend)
|
||||
else:
|
||||
rq = CcStmtRq(acctid=account.acctid, dtstart=dtstart, dtend=dtend)
|
||||
response = client.request_statements(secret, rq)
|
||||
parser = OFXTree()
|
||||
parser.parse(response)
|
||||
ofx = parser.convert()
|
||||
ts = [Transaction(t.dtposted.strftime("%m/%d/%Y"),
|
||||
t.name + " " + t.memo if t.memo else t.name,
|
||||
str(t.trnamt))
|
||||
for t in ofx.statements[0].banktranlist]
|
||||
return ts
|
||||
|
||||
|
||||
def process_account(client, secret, year, name, accttype, acctid, csv_file):
|
||||
dtstart = datetime.datetime(int(year), 1, 1, tzinfo=ofxtools.utils.UTC)
|
||||
dtend = datetime.datetime(int(year), 12, 31, tzinfo=ofxtools.utils.UTC)
|
||||
def write_csv(account: AccountConfig, transactions: List[Transaction]):
|
||||
|
||||
if accttype.upper() in ("CHECKING", "SAVINGS"):
|
||||
rq = StmtRq(acctid=acctid, accttype=accttype.upper(),
|
||||
dtstart=dtstart, dtend=dtend)
|
||||
def transaction_to_csv_row(t: Transaction) -> List[str]:
|
||||
""" This allows to user to specify how to order the fields in the CSV
|
||||
file. I have implemented this feature because the columns in my
|
||||
checking account and in my credit card accounts are different. If the
|
||||
field is one of 'date', 'description', or 'amount' we get that
|
||||
attribute from the transaction. Otherwise, we use the field itself
|
||||
(usually an empty string in my case). """
|
||||
return [getattr(t, f) if hasattr(t, f) else f
|
||||
for f in account.fields]
|
||||
|
||||
status = "no change"
|
||||
|
||||
csv_file = account.csv_file
|
||||
if not os.path.isfile(csv_file):
|
||||
status = "new"
|
||||
with open(account.csv_file, "w") as f:
|
||||
csv_writer = csv.writer(f)
|
||||
csv_writer.writerow(["date", "description", "amount"])
|
||||
for t in transactions:
|
||||
r = transaction_to_csv_row(t)
|
||||
csv_writer.writerow(r)
|
||||
else:
|
||||
rq = CcStmtRq(acctid=acctid, dtstart=dtstart, dtend=dtend)
|
||||
# TODO: diff rows and append only the new once.
|
||||
pass
|
||||
|
||||
response = client.request_statements(secret, rq)
|
||||
data = response.read().decode()
|
||||
# with open(csv_file.replace(".csv", ".xml"), "w") as f:
|
||||
# f.write(data)
|
||||
transactions = get_transactions(data)
|
||||
|
||||
with open(csv_file, "w") as f:
|
||||
csv_writer = csv.writer(f)
|
||||
csv_writer.writerow(["details", "date", "description",
|
||||
"amount", "type", "balance", "slip"])
|
||||
for t in transactions:
|
||||
csv_writer.writerow(t)
|
||||
#if t.date.startswith(year):
|
||||
logging.warning(f"{account.name:30} -> {account.csv_file:30} | {status}")
|
||||
|
||||
|
||||
def get_client(url, userid, org, fid, clientuid, bankid, version, **kwargs):
|
||||
return OFXClient(url, userid=userid, org=org, fid=fid,
|
||||
clientuid=clientuid, bankid=bankid, version=version,
|
||||
prettyprint=True)
|
||||
|
||||
def get_client(c: ClientConfig) -> OFXClient:
|
||||
return OFXClient(c.url, userid=c.userid, org=c.org, fid=c.fid,
|
||||
clientuid=c.clientuid, bankid=c.bankid,
|
||||
version=c.version, prettyprint=True)
|
||||
|
||||
|
||||
def main(config):
|
||||
client = get_client(**config["client"])
|
||||
year = config["year"]
|
||||
secret = config["secret"]
|
||||
for account in config["accounts"]:
|
||||
name = account["name"]
|
||||
logging.info(f"Processing {name}.")
|
||||
process_account(client, secret, year, **account)
|
||||
def parse_config(config_file: str) -> Config:
|
||||
with open(config_file, 'r') as f:
|
||||
# We could use the dacite package if the configureation
|
||||
# gets more complex and for automatical type ckecking, but
|
||||
# probably not worth it as this point.
|
||||
config = Config(**json.load(f))
|
||||
config.client = ClientConfig(**config.client)
|
||||
config.accounts = [AccountConfig(**a) for a in config.accounts]
|
||||
return config
|
||||
|
||||
|
||||
def main(config: Config):
|
||||
client = get_client(config.client)
|
||||
for account in config.accounts:
|
||||
transactions = get_transactions(client, config.secret, account)
|
||||
write_csv(account, transactions)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
logging.basicConfig(level=logging.WARNING, format='%(message)s')
|
||||
try:
|
||||
config_file = sys.argv[1]
|
||||
except IndexError:
|
||||
config_file = "gather.json"
|
||||
with open(config_file, 'r') as f:
|
||||
config = json.load(f)
|
||||
config_file = "getofx.json"
|
||||
config = parse_config(config_file)
|
||||
main(config)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user