47 lines
1.4 KiB
Python
47 lines
1.4 KiB
Python
#! /usr/bin/env python3
|
|
import mt940
|
|
import click
|
|
from sqlalchemy import create_engine, desc
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import sessionmaker
|
|
from models import Transaction
|
|
import sys
|
|
|
|
|
|
def convert_amount(t):
|
|
return int(t.amount * 100)
|
|
|
|
@click.command(name='import')
|
|
@click.argument('filename', type=click.Path(exists=True))
|
|
def command(filename):
|
|
click.echo(click.format_filename(filename))
|
|
engine = create_engine("sqlite:///app.db")
|
|
Base = declarative_base()
|
|
|
|
Session = sessionmaker(bind=engine)
|
|
session = Session()
|
|
|
|
latest = session.query(Transaction).order_by(desc(Transaction.date)).first()
|
|
|
|
transactions = mt940.parse("2020-01.TXT")
|
|
count = 0
|
|
for t in transactions:
|
|
data = t.data
|
|
|
|
date = data['date']
|
|
amount = convert_amount(data['amount'])
|
|
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)
|
|
|
|
t = Transaction(date=date, name=name, iban=iban, amount=amount, description=description)
|
|
session.add(t)
|
|
session.commit()
|
|
count += 1
|
|
print(".", end="", flush=True)
|
|
print()
|
|
print(f"Imported {count} transactions")
|