from __future__ import annotations import base64 import getpass import hashlib import json import os from dataclasses import dataclass from datetime import date, datetime, timedelta from decimal import Decimal from html.parser import HTMLParser from pathlib import Path from typing import Any from fints.client import FinTS3PinTanClient, NeedTANResponse from fints.models import SEPAAccount from fints.utils import minimal_interactive_cli_bootstrap DEFAULT_PROFILE = "default" DEFAULT_STATE_DIR = Path.home() / ".config" / "mousehold" / "fints" PRODUCT_ID_HELP = ( "FinTS product_id is required by python-fints >=4. Register one with Deutsche Kreditwirtschaft/ZKA " "and provide it as MOUSEHOLD_FINTS_PRODUCT_ID, --product-id, or in the FinTS web form." ) class FinTSConfigurationError(RuntimeError): pass @dataclass(frozen=True) class FinTSConfig: bank_identifier: str user_id: str pin: str server: str product_id: str customer_id: str | None = None product_version: str | None = None tan_mechanism: str | None = None tan_medium: str | None = None profile: str = DEFAULT_PROFILE state_dir: Path = DEFAULT_STATE_DIR store_state: bool = True interactive_bootstrap: bool = True @dataclass(frozen=True) class FinTSFetchResult: account: SEPAAccount transactions: list[Any] observations: list[dict[str, Any]] class _HTMLTextExtractor(HTMLParser): def __init__(self) -> None: super().__init__() self.parts: list[str] = [] def handle_data(self, data: str) -> None: stripped = data.strip() if stripped: self.parts.append(stripped) def text(self) -> str: return " ".join(self.parts) class FinTSProfileStore: def __init__(self, state_dir: Path, profile: str) -> None: self.state_dir = state_dir self.profile = profile self.path = state_dir / f"{profile}.json" def load_client_data(self) -> bytes | None: if not self.path.exists(): return None data = json.loads(self.path.read_text(encoding="utf-8")) client_data = data.get("client_data") if not client_data: return None return base64.b64decode(client_data.encode("ascii")) def save_client_data(self, client: FinTS3PinTanClient, config: FinTSConfig) -> None: self.state_dir.mkdir(parents=True, exist_ok=True) payload = { "profile": self.profile, "bank_identifier": config.bank_identifier, "server": config.server, "user_id": config.user_id, "customer_id": config.customer_id, "updated_at": datetime.now().isoformat(timespec="seconds"), "client_data": base64.b64encode(client.deconstruct()).decode("ascii"), } tmp_path = self.path.with_suffix(".tmp") tmp_path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8") os.chmod(tmp_path, 0o600) tmp_path.replace(self.path) os.chmod(self.path, 0o600) def config_from_args(args: Any) -> FinTSConfig: pin = getattr(args, "pin", None) or os.environ.get(getattr(args, "pin_env", "MOUSEHOLD_FINTS_PIN")) if not pin: pin = getpass.getpass("FinTS PIN: ") product_id = _required_arg(args, "product_id", "MOUSEHOLD_FINTS_PRODUCT_ID", PRODUCT_ID_HELP) product_version = getattr(args, "product_version", None) or os.environ.get("MOUSEHOLD_FINTS_PRODUCT_VERSION") return FinTSConfig( bank_identifier=_required_arg(args, "blz", "MOUSEHOLD_FINTS_BLZ"), user_id=_required_arg(args, "bank_user_id", "MOUSEHOLD_FINTS_USER_ID"), pin=pin, server=_required_arg(args, "server", "MOUSEHOLD_FINTS_SERVER"), customer_id=getattr(args, "customer_id", None) or os.environ.get("MOUSEHOLD_FINTS_CUSTOMER_ID"), product_id=product_id, product_version=product_version, tan_mechanism=getattr(args, "tan_mechanism", None) or os.environ.get("MOUSEHOLD_FINTS_TAN_MECHANISM"), tan_medium=getattr(args, "tan_medium", None) or os.environ.get("MOUSEHOLD_FINTS_TAN_MEDIUM"), profile=getattr(args, "profile", DEFAULT_PROFILE), state_dir=Path(getattr(args, "state_dir", DEFAULT_STATE_DIR)).expanduser(), store_state=not getattr(args, "no_state", False), interactive_bootstrap=not getattr(args, "no_bootstrap", False), ) def create_client(config: FinTSConfig) -> FinTS3PinTanClient: if not config.product_id: raise FinTSConfigurationError(PRODUCT_ID_HELP) kwargs: dict[str, Any] = {"product_id": config.product_id} if config.product_version: kwargs["product_version"] = config.product_version store = FinTSProfileStore(config.state_dir, config.profile) if config.store_state and (client_data := store.load_client_data()): kwargs["from_data"] = client_data return FinTS3PinTanClient( config.bank_identifier, config.user_id, config.pin, config.server, customer_id=config.customer_id, tan_medium=config.tan_medium, **kwargs, ) def bootstrap_client(client: FinTS3PinTanClient, config: FinTSConfig) -> None: if config.tan_mechanism and not client.get_current_tan_mechanism(): client.set_tan_mechanism(config.tan_mechanism) if config.interactive_bootstrap: minimal_interactive_cli_bootstrap(client) def persist_client(client: FinTS3PinTanClient, config: FinTSConfig) -> None: if config.store_state: FinTSProfileStore(config.state_dir, config.profile).save_client_data(client, config) def with_fints_dialog(config: FinTSConfig, operation: Any) -> Any: client = create_client(config) bootstrap_client(client, config) try: with client: while isinstance(client.init_tan_response, NeedTANResponse): client.init_tan_response = answer_tan_challenge(client, client.init_tan_response, config) return operation(client) finally: persist_client(client, config) def list_accounts(config: FinTSConfig, *, include_balances: bool = False) -> list[dict[str, Any]]: def operation(client: FinTS3PinTanClient) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for index, account in enumerate(client.get_sepa_accounts()): row = account_to_dict(account) row["index"] = index if include_balances: balance = resolve_tan_response(client, client.get_balance(account), config) row["balance"] = balance_to_dict(balance) rows.append(row) return rows return with_fints_dialog(config, operation) def fetch_transactions( config: FinTSConfig, *, iban: str | None, start_date: date, end_date: date, include_pending: bool = False, household_id: str | None = None, source_account_id: str | None = None, owner_user_id: str | None = None, ) -> FinTSFetchResult: def operation(client: FinTS3PinTanClient) -> FinTSFetchResult: account = select_account(client.get_sepa_accounts(), iban) response = client.get_transactions( account, start_date=start_date, end_date=end_date, include_pending=include_pending, ) transactions = resolve_tan_response(client, response, config) observations = [ transaction_to_observation( transaction, account=account, household_id=household_id, source_account_id=source_account_id, owner_user_id=owner_user_id, ) for transaction in transactions ] return FinTSFetchResult(account=account, transactions=list(transactions), observations=observations) return with_fints_dialog(config, operation) def resolve_tan_response( client: FinTS3PinTanClient, response: Any, config: FinTSConfig, ) -> Any: while isinstance(response, NeedTANResponse): response = answer_tan_challenge(client, response, config) return response def answer_tan_challenge( client: FinTS3PinTanClient, response: NeedTANResponse, config: FinTSConfig, ) -> Any: print("\nTAN required by bank.") challenge_text = challenge_to_text(response) if challenge_text: print(challenge_text) challenge_file = write_binary_challenge(response, config) if challenge_file: print(f"Challenge file: {challenge_file}") if getattr(response, "challenge_hhduc", None): print("A chipTAN/flicker challenge is available.") print("Use your bank reader/app if it can read the challenge shown by your terminal.") if getattr(response, "decoupled", False): input("Confirm the login/order in your banking app, then press Enter here.") return client.send_tan(response, "") tan = getpass.getpass("TAN: ") return client.send_tan(response, tan) def tan_challenge_to_dict(response: NeedTANResponse) -> dict[str, Any]: matrix = getattr(response, "challenge_matrix", None) matrix_payload = None if matrix: mime_type, payload = matrix matrix_payload = { "mime_type": str(mime_type), "data_base64": base64.b64encode(payload).decode("ascii"), } return { "challenge": challenge_to_text(response), "challenge_html": str(getattr(response, "challenge_html", "") or ""), "challenge_hhduc": getattr(response, "challenge_hhduc", None), "challenge_matrix": matrix_payload, "decoupled": bool(getattr(response, "decoupled", False)), } def challenge_to_text(response: NeedTANResponse) -> str: if getattr(response, "challenge", None): return str(response.challenge) if getattr(response, "challenge_html", None): parser = _HTMLTextExtractor() parser.feed(str(response.challenge_html)) return parser.text() if getattr(response, "challenge_raw", None): return str(response.challenge_raw) return "" def write_binary_challenge(response: NeedTANResponse, config: FinTSConfig) -> Path | None: matrix = getattr(response, "challenge_matrix", None) if not matrix: return None mime_type, payload = matrix suffix = ".bin" if "png" in str(mime_type).lower(): suffix = ".png" elif "jpeg" in str(mime_type).lower() or "jpg" in str(mime_type).lower(): suffix = ".jpg" output_dir = config.state_dir / "challenges" output_dir.mkdir(parents=True, exist_ok=True) path = output_dir / f"{config.profile}-{datetime.now().strftime('%Y%m%d%H%M%S')}{suffix}" path.write_bytes(payload) os.chmod(path, 0o600) return path def select_account(accounts: list[SEPAAccount], iban: str | None) -> SEPAAccount: if not accounts: raise RuntimeError("No SEPA accounts returned by the bank.") if not iban: if len(accounts) > 1: print("Multiple accounts returned; using the first one. Pass --iban to select explicitly.") return accounts[0] normalized = iban.replace(" ", "").upper() for account in accounts: if (account.iban or "").replace(" ", "").upper() == normalized: return account available = ", ".join(account.iban or "" for account in accounts) raise RuntimeError(f"IBAN not found. Available accounts: {available}") def date_range_from_args(args: Any) -> tuple[date, date]: end_date = _parse_date_arg(getattr(args, "end_date", None)) or date.today() start_date = _parse_date_arg(getattr(args, "start_date", None)) if not start_date: start_date = end_date - timedelta(days=getattr(args, "days", 30)) return start_date, end_date def transaction_to_observation( transaction: Any, *, account: SEPAAccount, household_id: str | None, source_account_id: str | None, owner_user_id: str | None, ) -> dict[str, Any]: raw_data = transaction_data(transaction) normalized_data = jsonable(raw_data) amount, currency = extract_amount(raw_data) booking_date = extract_date(raw_data, "date", "booking_date", "entry_date") value_date = extract_date(raw_data, "entry_date", "valuta_date", "date") counterparty = pick_text( raw_data, "applicant_name", "applicant_iban", "applicant_bin", "counterparty", "name", ) description = pick_text( raw_data, "purpose", "transaction_details", "extra_details", "posting_text", "prima_nota", "additional_position_reference", ) reference = pick_text( raw_data, "id", "transaction_reference", "customer_reference", "bank_reference", "end_to_end_reference", "eref", "mref", "svwz", ) fallback_reference = stable_transaction_reference(account, normalized_data) signed_amount = amount return { "household_id": household_id, "source_type": "fints", "source_account_id": source_account_id, "owner_user_id": owner_user_id, "source_reference": reference or fallback_reference, "provider_transaction_id": reference or fallback_reference, "raw_payload_json": { "account": account_to_dict(account), "transaction": normalized_data, "source": "fints", }, "amount": str(abs(signed_amount)), "currency": currency or "EUR", "event_date": (booking_date or value_date or date.today()).isoformat(), "booking_date": booking_date.isoformat() if booking_date else None, "merchant": counterparty, "counterparty": counterparty, "description": description, "event_type": infer_event_type(signed_amount, counterparty, description), "status": "confirmed", } def transaction_data(transaction: Any) -> dict[str, Any]: data = getattr(transaction, "data", transaction) if isinstance(data, dict): return data if hasattr(data, "_asdict"): return data._asdict() return {"value": data} def extract_amount(data: dict[str, Any]) -> tuple[Decimal, str | None]: value = data.get("amount") or data.get("transaction_amount") currency = data.get("currency") if hasattr(value, "amount"): currency = getattr(value, "currency", currency) return Decimal(str(value.amount)), currency if isinstance(value, dict): amount = value.get("amount") or value.get("value") currency = value.get("currency", currency) return Decimal(str(amount)), currency if value is None: raise RuntimeError(f"Transaction has no amount: {data}") return Decimal(str(value).replace(",", ".")), currency def extract_date(data: dict[str, Any], *keys: str) -> date | None: for key in keys: value = data.get(key) if isinstance(value, date): return value if isinstance(value, datetime): return value.date() if isinstance(value, str) and value: for fmt in ("%Y-%m-%d", "%d.%m.%Y", "%Y%m%d"): try: return datetime.strptime(value, fmt).date() except ValueError: continue return None def pick_text(data: dict[str, Any], *keys: str) -> str | None: values: list[str] = [] for key in keys: value = data.get(key) if value is None or value == "": continue if isinstance(value, list | tuple): values.extend(str(item).strip() for item in value if str(item).strip()) else: values.append(str(value).strip()) if not values: return None return " ".join(dict.fromkeys(values))[:500] def infer_event_type(amount: Decimal, counterparty: str | None, description: str | None) -> str: if amount >= 0: return "income" combined = f"{counterparty or ''} {description or ''}".lower() liability_markers = ( "paypal", "klarna", "kreditkarte", "credit card", "visa", "mastercard", "american express", "amex", ) if any(marker in combined for marker in liability_markers): return "liability_payment" return "expense" def account_to_dict(account: SEPAAccount) -> dict[str, Any]: return { "iban": account.iban, "bic": account.bic, "accountnumber": account.accountnumber, "subaccount": account.subaccount, "blz": account.blz, } def balance_to_dict(balance: Any) -> dict[str, Any]: amount = getattr(balance, "amount", None) return { "amount": str(getattr(amount, "amount", amount)), "currency": getattr(amount, "currency", None), "date": getattr(balance, "date", None).isoformat() if isinstance(getattr(balance, "date", None), date) else str(getattr(balance, "date", "")), "status": getattr(balance, "status", None), } def stable_transaction_reference(account: SEPAAccount, payload: dict[str, Any]) -> str: digest = hashlib.sha256( json.dumps( {"account": account_to_dict(account), "transaction": payload}, sort_keys=True, separators=(",", ":"), ).encode("utf-8") ).hexdigest() return f"fints:{account.iban or account.accountnumber}:{digest}" def jsonable(value: Any) -> Any: if isinstance(value, dict): return {str(key): jsonable(item) for key, item in value.items()} if isinstance(value, list | tuple | set): return [jsonable(item) for item in value] if isinstance(value, datetime | date): return value.isoformat() if isinstance(value, Decimal): return str(value) if hasattr(value, "amount"): return { "amount": str(value.amount), "currency": getattr(value, "currency", None), } if hasattr(value, "_asdict"): return jsonable(value._asdict()) return value if isinstance(value, str | int | float | bool | type(None)) else str(value) def display_accounts(accounts: list[dict[str, Any]]) -> None: rows = [] for account in accounts: balance = account.get("balance") balance_text = "" if isinstance(balance, dict) and balance.get("amount"): balance_text = f"{balance['amount']} {balance.get('currency') or ''}".strip() rows.append( [ str(account.get("index", "")), account.get("iban") or "", account.get("bic") or "", account.get("accountnumber") or "", balance_text, ] ) print_table(["#", "IBAN", "BIC", "Account", "Balance"], rows) def display_transactions(observations: list[dict[str, Any]]) -> None: rows = [ [ observation.get("event_date") or "", observation.get("event_type") or "", observation.get("amount") or "", observation.get("currency") or "", observation.get("counterparty") or "", observation.get("description") or "", ] for observation in observations ] print_table(["Date", "Type", "Amount", "Cur", "Counterparty", "Description"], rows) def print_table(headers: list[str], rows: list[list[str]]) -> None: widths = [ max(len(headers[index]), *(len(row[index]) for row in rows)) if rows else len(headers[index]) for index in range(len(headers)) ] print(" ".join(header.ljust(widths[index]) for index, header in enumerate(headers))) print(" ".join("-" * width for width in widths)) for row in rows: print(" ".join(row[index].ljust(widths[index]) for index in range(len(headers)))) def _parse_date_arg(value: str | None) -> date | None: if not value: return None return date.fromisoformat(value) def _required_arg(args: Any, attribute: str, env_name: str, message: str | None = None) -> str: value = getattr(args, attribute, None) or os.environ.get(env_name) if not value: raise RuntimeError(message or f"{attribute.replace('_', '-')} or {env_name} is required") return value