ofx-processor/ofx_processor/bpvf_processor/main.py
2020-02-08 19:13:01 +01:00

101 lines
2.8 KiB
Python

import os
import re
import sys
from collections import defaultdict
from xml.etree import ElementTree
import click
from ofxtools.Parser import OFXTree
from ofxtools.header import make_header
from ofx_processor.utils import ynab
def _process_name_and_memo(name, memo):
if "CB****" in name:
conversion = re.compile(r"\d+,\d{2}[a-zA-Z]{3}")
match = conversion.search(memo)
if match:
res_name = memo[: match.start() - 1]
res_memo = name + memo[match.start() - 1 :]
else:
res_name = memo
res_memo = name
return res_name, res_memo, True
return name, memo, False
def process_name_and_memo(transaction):
return _process_name_and_memo(transaction.name, transaction.memo)
@click.command()
@click.argument("ofx_filename")
@click.option(
"--ynab/--file-only",
"push_to_ynab",
default=False,
help="Push data directly to YNAB instead of just writing a file.",
)
def cli(ofx_filename, push_to_ynab):
parser = OFXTree()
try:
parser.parse(ofx_filename)
except FileNotFoundError:
click.secho("Couldn't open ofx file", fg="red")
sys.exit(1)
ofx = parser.convert()
if ofx is None:
click.secho("Couldn't parse ofx file", fg="red")
sys.exit(1)
ynab_transactions = []
transaction_ids = defaultdict(int)
for transaction in ofx.statements[0].transactions:
transaction.name, transaction.memo, edited = process_name_and_memo(transaction)
if edited:
click.secho(
"Edited transaction {} ({})".format(
transaction.checknum, transaction.name
),
fg="blue",
)
date = transaction.dtposted.isoformat().split("T")[0]
amount = int(transaction.trnamt * 1000)
import_id = f"YNAB:{amount}:{date}"
transaction_ids[import_id] += 1
occurrence = transaction_ids[import_id]
import_id = f"{import_id}:{occurrence}"
ynab_transactions.append(
{
"date": date,
"amount": amount,
"payee_name": transaction.name,
"memo": transaction.memo,
"import_id": import_id,
}
)
click.secho(f"Processed {len(ynab_transactions)} transactions total.", fg="blue")
header = str(make_header(version=102))
root = ofx.to_etree()
data = ElementTree.tostring(root).decode()
processed_file = os.path.join(os.path.dirname(ofx_filename), "processed.ofx")
with open(processed_file, "w") as f:
f.write(header + data)
click.secho("{} written".format(processed_file), fg="green")
if push_to_ynab:
ynab.push_transactions(ynab_transactions, "bpvf")
if __name__ == "__main__":
cli()