diff --git a/.gitignore b/.gitignore index 3004ef6..352cc03 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ .ing.conf **/logs/ -**/Transaktionen/ \ No newline at end of file +**/Transaktionen/ +Kategorien.json \ No newline at end of file diff --git a/categorize_transactions.py b/categorize_transactions.py new file mode 100644 index 0000000..dee4a6c --- /dev/null +++ b/categorize_transactions.py @@ -0,0 +1,242 @@ +#!/usr/bin/env python3 + +from pathlib import Path +from datetime import datetime +import json +import csv +import sys + +# -------------------------------------------------- +# Kalenderwoche bestimmen +# -------------------------------------------------- + +if len(sys.argv) == 3: + year = int(sys.argv[1]) + week = int(sys.argv[2]) +else: + year, week, _ = datetime.now().isocalendar() + +# -------------------------------------------------- +# Kategorien laden +# -------------------------------------------------- + +categories_file = Path.home() / "Kategorien.json" + +if not categories_file.exists(): + categories_file = Path(__file__).parent / "Kategorien.json" + +with open(categories_file, encoding="utf-8") as f: + categories = json.load(f) + +# -------------------------------------------------- +# Verzeichnisse bestimmen +# -------------------------------------------------- + +base_dir = Path.home() / "Transaktionen" + +if not base_dir.exists(): + base_dir = Path(__file__).parent / "Transaktionen" + +year_dir = base_dir / str(year) + +source_json = ( + year_dir + / "json" + / f"transactions_{year}_KW{week:02d}.json" +) + +categorized_csv_dir = year_dir / "categorized_csv" +categorized_json_dir = year_dir / "categorized_json" +summary_dir = year_dir / "summary" + +categorized_csv_dir.mkdir(parents=True, exist_ok=True) +categorized_json_dir.mkdir(parents=True, exist_ok=True) +summary_dir.mkdir(parents=True, exist_ok=True) + +csv_file = ( + categorized_csv_dir + / f"categorized_transactions_{year}_KW{week:02d}.csv" +) + +json_file = ( + categorized_json_dir + / f"categorized_transactions_{year}_KW{week:02d}.json" +) + +summary_file = ( + summary_dir + / f"category_summary_{year}_KW{week:02d}.json" +) + +# -------------------------------------------------- +# Prüfen ob Quelldatei existiert +# -------------------------------------------------- + +if not source_json.exists(): + print(f"Datei nicht gefunden: {source_json}") + sys.exit(1) + +# -------------------------------------------------- +# Transaktionen laden +# -------------------------------------------------- + +with open(source_json, encoding="utf-8") as f: + transactions = json.load(f) + +# -------------------------------------------------- +# Kategorien vorbereiten +# -------------------------------------------------- + +category_totals = {} + +for category in categories.keys(): + category_totals[category] = { + "income": 0.0, + "expenses": 0.0, + "saldo": 0.0 + } + +category_totals["Unbekannt"] = { + "income": 0.0, + "expenses": 0.0, + "saldo": 0.0 +} + +categorized_transactions = [] + +# -------------------------------------------------- +# Transaktionen kategorisieren +# -------------------------------------------------- + +for transaction in transactions: + + search_text = ( + f"{transaction.get('applicant_name', '')} " + f"{transaction.get('purpose', '')} " + f"{transaction.get('posting_text', '')}" + ).lower() + + category_found = "Unbekannt" + + for category, keywords in categories.items(): + + found = False + + for keyword in keywords: + + if keyword.lower() in search_text: + category_found = category + found = True + break + + if found: + break + + categorized_transaction = transaction.copy() + categorized_transaction["category"] = category_found + + categorized_transactions.append(categorized_transaction) + + amount = float(transaction.get("amount", 0)) + + if amount >= 0: + category_totals[category_found]["income"] += amount + else: + category_totals[category_found]["expenses"] += abs(amount) + + category_totals[category_found]["saldo"] += amount + +# -------------------------------------------------- +# Kategorien runden +# -------------------------------------------------- + +for category in category_totals: + + category_totals[category]["income"] = round( + category_totals[category]["income"], + 2 + ) + + category_totals[category]["expenses"] = round( + category_totals[category]["expenses"], + 2 + ) + + category_totals[category]["saldo"] = round( + category_totals[category]["saldo"], + 2 + ) + +# -------------------------------------------------- +# CSV exportieren +# -------------------------------------------------- + +with open(csv_file, "w", newline="", encoding="utf-8") as csvfile: + + writer = csv.DictWriter( + csvfile, + fieldnames=[ + "date", + "amount", + "category", + "posting_text", + "applicant_name", + "purpose" + ], + delimiter=";" + ) + + writer.writeheader() + writer.writerows(categorized_transactions) + +# -------------------------------------------------- +# JSON exportieren +# -------------------------------------------------- + +with open(json_file, "w", encoding="utf-8") as outfile: + + json.dump( + categorized_transactions, + outfile, + ensure_ascii=False, + indent=2 + ) + +# -------------------------------------------------- +# Kategorien-Summen exportieren +# -------------------------------------------------- + +with open(summary_file, "w", encoding="utf-8") as outfile: + + json.dump( + category_totals, + outfile, + ensure_ascii=False, + indent=2 + ) + +# -------------------------------------------------- +# Ausgabe +# -------------------------------------------------- + +print(f"Jahr : {year}") +print(f"Kalenderwoche : {week}") +print(f"Transaktionen : {len(categorized_transactions)}") +print() + +print("Summen pro Kategorie:") +print() + +for category, values in sorted(category_totals.items()): + + print( + f"{category:20} " + f"Einnahmen: {values['income']:10.2f} EUR " + f"Ausgaben: {values['expenses']:10.2f} EUR " + f"Saldo: {values['saldo']:10.2f} EUR" + ) + +print() +print(f"CSV : {csv_file}") +print(f"JSON : {json_file}") +print(f"Summary : {summary_file}") \ No newline at end of file