ING/transactions.py

#!/usr/bin/env python3

from pathlib import Path
from datetime import date, timedelta, datetime
from urllib.request import urlopen
from fints.client import FinTS3PinTanClient
import csv
import json
import sys
import logging
import platform
from pathlib import Path

if platform.system() == "Windows":
    log_dir = Path(__file__).parent / "logs"
else:
    log_dir = Path.home() / "logs"

log_dir.mkdir(exist_ok=True)

log_file = log_dir / "ing.log"

logging.basicConfig(
    filename=log_file,
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    force=True
)

logging.info("[transactions] === Scriptstart ===")

# --------------------------------------------------
# Konfiguration laden
# --------------------------------------------------

config = {}

config_file = Path.home() / ".ing.conf"

if not config_file.exists():
    config_file = Path(__file__).parent / ".ing.conf"

with open(config_file, encoding="utf-8") as f:
    for line in f:
        line = line.strip()

        if not line or line.startswith("#"):
            continue

        if "=" not in line:
            continue

        key, value = line.split("=", 1)
        config[key] = value

# --------------------------------------------------
# ING-Konfiguration
# --------------------------------------------------

USER = config["ING_USER"]
PIN = config["ING_PIN"]
TARGET_IBAN = config["ING_IBAN"]

# --------------------------------------------------
# ioBroker-Konfiguration
# --------------------------------------------------

IOBROKER_HOST = config["IOBROKER_HOST"]
IOBROKER_PORT = config["IOBROKER_PORT"]

IOBROKER_DP_TRANSACTIONS = config["IOBROKER_DP_TRANSACTIONS"]
IOBROKER_DP_WEEKSUM = config["IOBROKER_DP_WEEKSUM"]
IOBROKER_DP_WEEKEXPENSES = config["IOBROKER_DP_WEEKEXPENSES"]

# --------------------------------------------------
# Bank
# --------------------------------------------------

BLZ = "50010517"

# --------------------------------------------------
# Exportverzeichnis bestimmen
# --------------------------------------------------

# --------------------------------------------------
# Exportverzeichnis bestimmen
# --------------------------------------------------

base_dir = Path(__file__).parent / "Transaktionen"
# --------------------------------------------------
# Kalenderwoche bestimmen
# --------------------------------------------------

if len(sys.argv) == 3:

    year = int(sys.argv[1])
    week = int(sys.argv[2])

else:

    year, week, _ = datetime.now().isocalendar()

# Montag der Kalenderwoche

start_date = datetime.strptime(
    f"{year}-W{week:02d}-1",
    "%G-W%V-%u"
).date()

# Sonntag der Kalenderwoche

end_date = start_date + timedelta(days=6)
today = date.today()

if end_date > today:
    end_date = today

if start_date > today:
    logging.error(
        "[transactions] Die angeforderte "
        "Kalenderwoche liegt vollständig "
        "in der Zukunft."
    )

    sys.exit(1)

year_dir = base_dir / str(year)
csv_dir = year_dir / "csv"
json_dir = year_dir / "json"

csv_dir.mkdir(parents=True, exist_ok=True)
json_dir.mkdir(parents=True, exist_ok=True)

CSV_FILE = csv_dir / f"transactions_{year}_KW{week:02d}.csv"
JSON_FILE = json_dir / f"transactions_{year}_KW{week:02d}.json"

# --------------------------------------------------
# ING-Verbindung
# --------------------------------------------------

client = FinTS3PinTanClient(
    BLZ,
    USER,
    PIN,
    "https://fints.ing.de/fints/",
    product_id="PythonFinTS"
)

export_data = []
transactions = []

# --------------------------------------------------
# Transaktionen abrufen
# --------------------------------------------------

with client:
    account_found = False

    for account in client.get_sepa_accounts():

        if account.iban != TARGET_IBAN:
            continue

        account_found = True

        transactions = client.get_transactions(
            account,
            start_date=start_date,
            end_date=end_date
        )

        logging.info(
            f"[transactions] Lade Transaktionen für KW{week:02d}/{year}"
        )

        logging.info(
            f"[transactions] Zeitraum: {start_date} bis {end_date}"
        )

        for transaction in transactions:

            data = transaction.data
            amount = data.get("amount")

            row = {
                "date": str(data.get("date", "")),
                "amount": round(float(amount.amount), 2) if amount else 0.0,
                "posting_text": data.get("posting_text", ""),
                "applicant_name": data.get("applicant_name", ""),
                "purpose": data.get("purpose", "")
            }

            export_data.append(row)

        break

    if not account_found:
        raise RuntimeError(
            f"IBAN nicht gefunden: {TARGET_IBAN}"
        )

# --------------------------------------------------
# CSV schreiben
# --------------------------------------------------

with open(CSV_FILE, "w", newline="", encoding="utf-8") as csvfile:

    writer = csv.DictWriter(
        csvfile,
        fieldnames=[
            "date",
            "amount",
            "posting_text",
            "applicant_name",
            "purpose"
        ],
        delimiter=";"
    )

    writer.writeheader()
    writer.writerows(export_data)

# --------------------------------------------------
# JSON schreiben
# --------------------------------------------------

with open(JSON_FILE, "w", encoding="utf-8") as jsonfile:

    json.dump(
        export_data,
        jsonfile,
        ensure_ascii=False,
        indent=2
    )

# --------------------------------------------------
# Statistiken berechnen
# --------------------------------------------------

transaction_count = len(export_data)

week_sum = 0.0
week_expenses = 0.0

for transaction in transactions:

    amount = float(transaction.data["amount"].amount)

    week_sum += amount

    if amount < 0:
        week_expenses += abs(amount)

week_sum = round(week_sum, 2)
week_expenses = round(week_expenses, 2)

# --------------------------------------------------
# An ioBroker senden
# --------------------------------------------------

base_url = f"http://{IOBROKER_HOST}:{IOBROKER_PORT}/set"

urlopen(
    f"{base_url}/{IOBROKER_DP_TRANSACTIONS}"
    f"?value={transaction_count}"
).read()

urlopen(
    f"{base_url}/{IOBROKER_DP_WEEKSUM}"
    f"?value={week_sum}"
).read()

urlopen(
    f"{base_url}/{IOBROKER_DP_WEEKEXPENSES}"
    f"?value={week_expenses}"
).read()

# --------------------------------------------------
# Erfolgsmeldung
# --------------------------------------------------

logging.info(
    f"[transactions] Export abgeschlossen | "
    f"Transaktionen={transaction_count} | "
    f"Wochensumme={week_sum:.2f} EUR | "
    f"Ausgaben={week_expenses:.2f} EUR"
)

logging.info(
    f"[transactions] CSV={CSV_FILE}"
)

logging.info(
    f"[transactions] JSON={JSON_FILE}"
)