ofx-processor/ofx_processor/revolut_processor/main.py

69 lines
1.6 KiB
Python
Raw Normal View History

2020-01-16 19:02:17 +01:00
import csv
import os
import click
import dateparser
def process_amount(amount):
if amount:
return float(amount.replace(",", "."))
return ""
def process_memo(line):
return " - ".join(
2020-01-17 16:18:47 +01:00
filter(
None,
map(str.strip, [line.get("Category", ""), line.get("Exchange Rate", "")]),
)
2020-01-16 19:02:17 +01:00
)
2020-01-17 16:18:47 +01:00
def process_date(line):
return dateparser.parse(line.get("Completed Date")).strftime("%Y-%m-%d")
def process_inflow(line):
return process_amount(line.get("Paid In (EUR)"))
def process_outflow(line):
return process_amount(line.get("Paid Out (EUR)"))
2020-01-16 19:02:17 +01:00
@click.command()
@click.argument("csv_filename")
def cli(csv_filename):
formatted_data = []
with open(csv_filename) as f:
reader = csv.DictReader(f, delimiter=";")
for line in reader:
formatted_data.append(
{
2020-01-17 16:18:47 +01:00
"Date": process_date(line),
2020-01-16 19:02:17 +01:00
"Payee": line["Reference"],
"Memo": process_memo(line),
2020-01-17 16:18:47 +01:00
"Outflow": process_outflow(line),
"Inflow": process_inflow(line),
2020-01-16 19:02:17 +01:00
}
)
if not formatted_data:
click.secho("Nothing to write.")
processed_file = os.path.join(os.path.dirname(csv_filename), "processed.csv")
with open(processed_file, "w") as f:
writer = csv.DictWriter(
f, delimiter=",", quotechar='"', fieldnames=formatted_data[0].keys()
)
writer.writeheader()
writer.writerows(formatted_data)
click.secho("{} written".format(processed_file), fg="green")
if __name__ == "__main__":
cli()