Finanzfluss DKB

dkb.py

import os
import json
import argparse
import requests

INFO_FILE = "info.json"
LOGIN_URL = "https://hippoverse.finanzfluss.de/api/v2/auth/login"
USER_URL = "https://wealthapi.eu/api/v1/users/myself"
ACCOUNTS_URL = "https://wealthapi.eu/api/v1/accounts"
TRANSACTIONS_URL = "https://wealthapi.eu/api/v1/transactions"
TPP_WEBFORM_URL = "https://wealthapi.eu/api/v1/tppWebForms"
BANK_CONNECTION_FLOW_URL = "https://wealthapi.eu/api/v1/bankConnections/webFormFlow"


def login():
    email = os.getenv("EMAIL")
    password = os.getenv("PASSWORD")

    if not email or not password:
        raise ValueError("EMAIL and PASSWORD must be set as environment variables.")

    session = requests.Session()
    res = session.post(LOGIN_URL, data={"email": email, "password": password})
    auth_dict = res.json()

    session.headers.update({
        "Authorization": f"Bearer {auth_dict['wapiAccessToken']}",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
    })

    return session


def get_account_info(session):
    res = session.get(ACCOUNTS_URL)
    account = res.json()["accounts"][0]
    return account


def update_banking_connection(session, account):
    res = session.post(TPP_WEBFORM_URL, json={
        "bankConnectionId": account["bankConnectionId"],
        "redirectUrl": "https://www.finanzfluss.de/user/depots-accounts"
    })
    webflow = res.json()

    res = session.post(f"{BANK_CONNECTION_FLOW_URL}/{webflow['id']}", json={
        "bankingInterface": "XS2A",
        "storeSecrets": True,
        "loginCredentials": [],
        "autoUpdate": True,
        "redirectUrl": ""
    })
    print("Banking connection updated.")


def fetch_balance_and_transactions(session, account):
    res = session.get(f"{ACCOUNTS_URL}/{account['id']}")
    account_info = res.json()

    res = session.get(f"{TRANSACTIONS_URL}?accountId={account['id']}")
    transactions = res.json().get("transactions", [])[:5]  # Get the last 5 transactions

    balance_data = {
        "balance": account_info["balance"],
        "last_updated": account_info["timestamps"]["updatedAt"],
        "transactions": [
            {
                "counterpart_name": txn["counterpartName"],
                "amount": txn["amount"],
                "value_date": txn["valueDate"]
            }
            for txn in transactions
        ]
    }

    with open(INFO_FILE, "w") as f:
        json.dump(balance_data, f, indent=4)

    print(f"Balance, last update, and last 5 transactions saved to {INFO_FILE}.")


def main():
    parser = argparse.ArgumentParser(description="Banking Data Crawler")
    parser.add_argument("--update", action="store_true",
                        help="Update banking connection instead of fetching balance and transactions.")
    args = parser.parse_args()

    session = login()
    account = get_account_info(session)

    if args.update:
        update_banking_connection(session, account)
    else:
        fetch_balance_and_transactions(session, account)


if __name__ == "__main__":
    main()