format-dkb-statement

#! /usr/bin/env nix-shell
#! nix-shell -i python -p python3 -p clevercsv -p python311Packages.babel

import re
import sys
import tempfile
import os.path
from datetime import datetime
from typing import List, Dict
import clevercsv as csv
from babel.numbers import parse_decimal

# TOOD: Change the encoding to latin1 / Windows-1252
# Read from STDIN / STDOUT and or FILE


# Clean the strings of characters that might make the CSV difficult to parse.
# Remove unnecessary formatting
def clean_string(s):
    s = s.strip() # Remove space at the front and end
    s = re.sub(r"\s+", "_", s) # Replace multiple spaces with a single
    s = re.sub(r'[\\\'";,]', "", s) # Remove special characters \ ' " ; ,
    s = re.sub(r"_+", "_", s) # Replace multiple underscores with a single
    s = re.sub(r"-+", "-", s) # Replace mulitple minus with a single
    s = re.sub(r"\.+", ".", s) # Replace mulitple dots with a single

    return s


# Turns 24.12.2002 into 2002-12-24 (ISO 8601)
def transform_date(string: str) -> str:
    if clean_string(string) == "":
        return ""
    else:
        return datetime.strptime(string, "%d.%m.%y").strftime("%Y-%m-%d")


def transform_transaction_dict(data_dict):
    result = {}

    result["booking_date"] = transform_date(data_dict.get("Buchungsdatum"))
    result["value_date"] = transform_date(data_dict.get("Wertstellung"))
    result["status"] = clean_string(data_dict.get("Status"))
    result["payer"] = clean_string(data_dict.get("Zahlungspflichtige*r"))
    result["payee"] = clean_string(data_dict.get("Zahlungsempfänger*in"))
    result["purpose"] = clean_string(data_dict.get("Verwendungszweck"))
    result["type"] = clean_string(data_dict.get("Umsatztyp"))
    result["iban"] = clean_string(data_dict.get("IBAN"))
    result["amount_eur"] = "%.2f" % parse_decimal(
        data_dict.get("Betrag (€)"), locale="de"
    )
    result["creditor_id"] = clean_string(data_dict.get("Gläubiger-ID"))
    result["mandate_reference"] = clean_string(data_dict.get("Mandatsreferenz"))
    result["customer_reference"] = clean_string(data_dict.get("Kundenreferenz"))

    return result

def read_from_file(file_path: str) -> List[Dict[str, str]]:
    with open(file_path, "r", newline="") as fp:
        # Skip the first 4 lines
        for _ in range(4):
            next(fp)

        # Create a temporary file and write the remaining lines
        with tempfile.NamedTemporaryFile(mode="w+") as temp:
            temp.write(fp.read())
            return list(csv.wrappers.read_dicts(temp.name))

def write_to_file(transactions: List[Dict[str, str]], filename: str):
    if not transactions:
        print("List is empty. Cannot export to CSV.")
        return

    with open(filename, "w", newline="") as csvfile:
        # Extract fieldnames from the keys of the first entry
        fieldnames: List[str] = list(transactions[0].keys())
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(transactions)

def write_to_stdout(transactions: List[Dict[str, str]]):
    if not transactions:
        print("List is empty. Cannot print CSV.")
        return

    # Extract fieldnames from the keys of the first entry
    fieldnames: List[str] = list(transactions[0].keys())
    writer = csv.DictWriter(sys.stdout, fieldnames=fieldnames)
    writer.writeheader()
    writer.writerows(transactions)

def main():
    # Check if the file path is provided as an argument
    if (len(sys.argv) != 2 and len(sys.argv) != 3) or sys.argv[1] == "-h" or sys.argv[1] == "--help":
        print("Format a CSV statement from 'Deutsche Kredit Bank' so it is easier to use with accounting software")
        print("Usage: format-dkb-statement <input_path> [<output_path>]")
        print("If <output_path> is not provided, result is written to STDOUT")
        return

    # Extract the file path from command-line arguments
    input_path = sys.argv[1]

    # Check if the file exists
    if not os.path.exists(input_path):
        print(f"The file '{input_path}' does not exist.")
        return

    transactions: List[Dict[str, str]] = list(
        map(
            lambda t: transform_transaction_dict(t),
            read_from_file(input_path)
        ),
    )

    if len(sys.argv) == 2:
        write_to_stdout(transactions)
    else:
        output_path = sys.argv[2]
        write_to_file(transactions, output_path)

if __name__ == "__main__":
    main()