121 lines
3.2 KiB
Python
121 lines
3.2 KiB
Python
#! /usr/bin/env python3
|
|
import locale
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
import yaml
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from xdg import XDG_CONFIG_HOME, XDG_DATA_HOME
|
|
|
|
import schmeckels.models
|
|
from schmeckels import banks, models
|
|
|
|
try:
|
|
from yaml import CLoader as Loader
|
|
except ImportError:
|
|
from yaml import Loader
|
|
|
|
|
|
def format_amount(d):
|
|
if type(d) is dict:
|
|
for k, v in d.items():
|
|
amount = v / 100
|
|
d[k] = f"{amount:,.2f}"
|
|
return d
|
|
elif type(d) is int:
|
|
amount = d / 100
|
|
return f"{amount:,.2f}"
|
|
else:
|
|
print(d)
|
|
raise Exception()
|
|
|
|
|
|
def get_data_dir():
|
|
fallback_dir = os.path.join(XDG_DATA_HOME, "schmeckels")
|
|
return os.getenv("SCHMECKELS_DIR", fallback_dir)
|
|
|
|
|
|
def build_database_filename(base_path):
|
|
return f"{base_path}/schmeckels.db"
|
|
|
|
|
|
def build_rules_filename(base_path):
|
|
return f"{base_path}/rules.yaml"
|
|
|
|
|
|
def get_session():
|
|
data_dir = get_data_dir()
|
|
filename = build_database_filename(data_dir)
|
|
if os.path.exists(filename) and os.path.isfile(filename):
|
|
engine = create_engine(f"sqlite:///{filename}")
|
|
Session = sessionmaker(bind=engine)
|
|
return Session()
|
|
else:
|
|
print(f"No database found in {data_dir}! Did you run 'init'?")
|
|
sys.exit(1)
|
|
|
|
|
|
def get_rules():
|
|
data_dir = get_data_dir()
|
|
filename = build_rules_filename(data_dir)
|
|
if os.path.exists(filename) and os.path.isfile(filename):
|
|
with open(filename) as fh:
|
|
data = yaml.load(fh, Loader=Loader)
|
|
if data:
|
|
for rule in data:
|
|
name = rule.get("name", None)
|
|
description = rule.get("description", None)
|
|
|
|
if name == "" or description == "":
|
|
print("Found invalid rule with empty name or description string!")
|
|
print(rule)
|
|
sys.exit(1)
|
|
|
|
if name:
|
|
rule["name_regex"] = re.compile(rule["name"])
|
|
if description:
|
|
rule["desc_regex"] = re.compile(rule["description"])
|
|
else:
|
|
data = []
|
|
return data
|
|
else:
|
|
print(f"No rules.yaml found in {data_dir}! Did you run 'init'?")
|
|
sys.exit(1)
|
|
|
|
|
|
def create_tag(name, session):
|
|
t = session.query(models.Tag).filter(models.Tag.name == name).first()
|
|
if not t:
|
|
t = models.Tag(name=name)
|
|
session.add(t)
|
|
session.commit()
|
|
return t
|
|
|
|
|
|
def import_transactions(session, filetype, filename):
|
|
if filetype not in [b[0] for b in banks.SUPPORTED_BANKS]:
|
|
raise ValueError("Unsupported filetype")
|
|
|
|
new = []
|
|
|
|
if filetype == "sparkasse-mt940":
|
|
bank = banks.Sparkasse_MT940(filename)
|
|
elif filetype == "bunq-csv":
|
|
bank = banks.Bunq(filename)
|
|
elif filetype == "dkb":
|
|
bank = banks.DKB(filename)
|
|
|
|
for t in bank.get_transactions():
|
|
if (
|
|
not session.query(models.Transaction)
|
|
.filter_by(iban=t.iban)
|
|
.filter_by(amount=t.amount)
|
|
.filter_by(date=t.date)
|
|
.filter_by(description=t.description)
|
|
.first()
|
|
):
|
|
new.append(t)
|
|
|
|
return new
|