82 lines
3.2 KiB
Python
82 lines
3.2 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
|
|
from helper import get_session
|
|
import sys
|
|
|
|
|
|
@click.command(name="import")
|
|
@click.option("--filetype", "-t", type=click.Choice(["dkb", "sparkasse-mt940", "bunq-csv"], case_sensitive=False))
|
|
@click.option("--profile", "-p")
|
|
@click.option("--force", "-f", default=False, is_flag=True)
|
|
@click.argument("filename", type=click.Path(exists=True))
|
|
def command(filename, profile, force, filetype):
|
|
session = get_session(profile)
|
|
|
|
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 not force and 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=";")
|
|
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 not force and 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), encoding="ISO-8859-1") as fh:
|
|
csv_reader = csv.reader(fh, delimiter=";")
|
|
next(csv_reader, None)
|
|
for line in csv_reader:
|
|
date = datetime.strptime(line[0], "%d.%m.%Y")
|
|
amount = int(line[7].replace(".", "").replace(",", ""))
|
|
iban = line[5]
|
|
name = line[3]
|
|
description = line[4]
|
|
|
|
if description == "Tagessaldo":
|
|
continue
|
|
|
|
if not force and 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")
|