583 lines
20 KiB
Python
583 lines
20 KiB
Python
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 "<no iban>" 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
|