ING/categorize_transactions.py

#!/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}")