#!/usr/bin/env python3 """Full-scan bank landing stats (all accounts, all ledger money). Generic for any currency stack (GOA, TESTPAYSAN, KUDOS, …). Run on the ops host (with admin API access). Writes stats.json for public landings + amount_alt. Env (selected): BANK_URL default http://127.0.0.1:9012 BANK_CURRENCY optional; else from exchange /config or first ledger amount BANK_ADMIN_USER default admin BANK_ADMIN_PASS or BANK_ADMIN_PASS_FILE / pass via --admin-pass-file BANK_EXPLORER_USER default explorer BANK_EXPLORER_PASS optional (balance_explorer) EXCHANGE_CONFIG_URL for alt_unit_names + currency BANK_PUBLIC_URL public base for latency probes OUT output path (default stdout if -) SCAN_SKIP comma usernames excluded from flow (default: exchange) TX_PAGE / MAX_TX_PAGES / ACCOUNTS_DELTA / RECENT_WD_N / TZ STATS_SOURCE_LABEL optional string in stats.source field """ from __future__ import annotations import argparse import json import os import re import sys import time import urllib.error import urllib.parse import urllib.request from datetime import datetime from decimal import Decimal from pathlib import Path from typing import Any, Dict, List, Optional, Set, Tuple from zoneinfo import ZoneInfo # local import (same directory) sys.path.insert(0, str(Path(__file__).resolve().parent)) from goa_amounts import ( # noqa: E402 DEFAULT_ALT, enrich_stats_tree, format_amount_alt, load_alt_from_config, parse_amount, ) RESERVE_RE = re.compile(r"(?i)withdrawal\s+([A-Za-z0-9]+)") def env(name: str, default: str = "") -> str: return os.environ.get(name, default) def read_pass_file(path: str) -> str: p = Path(path) if p.is_file(): return p.read_text(encoding="utf-8", errors="replace").strip() return "" def http_json( url: str, *, method: str = "GET", headers: Optional[Dict[str, str]] = None, data: Optional[bytes] = None, timeout: float = 30.0, auth: Optional[Tuple[str, str]] = None, ) -> Tuple[int, Any, bytes]: h = dict(headers or {}) if auth: import base64 token = base64.b64encode(f"{auth[0]}:{auth[1]}".encode()).decode("ascii") h["Authorization"] = f"Basic {token}" req = urllib.request.Request(url, data=data, headers=h, method=method) try: with urllib.request.urlopen(req, timeout=timeout) as resp: body = resp.read() code = getattr(resp, "status", 200) or 200 except urllib.error.HTTPError as e: body = e.read() if e.fp else b"" code = e.code except Exception as e: return 0, None, str(e).encode() if not body: return code, None, body try: return code, json.loads(body.decode("utf-8", errors="replace")), body except Exception: return code, None, body def measure_ms(url: str, timeout: float = 8.0) -> Tuple[Optional[int], str]: t0 = time.perf_counter() code = "000" try: req = urllib.request.Request(url, method="GET") with urllib.request.urlopen(req, timeout=timeout) as resp: resp.read() code = str(getattr(resp, "status", 200) or 200) except urllib.error.HTTPError as e: code = str(e.code) except Exception: return None, "000" ms = int(round((time.perf_counter() - t0) * 1000)) if ms == 0: ms = 1 return ms, code def get_token(bank: str, user: str, password: str) -> str: code, data, _ = http_json( f"{bank}/accounts/{user}/token", method="POST", headers={"Content-Type": "application/json"}, data=b'{"scope":"readonly"}', auth=(user, password), timeout=15, ) if code not in (200, 201) or not isinstance(data, dict): raise RuntimeError(f"token failed for {user}: HTTP {code}") tok = data.get("access_token") or data.get("token") or "" if not tok: raise RuntimeError(f"token empty for {user}") return str(tok) def list_all_accounts(bank: str, token: str, delta: int) -> List[str]: """List accounts; page with start if needed.""" names: List[str] = [] seen: Set[str] = set() start: Optional[int] = None pages = 0 max_pages = int(env("MAX_ACCOUNT_PAGES", "50") or "50") while pages < max_pages: pages += 1 q = f"delta={delta}" if start is not None: q += f"&start={start}" code, data, raw = http_json( f"{bank}/accounts?{q}", headers={"Authorization": f"Bearer {token}"}, timeout=45, ) if code == 204 or not data: break if code != 200: raise RuntimeError(f"list accounts HTTP {code}: {raw[:200]!r}") batch: List[Dict[str, Any]] = [] if isinstance(data, dict): if isinstance(data.get("accounts"), list): batch = data["accounts"] elif isinstance(data.get("users"), list): batch = data["users"] elif isinstance(data, list): batch = data if not batch: # flat object with username? try regex fallback text = raw.decode("utf-8", errors="replace") for m in re.finditer(r'"username"\s*:\s*"([^"]+)"', text): u = m.group(1) if u not in seen: seen.add(u) names.append(u) break min_row = None for item in batch: if not isinstance(item, dict): continue u = item.get("username") or item.get("name") or "" if u and u not in seen: seen.add(u) names.append(str(u)) rid = item.get("row_id") or item.get("rowId") if rid is not None: try: rid_i = int(rid) min_row = rid_i if min_row is None else min(min_row, rid_i) except Exception: pass if len(batch) < abs(delta): break if min_row is None: break # next older page start = min_row # avoid infinite loop on same start if pages > 1 and len(names) == len(seen): # if no growth and full page, still advance pass return names def iter_transactions( bank: str, token: str, username: str, page: int, max_pages: int, ) -> List[Dict[str, Any]]: """Return all transactions for account (paginated).""" out: List[Dict[str, Any]] = [] start: Optional[int] = None pages = 0 safety = max_pages if max_pages > 0 else 10_000 while pages < safety: pages += 1 q = f"delta=-{abs(page)}" if start is not None: q += f"&start={start}" code, data, raw = http_json( f"{bank}/accounts/{urllib.parse.quote(username)}/transactions?{q}", headers={"Authorization": f"Bearer {token}"}, timeout=float(env("TX_CURL_TIMEOUT", "45") or "45"), ) if code in (204, 404) or not data: break if code != 200: # soft-fail one account break txs: List[Dict[str, Any]] = [] if isinstance(data, dict) and isinstance(data.get("transactions"), list): txs = data["transactions"] elif isinstance(data, list): txs = data else: # tolerate raw array-ish break if not txs: break min_row = None for tx in txs: if not isinstance(tx, dict): continue out.append(tx) rid = tx.get("row_id") or tx.get("rowId") if rid is not None: try: rid_i = int(rid) min_row = rid_i if min_row is None else min(min_row, rid_i) except Exception: pass if len(txs) < abs(page): break if min_row is None: break start = min_row return out def tx_fields(tx: Dict[str, Any]) -> Tuple[str, str, int, str]: """direction, amount, unix_ts, subject""" direction = str(tx.get("direction") or "").lower() amount = str(tx.get("amount") or "") subject = str(tx.get("subject") or tx.get("description") or "") ts = 0 when = tx.get("when") or tx.get("date") or tx.get("timestamp") if isinstance(when, dict): # Taler AbsoluteTime: t_s seconds or t_ms if "t_s" in when: try: ts = int(when["t_s"]) except Exception: ts = 0 elif "t_ms" in when: try: ts = int(int(when["t_ms"]) / 1000) except Exception: ts = 0 elif isinstance(when, (int, float)): ts = int(when) if ts > 10_000_000_000: # ms ts //= 1000 elif tx.get("t_s") is not None: try: ts = int(tx["t_s"]) except Exception: ts = 0 return direction, amount, ts, subject def reserve_from_subject(subject: str) -> str: m = RESERVE_RE.search(subject or "") if m: return re.sub(r"[^A-Za-z0-9]", "", m.group(1)) # fallback: last token parts = (subject or "").split() if parts: return re.sub(r"[^A-Za-z0-9]", "", parts[-1]) return "" def now_parts(tz_name: str) -> Tuple[int, str, str]: tz = ZoneInfo(tz_name) dt = datetime.now(tz) unix = int(dt.timestamp()) iso = dt.strftime("%Y-%m-%dT%H:%M%z") # +0200 → +02:00 if len(iso) >= 5 and iso[-5] in "+-" and ":" not in iso[-5:]: iso = iso[:-2] + ":" + iso[-2:] human = dt.strftime("%Y-%m-%d %H:%M %Z") return unix, iso, human def human_from_unix(ts: int, tz_name: str) -> Tuple[str, str]: if not ts: return "", "" tz = ZoneInfo(tz_name) dt = datetime.fromtimestamp(ts, tz) iso = dt.strftime("%Y-%m-%dT%H:%M%z") if len(iso) >= 5 and iso[-5] in "+-" and ":" not in iso[-5:]: iso = iso[:-2] + ":" + iso[-2:] human = dt.strftime("%Y-%m-%d %H:%M %Z") return human, iso def main() -> int: ap = argparse.ArgumentParser(description="Collect full bank landing stats") ap.add_argument("--bank", default=env("BANK_URL", "http://127.0.0.1:9012")) ap.add_argument("--admin-user", default=env("BANK_ADMIN_USER", "admin")) ap.add_argument("--admin-pass", default=env("BANK_ADMIN_PASS", "")) ap.add_argument("--admin-pass-file", default=env("BANK_ADMIN_PASS_FILE", "")) ap.add_argument("--explorer-user", default=env("BANK_EXPLORER_USER", "explorer")) ap.add_argument("--explorer-pass", default=env("BANK_EXPLORER_PASS", "")) ap.add_argument("--explorer-pass-file", default=env("BANK_EXPLORER_PASS_FILE", "")) ap.add_argument( "--exchange-config", default=env("EXCHANGE_CONFIG_URL", "https://exchange.hacktivism.ch/config"), ) ap.add_argument( "--currency", default=env("BANK_CURRENCY", ""), help="ledger currency code (optional; else exchange /config or first amount)", ) ap.add_argument("--out", default=env("OUT", "-")) ap.add_argument("--run-out", default=env("RUN_OUT", "")) ap.add_argument( "--skip", default=env("SCAN_SKIP", "exchange"), help="comma usernames excluded from flow scan (default: exchange)", ) ap.add_argument("--tx-page", type=int, default=int(env("TX_PAGE", "500") or "500")) ap.add_argument( "--max-tx-pages", type=int, default=int(env("MAX_TX_PAGES", "0") or "0"), help="0 = unlimited pages per account", ) ap.add_argument( "--accounts-delta", type=int, default=int(env("ACCOUNTS_DELTA", "-10000") or "-10000"), ) ap.add_argument("--recent", type=int, default=int(env("RECENT_WD_N", "10") or "10")) ap.add_argument("--public-base", default=env("BANK_PUBLIC_URL", "https://bank.hacktivism.ch")) ap.add_argument( "--source-label", default=env("STATS_SOURCE_LABEL", "host collect_bank_stats.py"), help="stats.source field (identify stack / collector)", ) args = ap.parse_args() tz_name = env("TZ", "Europe/Zurich") or "Europe/Zurich" os.environ["TZ"] = tz_name admin_pass = args.admin_pass or ( read_pass_file(args.admin_pass_file) if args.admin_pass_file else "" ) explorer_pass = args.explorer_pass or ( read_pass_file(args.explorer_pass_file) if args.explorer_pass_file else "" ) bank = args.bank.rstrip("/") skip = {s.strip() for s in (args.skip or "").split(",") if s.strip()} currency_hint = (args.currency or "").strip().upper() def write_run(ok: bool, err: Optional[str] = None) -> None: if not args.run_out: return unix, iso, human = now_parts(tz_name) payload = { "ok": ok, "at": iso, "at_human": human, "error": err, } Path(args.run_out).parent.mkdir(parents=True, exist_ok=True) Path(args.run_out).write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") try: if not admin_pass: raise RuntimeError("no admin password (BANK_ADMIN_PASS or --admin-pass-file)") token = get_token(bank, args.admin_user, admin_pass) alt = load_alt_from_config(args.exchange_config) # Currency: explicit → exchange /config → first ledger amount → GOA currency = currency_hint if not currency: try: code_c, cfg_d, _ = http_json(args.exchange_config, timeout=12) if code_c == 200 and isinstance(cfg_d, dict) and cfg_d.get("currency"): currency = str(cfg_d["currency"]).strip().upper() except Exception: pass if not currency: try: code_c, cfg_d, _ = http_json(f"{bank}/config", timeout=12) if code_c == 200 and isinstance(cfg_d, dict) and cfg_d.get("currency"): currency = str(cfg_d["currency"]).strip().upper() except Exception: pass if not currency: currency = "GOA" if not alt: alt = dict(DEFAULT_ALT) accounts = list_all_accounts(bank, token, args.accounts_delta) if not accounts: raise RuntimeError("accounts list empty") # ALL accounts except skip set (default: only exchange — avoid double-count) scan_users = [u for u in accounts if u not in skip] # ensure explorer is included when present if args.explorer_user not in skip and args.explorer_user in accounts: if args.explorer_user not in scan_users: scan_users.append(args.explorer_user) withdraws: List[Dict[str, Any]] = [] incomings: List[Dict[str, Any]] = [] total_in = Decimal(0) total_wd = Decimal(0) total_other = Decimal(0) n_incoming = 0 scan_ok = 0 scan_empty = 0 scan_fail = 0 for uname in scan_users: try: txs = iter_transactions( bank, token, uname, args.tx_page, args.max_tx_pages ) except Exception: scan_fail += 1 continue if not txs: scan_empty += 1 continue scan_ok += 1 for tx in txs: direction, amount, ts, subject = tx_fields(tx) if not amount: continue try: _cur, n = parse_amount(amount) except Exception: continue low = (subject or "").lower() # Normalise bare numbers to stack currency if ":" not in str(amount): amount = f"{currency}:{amount}" else: # adopt ledger currency if we only had a hint try: cur0, _n0 = parse_amount(amount) if cur0 and currency_hint == "" and cur0.upper() != "GOA": currency = cur0.upper() except Exception: pass if direction == "credit": total_in += n n_incoming += 1 incomings.append( { "kind": "incoming", "amount": amount, "at_unix": ts, "account": uname, "subject": subject, } ) elif direction == "debit" and "withdraw" in low: total_wd += n res = reserve_from_subject(subject) withdraws.append( { "kind": "withdraw", "amount": amount, "at_unix": ts, "account": uname, "subject": subject, "reserve": res, } ) elif direction == "debit": total_other += n withdraws.sort(key=lambda r: r.get("at_unix") or 0, reverse=True) incomings.sort(key=lambda r: r.get("at_unix") or 0, reverse=True) now_u, gen_iso, gen_human = now_parts(tz_name) day = now_u - 86400 week = now_u - 7 * 86400 w24 = Decimal(0) w7 = Decimal(0) n24 = 0 n7 = 0 for w in withdraws: ts = int(w.get("at_unix") or 0) try: _c, n = parse_amount(w.get("amount")) except Exception: n = Decimal(0) if ts >= day: w24 += n n24 += 1 if ts >= week: w7 += n n7 += 1 wallets = sorted( {w.get("reserve") for w in withdraws if w.get("reserve")} ) accounts_with_wd = sorted( {w.get("account") for w in withdraws if w.get("account")} ) users_n = sum(1 for u in accounts if u not in ("admin", "exchange")) def pack_amt(d: Decimal) -> Dict[str, Any]: f = format_amount_alt(f"{currency}:{d}", alt) return { "amount": f["amount"], "amount_alt": f["amount_alt"], "amount_full": f["amount_full"], "value": f["value"], "value_str": f["value_str"], } fin = pack_amt(total_in) fwd = pack_amt(total_wd) foth = pack_amt(total_other) fout = pack_amt(total_wd + total_other) p24 = pack_amt(w24) p7 = pack_amt(w7) last = withdraws[0] if withdraws else None last_amt = None last_at = last_iso = last_subj = None last_unix = None last_alt = None if last: lf = format_amount_alt(last["amount"], alt) last_amt = lf["amount"] last_alt = lf["amount_alt"] last_unix = last.get("at_unix") last_at, last_iso = human_from_unix(int(last_unix or 0), tz_name) last_subj = last.get("subject") recent = [] for w in withdraws[: max(0, args.recent)]: f = format_amount_alt(w["amount"], alt) at_h, at_iso = human_from_unix(int(w.get("at_unix") or 0), tz_name) recent.append( { "kind": "withdraw", "amount": f["amount"], "amount_alt": f["amount_alt"], "amount_full": f["amount_full"], "at": at_h, "at_iso": at_iso, "at_unix": w.get("at_unix") or None, "account": w.get("account"), "reserve": w.get("reserve") or "", } ) recent_in = [] for row in incomings[:5]: f = format_amount_alt(row["amount"], alt) at_h, at_iso = human_from_unix(int(row.get("at_unix") or 0), tz_name) recent_in.append( { "kind": "incoming", "amount": f["amount"], "amount_alt": f["amount_alt"], "amount_full": f["amount_full"], "at": at_h, "at_iso": at_iso, "at_unix": row.get("at_unix") or None, "account": row.get("account"), } ) # explorer / pool balance (shared demo account when present) zero = f"{currency}:0" balance = zero bal_alt = zero bal_full = zero if explorer_pass: try: etok = get_token(bank, args.explorer_user, explorer_pass) code, data, _ = http_json( f"{bank}/accounts/{args.explorer_user}", headers={"Authorization": f"Bearer {etok}"}, timeout=12, ) if code == 200 and isinstance(data, dict): balance = str( data.get("balance", {}).get("amount") if isinstance(data.get("balance"), dict) else data.get("amount") or data.get("balance") or zero ) # sometimes balance is object with amount if isinstance(data.get("balance"), dict) and data["balance"].get( "amount" ): balance = str(data["balance"]["amount"]) bf = format_amount_alt(balance, alt) balance, bal_alt, bal_full = ( bf["amount"], bf["amount_alt"], bf["amount_full"], ) except Exception: pass else: # try admin-read of explorer account try: code, data, _ = http_json( f"{bank}/accounts/{args.explorer_user}", headers={"Authorization": f"Bearer {token}"}, timeout=12, ) if code == 200 and isinstance(data, dict): if isinstance(data.get("balance"), dict): balance = str(data["balance"].get("amount") or zero) elif data.get("amount"): balance = str(data["amount"]) bf = format_amount_alt(balance, alt) balance, bal_alt, bal_full = ( bf["amount"], bf["amount_alt"], bf["amount_full"], ) except Exception: pass pub = args.public_base.rstrip("/") config_ms, config_http = measure_ms(f"{pub}/config") int_ms, int_http = measure_ms(f"{pub}/taler-integration/config") webui_ms, webui_http = measure_ms(f"{pub}/webui/") if config_http != "200": config_ms, config_http = measure_ms(f"{bank}/config") if int_http != "200": int_ms, int_http = measure_ms(f"{bank}/taler-integration/config") # loadavg filled later by host merge from *inside* the bank container # (host /proc would be wrong when this script runs on koopa outside podman) loadavg = "" total_pack = pack_amt(total_wd) stats = { "ok": True, "currency": currency, "timezone": tz_name, "generated_at": gen_iso, "generated_at_human": gen_human, "generated_at_unix": now_u, "source": str(args.source_label or "host collect_bank_stats.py"), "bank_url": bank, "scan": { "tx_page": args.tx_page, "max_tx_pages": args.max_tx_pages, "accounts_delta": args.accounts_delta, "skip_users": sorted(skip), "accounts_listed": len(accounts), "accounts_scanned_ok": scan_ok, "accounts_empty_tx": scan_empty, "accounts_scan_fail": scan_fail, "accounts_scanned": len(scan_users), "note": "all accounts except skip_users; full tx pagination; empty includes HTTP 204", }, "bank_accounts": { "total": len(accounts), "users": users_n, "with_withdraws": len(accounts_with_wd), }, "wallets": { "unique_reserves": len(wallets), "note": "unique reserve pubs from Taler withdrawals (one per wallet withdraw)", }, "balance_explorer": balance, "balance_explorer_alt": bal_alt, "balance_explorer_full": bal_full, "flow": { "incoming": { "label": "Incoming bank credits", "count": n_incoming, "amount": fin["amount"], "amount_alt": fin["amount_alt"], "amount_full": fin["amount_full"], "value": fin["value"], "value_str": fin["value_str"], }, "withdraw": { "label": "Taler withdrawals to wallets", "count": len(withdraws), "amount": fwd["amount"], "amount_alt": fwd["amount_alt"], "amount_full": fwd["amount_full"], "value": fwd["value"], "value_str": fwd["value_str"], }, "other_out": { "label": "Other debits (non-withdraw)", "amount": foth["amount"], "amount_alt": foth["amount_alt"], "amount_full": foth["amount_full"], "value": foth["value"], "value_str": foth["value_str"], }, "total_in": fin["amount"], "total_in_alt": fin["amount_alt"], "total_in_value": fin["value"], "total_out": fout["amount"], "total_out_alt": fout["amount_alt"], "total_out_value": fout["value"], "note": "incoming=credits; withdraw=Taler withdrawal debits; skip_users excluded (default exchange)", }, "withdraws": { "count": len(withdraws), "total_amount": total_pack["amount"], "total_amount_alt": total_pack["amount_alt"], "total_amount_full": total_pack["amount_full"], "total_value": total_pack["value"], "total_value_str": total_pack["value_str"], "last_amount": last_amt, "last_amount_alt": last_alt, "last_at": last_at, "last_at_iso": last_iso, "last_at_unix": last_unix, "last_subject": last_subj, "last_24h": { "count": n24, "amount": p24["amount"], "amount_alt": p24["amount_alt"], "amount_full": p24["amount_full"], "value": p24["value"], }, "last_7d": { "count": n7, "amount": p7["amount"], "amount_alt": p7["amount_alt"], "amount_full": p7["amount_full"], "value": p7["value"], }, }, "recent_withdraws": recent, "recent_incoming": recent_in, "demo": { "uri": None, "amount": None, "created": None, "withdrawal_id": None, "status": None, "ready": False, }, "performance": { "config_http": config_http, "config_ms": config_ms, "integration_http": int_http, "integration_ms": int_ms, "webui_http": webui_http, "webui_ms": webui_ms, "loadavg": loadavg, "memory": { "container_rss_human": "—", "container_rss_label": "—", "note": "filled by collect_container_resources.sh (in-container /proc + cgroup)", }, }, "alt_unit_names": alt, } # pull demo withdraw files + memory from bank container if OUT is path and caller merged # demo files optional via DEMO_DIR demo_dir = env("DEMO_DIR", "") if demo_dir: dpath = Path(demo_dir) uri = (dpath / "withdraw.uri").read_text().strip() if (dpath / "withdraw.uri").is_file() else "" amt = ( (dpath / "withdraw.amount").read_text().strip() if (dpath / "withdraw.amount").is_file() else "" ) created = ( (dpath / "withdraw.created").read_text().strip() if (dpath / "withdraw.created").is_file() else "" ) if uri: wid = uri.rstrip("/").split("/")[-1] stats["demo"] = { "uri": uri, "amount": amt or None, "created": created or None, "withdrawal_id": wid, "status": None, "ready": True, } stats = enrich_stats_tree(stats, alt) text = json.dumps(stats, indent=2, ensure_ascii=False) + "\n" if args.out == "-" or not args.out: sys.stdout.write(text) else: outp = Path(args.out) outp.parent.mkdir(parents=True, exist_ok=True) tmp = outp.with_suffix(outp.suffix + f".tmp.{os.getpid()}") tmp.write_text(text, encoding="utf-8") tmp.replace(outp) write_run(True, None) print( f"ok accounts={len(accounts)} scanned={scan_ok} withdraws={len(withdraws)} " f"total={total_pack['amount']} alt={total_pack['amount_alt']}", file=sys.stderr, ) return 0 except Exception as e: write_run(False, str(e)) print(f"error: {e}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())