generated from felixm/defaultpy
100 lines
3.6 KiB
Python
100 lines
3.6 KiB
Python
import csv
|
|
import datetime
|
|
import logging
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
import toldg.models
|
|
import toldg.predict
|
|
import toldg.utils
|
|
import toldg.write
|
|
from toldg.models import Config, CsvConfig, Mapping, Transaction
|
|
|
|
|
|
def process_ldg_files(config: Config):
|
|
with open(config.output_file, "a") as f_out:
|
|
for ldg_file in toldg.utils.get_ldg_files(config.input_directory):
|
|
ldg_rel = os.path.relpath(ldg_file, os.path.dirname(config.output_file))
|
|
f_out.write(f'include "{ldg_rel}"\n')
|
|
|
|
|
|
def get_csv_config(csv_file: str, csv_configs: list[CsvConfig]) -> CsvConfig:
|
|
cs = [c for c in csv_configs if re.match(c.file_match_regex, csv_file)]
|
|
if not cs:
|
|
logging.critical(f"No CSV config for {csv_file}.")
|
|
sys.exit(1)
|
|
elif len(cs) > 1:
|
|
logging.critical(f"Multiple CSV configs for {csv_file}.")
|
|
sys.exit(1)
|
|
return cs[0]
|
|
|
|
|
|
def get_transactions(csv_file: str, config: CsvConfig) -> list[Transaction]:
|
|
def date_to_date(date: str) -> str:
|
|
d = datetime.datetime.strptime(date, config.input_date_format)
|
|
return d.strftime(config.output_date_format)
|
|
|
|
def flip_sign(amount: str) -> str:
|
|
return amount[1:] if amount.startswith("-") else "-" + amount
|
|
|
|
def row_to_transaction(idx, row, fields):
|
|
"""The user can configure the mapping of CSV fields to the three
|
|
required fields date, amount and description via the CsvConfig."""
|
|
t = {field: row[index] for index, field in fields}
|
|
amount = t["amount"]
|
|
return Transaction(
|
|
currency=config.currency,
|
|
debit=flip_sign(amount),
|
|
credit=amount,
|
|
date=date_to_date(t["date"]),
|
|
account1=config.account1,
|
|
account2=toldg.models.UNKNOWN_CATEGORY,
|
|
description=t["description"],
|
|
csv_file=csv_file,
|
|
row=", ".join(row),
|
|
index=idx,
|
|
)
|
|
|
|
fields = [(i, f) for i, f in enumerate(config.fields) if f]
|
|
with open(csv_file, "r") as f:
|
|
reader = csv.reader(f, delimiter=config.delimiter, quotechar=config.quotechar)
|
|
for _ in range(config.skip):
|
|
next(reader)
|
|
rows = [row for row in reader if row]
|
|
transactions = [row_to_transaction(i, row, fields) for i, row in enumerate(reversed(rows))]
|
|
return transactions
|
|
|
|
|
|
def apply_mappings(transactions: list[Transaction], mappings: dict[str, Mapping]):
|
|
"""Apply mappings to transactions."""
|
|
for t in transactions:
|
|
if t.key() in mappings:
|
|
mapping = mappings[t.key()]
|
|
assert isinstance(mapping, Mapping)
|
|
assert mapping.count > 0, f"{mapping} used by {t} but count is not greater than '0'."
|
|
mapping.count -= 1
|
|
t.mapping = mapping
|
|
else:
|
|
logging.warning(f"No mapping for '{t}'.")
|
|
|
|
for mapping in mappings.values():
|
|
assert mapping.count == 0, f"{mapping} was not used as often as expected!"
|
|
|
|
|
|
def process_csv_files(config: Config, write_outputs: bool = True) -> list[Transaction]:
|
|
csv_files = toldg.utils.get_csv_files(config.input_directory)
|
|
transactions = []
|
|
for csv_file in csv_files:
|
|
csv_file = str(csv_file)
|
|
csv_config = get_csv_config(csv_file, config.csv_configs)
|
|
transactions += get_transactions(csv_file, csv_config)
|
|
|
|
mappings = toldg.utils.read_mappings(config.mappings_file)
|
|
apply_mappings(transactions, mappings)
|
|
toldg.predict.add_account2(config.model, transactions, config.categories)
|
|
if write_outputs:
|
|
toldg.utils.write_mappings(transactions, config.mappings_file)
|
|
toldg.write.render_to_file(transactions, config)
|
|
return transactions
|