79 lines
3 KiB
Python
79 lines
3 KiB
Python
#! /usr/bin/env python3
|
|
import mt940
|
|
import csv
|
|
import click
|
|
from datetime import datetime
|
|
from sqlalchemy import create_engine, desc
|
|
from sqlalchemy.orm import sessionmaker
|
|
from models import Transaction
|
|
import sys
|
|
|
|
|
|
@click.command(name="import")
|
|
@click.option("--filetype", "-t", type=click.Choice(["dkb", "sparkasse-mt940", "bunq-csv"], case_sensitive=False))
|
|
@click.argument("filename", type=click.Path(exists=True))
|
|
def command(filename, filetype):
|
|
engine = create_engine("sqlite:///app.db")
|
|
Session = sessionmaker(bind=engine)
|
|
session = Session()
|
|
|
|
latest = session.query(Transaction).order_by(desc(Transaction.date)).first()
|
|
new = []
|
|
if filetype == "sparkasse-mt940":
|
|
transactions = mt940.parse(click.format_filename(filename))
|
|
for t in transactions:
|
|
data = t.data
|
|
|
|
date = data["date"]
|
|
amount = int(data["amount"].amount * 100)
|
|
iban = data["applicant_iban"]
|
|
name = data["applicant_name"]
|
|
description = data["purpose"]
|
|
|
|
# if latest and data['date'] < latest.date:
|
|
# print("Found transaction older than then oldest transction in the DB. Aborting")
|
|
# sys.exit(1)
|
|
|
|
new.append(Transaction(date=date, name=name, iban=iban, amount=amount, description=description))
|
|
print(".", end="", flush=True)
|
|
|
|
elif filetype == "bunq-csv":
|
|
with open(click.format_filename(filename)) as fh:
|
|
fh.readline() # Get rid of first line
|
|
csv_reader = csv.reader(fh, delimiter=";")
|
|
count = 0
|
|
for line in csv_reader:
|
|
date = datetime.strptime(line[0], "%Y-%m-%d")
|
|
amount = int(float(line[2].replace(",", "")) * 100)
|
|
iban = line[4]
|
|
name = line[5]
|
|
description = line[6]
|
|
|
|
# if latest and date < latest.date:
|
|
# print("Found transaction older than then oldest transction in the DB. Aborting")
|
|
# sys.exit(1)
|
|
|
|
new.append(Transaction(date=date, name=name, iban=iban, amount=amount, description=description))
|
|
print(".", end="", flush=True)
|
|
|
|
elif filetype == "dkb":
|
|
with open(click.format_filename(filename), 'r', encoding='utf8') as fh:
|
|
fh.readline() # Get rid of first line
|
|
csv_reader = csv.reader(fh, delimiter=";")
|
|
for line in csv_reader:
|
|
date = datetime.strptime(line[0], "%Y-%m-%d")
|
|
amount = line[5]
|
|
iban = line[4]
|
|
name = line[3]
|
|
description = line[4]
|
|
|
|
# if latest and date < latest.date:
|
|
# print("Found transaction older than then oldest transction in the DB. Aborting")
|
|
# sys.exit(1)
|
|
|
|
new.append(Transaction(date=date, name=name, iban=iban, amount=amount, description=description))
|
|
print(".", end="", flush=True)
|
|
|
|
session.bulk_save_objects(new)
|
|
session.commit()
|
|
print(f"Imported {len(new)} transactions")
|