forked from gaugendre/ofx-processor
111 lines
3 KiB
Python
111 lines
3 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/--no-ynab",
|
|
"push_to_ynab",
|
|
default=True,
|
|
help="Push data directly to YNAB.",
|
|
show_default=True,
|
|
)
|
|
@click.option(
|
|
"--file/--no-file",
|
|
"output_file",
|
|
default=False,
|
|
help="Write a processed file.",
|
|
show_default=True,
|
|
)
|
|
def cli(ofx_filename, push_to_ynab, output_file):
|
|
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")
|
|
|
|
if output_file:
|
|
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()
|