kvitt/docs/spikes/k0/app.py

#!/usr/bin/env python3
"""K0 spike: a local test app for Enable Banking (card account via BankID).

Python 3 standard library only. The RS256 JWT is signed by shelling out to openssl.
Run: python3 docs/spikes/k0/app.py   (serves http://localhost:8765/ and opens the browser)

Config and state live in docs/spikes/k0/.local/ (git-ignored):
  <application-id>.pem   the private key from the Enable Banking control panel
  config.json            optional: {"application_id": "...", "key_file": "...pem"}
  state.json             session, account, settings and the last fetched transactions

stdout carries only method, path and status. Never response bodies, texts or amounts.
"""

from __future__ import annotations

import argparse
import base64
import datetime as dt
import html
import json
import os
import re
import secrets
import shutil
import socket
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import webbrowser
from decimal import Decimal, InvalidOperation
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

HERE = os.path.dirname(os.path.abspath(__file__))
DEFAULT_LOCAL_DIR = os.path.join(HERE, ".local")
DEFAULT_API_BASE = "https://api.enablebanking.com"
DEFAULT_PORT = 8765
# The control panel rejects http://localhost ("uses unsupported scheme"), so the
# registered https URL is the default and the paste box is the main path. The
# localhost callback handler stays for a future sandbox or https tunnel.
REDIRECT_CHOICES = [
    "https://kvitt.suorra.duckdns.org/cards/callback",
    "http://localhost:8765/callback",
]
PASTE_HINT = (
    "När BankID är klart hamnar du på en sida som inte laddar. "
    "Kopiera hela adressen från adressfältet och klistra in den här."
)
HISTORY_CHOICES = [7, 30, 90, 180, 365, 730]
CONSENT_CAP_SECONDS = 180 * 24 * 3600
CONSENT_FALLBACK_SECONDS = 90 * 24 * 3600  # used only when the ASPSP list is not loaded
MATCH_WINDOW_DAYS = 7  # spike heuristic for pending->booked matching by amount and date
MAX_PAGES = 200
HIGHLIGHT = re.compile(r"\bSJ\b|SEB\s*Kort|Eurocard", re.IGNORECASE)
DEFAULT_ASPSP = re.compile(r"SJ\s*Prio", re.IGNORECASE)
UUID_PEM = re.compile(
    r"^([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})\.pem$"
)
PSU_HEADER_CANON = {
    "psu-ip-address": "Psu-Ip-Address",
    "psu-user-agent": "Psu-User-Agent",
    "psu-referer": "Psu-Referer",
    "psu-accept": "Psu-Accept",
    "psu-accept-charset": "Psu-Accept-Charset",
    "psu-accept-encoding": "Psu-Accept-Encoding",
    "psu-accept-language": "Psu-Accept-Language",
    "psu-geo-location": "Psu-Geo-Location",
}

LOCK = threading.RLock()
CFG = {
    "local_dir": DEFAULT_LOCAL_DIR,
    "api_base": DEFAULT_API_BASE,
    "port": DEFAULT_PORT,
    "open_browser": True,
}
LAST = {}  # last API response or message shown on the page (in memory only)
JWT_CACHE = {"token": None, "exp": 0, "kid": None}


def log(msg):
    print(time.strftime("%H:%M:%S"), msg, flush=True)


# ----------------------------------------------------------------------------- config and state


def local_path(name):
    return os.path.join(CFG["local_dir"], name)


def read_json(path, default):
    try:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except FileNotFoundError:
        return default
    except (OSError, ValueError) as e:
        log("could not read %s: %s" % (os.path.basename(path), type(e).__name__))
        return default


def load_config():
    """Return (application_id, key_path, problems)."""
    problems = []
    cfg = read_json(local_path("config.json"), {})
    app_id = cfg.get("application_id") if isinstance(cfg, dict) else None
    key_file = cfg.get("key_file") if isinstance(cfg, dict) else None
    pems = []
    try:
        pems = sorted(n for n in os.listdir(CFG["local_dir"]) if UUID_PEM.match(n))
    except FileNotFoundError:
        problems.append("Mappen %s finns inte." % CFG["local_dir"])
    if not app_id:
        if len(pems) == 1:
            m = UUID_PEM.match(pems[0])
            app_id = m.group(1) if m else None
        elif len(pems) > 1:
            problems.append(
                "Flera <id>.pem i .local; ange application_id i config.json."
            )
    if key_file:
        key_path = key_file if os.path.isabs(key_file) else local_path(key_file)
    elif app_id:
        key_path = local_path(app_id + ".pem")
    else:
        key_path = None
    if not app_id:
        problems.append("Inget application id (lägg <id>.pem i .local eller config.json).")
    if key_path and not os.path.isfile(key_path):
        problems.append("Nyckelfilen %s saknas." % os.path.basename(key_path))
    return app_id, key_path, problems


def default_state():
    return {
        "settings": {
            "aspsp_name": "",
            "history_days": 90,
            "redirect_url": REDIRECT_CHOICES[0],
            "auth_method": "",
            "strategy": "default",
            "psu_ip": "",
        },
        "aspsps": [],
        "aspsps_fetched_at": None,
        "pending_auth": None,
        "session": None,
        "account_uid": None,
        "transactions": [],
        "transactions_meta": None,
        "previous_transactions": [],
        "last_diff": None,
        "balances": None,
        "fetch_log": [],
    }


def load_state():
    st = default_state()
    saved = read_json(local_path("state.json"), {})
    if isinstance(saved, dict):
        for k, v in saved.items():
            if k == "settings" and isinstance(v, dict):
                st["settings"].update(v)
            else:
                st[k] = v
    return st


def save_state(st):
    os.makedirs(CFG["local_dir"], exist_ok=True)
    try:
        os.chmod(CFG["local_dir"], 0o700)
    except OSError as e:
        log("chmod .local failed: %s" % type(e).__name__)
    path = local_path("state.json")
    tmp = path + ".tmp"
    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w", encoding="utf-8") as f:
        json.dump(st, f, ensure_ascii=False, indent=1)
    os.replace(tmp, path)


# ----------------------------------------------------------------------------- JWT


def b64url(raw):
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")


def openssl_bin():
    return shutil.which("openssl") or "/usr/bin/openssl"


def make_jwt(app_id, key_path, now=None):
    iat = int(now if now is not None else time.time())
    header = {"typ": "JWT", "alg": "RS256", "kid": app_id}
    payload = {
        "iss": "enablebanking.com",
        "aud": "api.enablebanking.com",
        "iat": iat,
        "exp": iat + 3600,
    }
    signing_input = (
        b64url(json.dumps(header, separators=(",", ":")).encode())
        + "."
        + b64url(json.dumps(payload, separators=(",", ":")).encode())
    )
    proc = subprocess.run(
        [openssl_bin(), "dgst", "-sha256", "-sign", key_path],
        input=signing_input.encode("ascii"),
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        check=False,
    )
    if proc.returncode != 0 or not proc.stdout:
        raise RuntimeError(
            "openssl kunde inte signera (exit %d): %s"
            % (proc.returncode, proc.stderr.decode("utf-8", "replace").strip()[:300])
        )
    return signing_input + "." + b64url(proc.stdout), payload["exp"]


def get_jwt():
    app_id, key_path, problems = load_config()
    if not app_id or not key_path or not os.path.isfile(key_path):
        raise RuntimeError("; ".join(problems) or "Nyckel saknas.")
    now = time.time()
    if JWT_CACHE["token"] and JWT_CACHE["kid"] == app_id and JWT_CACHE["exp"] - now > 300:
        return JWT_CACHE["token"]
    token, exp = make_jwt(app_id, key_path, now)
    JWT_CACHE.update(token=token, exp=exp, kid=app_id)
    return token


# ----------------------------------------------------------------------------- HTTP client


class ApiResult(object):
    def __init__(self, status, data, raw_text, error=None):
        self.status = status
        self.data = data
        self.raw_text = raw_text
        self.error = error

    @property
    def ok(self):
        return 200 <= self.status < 300 and self.error is None


def api(method, path, route, query=None, body=None, headers=None):
    """Call Enable Banking. `route` is the path template that is logged (no ids)."""
    try:
        token = get_jwt()
    except (RuntimeError, OSError) as e:
        return ApiResult(0, None, "", error=str(e))
    url = CFG["api_base"].rstrip("/") + path
    if query:
        url += "?" + urllib.parse.urlencode(query)
    data = json.dumps(body).encode("utf-8") if body is not None else None
    req = urllib.request.Request(url, data=data, method=method)
    req.add_header("Authorization", "Bearer " + token)
    req.add_header("Accept", "application/json")
    if data is not None:
        req.add_header("Content-Type", "application/json")
    for k, v in (headers or {}).items():
        req.add_header(k, v)
    status = 0
    raw = b""
    try:
        with urllib.request.urlopen(req, timeout=90) as resp:
            status = resp.status
            raw = resp.read()
    except urllib.error.HTTPError as e:
        status = e.code
        try:
            raw = e.read()
        except OSError:
            raw = b""
    except (urllib.error.URLError, OSError) as e:
        log("EB %s %s network-error" % (method, route))
        reason = getattr(e, "reason", e)
        return ApiResult(0, None, "", error="Nätverksfel: %s" % reason)
    log("EB %s %s %d" % (method, route, status))
    text = raw.decode("utf-8", "replace")
    try:
        parsed = json.loads(text) if text.strip() else None
    except ValueError:
        parsed = None
    return ApiResult(status, parsed, text)


def set_last(kind, title, result=None, message=None, section="top"):
    entry = {"kind": kind, "title": title, "section": section, "at": now_iso()}
    if result is not None:
        entry["status"] = result.status
        if result.error:
            entry["body"] = result.error
        elif result.data is not None:
            entry["body"] = json.dumps(result.data, ensure_ascii=False, indent=2)
        else:
            entry["body"] = result.raw_text[:4000]
    if message:
        entry["message"] = message
    LAST.clear()
    LAST.update(entry)


def fail(title, result, section):
    set_last("error", title, result=result, section=section)


def now_iso():
    return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat()


def public_ip():
    try:
        with urllib.request.urlopen("https://api.ipify.org?format=text", timeout=10) as r:
            ip = r.read().decode("ascii", "replace").strip()
            log("ipify GET / %d" % r.status)
            if re.match(r"^[0-9a-fA-F:.]+$", ip):
                return ip
    except (urllib.error.URLError, OSError):
        log("ipify GET / network-error")
    return None


# ----------------------------------------------------------------------------- helpers on data


def chosen_aspsp(st):
    name = st["settings"].get("aspsp_name")
    for a in st.get("aspsps") or []:
        if a.get("name") == name:
            return a
    return None


def default_aspsp_name(aspsps):
    for a in aspsps:
        if DEFAULT_ASPSP.search(a.get("name") or ""):
            return a.get("name")
    return ""


def mask(value):
    s = str(value)
    if len(s) <= 4:
        return "••••"
    return "••••" + s[-4:]


def mask_ids(obj):
    """Mask every string leaf of an identifier structure to its last 4 characters,
    keeping scheme names readable."""
    if isinstance(obj, dict):
        out = {}
        for k, v in obj.items():
            if k in ("scheme_name", "issuer") and isinstance(v, str):
                out[k] = v
            else:
                out[k] = mask_ids(v)
        return out
    if isinstance(obj, list):
        return [mask_ids(v) for v in obj]
    if obj is None:
        return None
    return mask(obj)


def amount_of(t):
    ta = t.get("transaction_amount") or {}
    try:
        val = abs(Decimal(str(ta.get("amount"))))
    except (InvalidOperation, ValueError, TypeError):
        return None
    if t.get("credit_debit_indicator") == "DBIT":
        val = -val
    return val


def currency_of(t):
    return (t.get("transaction_amount") or {}).get("currency") or ""


def fmt_kr(val):
    if val is None:
        return "?"
    sign = "−" if val < 0 else "+"
    s = "{:,.2f}".format(abs(val)).replace(",", " ").replace(".", ",")
    return sign + s


def text_of(t):
    parts = []
    party = t.get("creditor") if t.get("credit_debit_indicator") == "DBIT" else t.get("debtor")
    if isinstance(party, dict) and party.get("name"):
        parts.append(str(party["name"]))
    ri = t.get("remittance_information")
    if isinstance(ri, list):
        parts.extend(str(x) for x in ri if x)
    elif ri:
        parts.append(str(ri))
    if t.get("note"):
        parts.append(str(t["note"]))
    return " / ".join(parts)


def date_of(t):
    return t.get("transaction_date") or t.get("booking_date") or t.get("value_date")


def parse_date(s):
    try:
        return dt.date.fromisoformat(str(s)[:10])
    except (ValueError, TypeError):
        return None


def present(v):
    return v is not None and v != "" and v != [] and v != {}


def flatten_presence(obj, prefix, out):
    for k, v in obj.items():
        p = prefix + k
        if isinstance(v, dict) and v:
            out[p] = True
            flatten_presence(v, p + ".", out)
        else:
            out[p] = present(v)


def field_presence(txs):
    seen = {}
    for t in txs:
        flat = {}
        flatten_presence(t, "", flat)
        for k, v in flat.items():
            seen.setdefault(k, 0)
            if v:
                seen[k] += 1
    return seen


# ----------------------------------------------------------------------------- diff


def _er(t):
    v = t.get("entry_reference")
    return str(v) if present(v) else None


def diff_transactions(prev, curr):
    """Match the previous fetch to the current one.

    1. equal entry_reference;
    2a. same amount, direction, currency, date and text (entry_reference ignored);
    2b. same amount, direction and currency, date within MATCH_WINDOW_DAYS, only
        when at least one side has no entry_reference (closest date wins).
    """
    matched_prev = {}
    matched_curr = {}
    by = {}
    prev_er = {}
    for i, t in enumerate(prev):
        er = _er(t)
        if er is not None:
            prev_er.setdefault(er, []).append(i)
    for j, t in enumerate(curr):
        er = _er(t)
        if er is None:
            continue
        for i in prev_er.get(er, []):
            if i not in matched_prev:
                matched_prev[i] = j
                matched_curr[j] = i
                by[j] = "entry_reference"
                break

    def fp(t, with_text):
        return (
            str(amount_of(t)),
            t.get("credit_debit_indicator"),
            currency_of(t),
            date_of(t) if with_text else None,
            text_of(t) if with_text else None,
        )

    strict = {}
    for i, t in enumerate(prev):
        if i not in matched_prev:
            strict.setdefault(fp(t, True), []).append(i)
    for j, t in enumerate(curr):
        if j in matched_curr:
            continue
        for i in strict.get(fp(t, True), []):
            if i not in matched_prev:
                matched_prev[i] = j
                matched_curr[j] = i
                by[j] = "amount_date"
                break

    for j, t in enumerate(curr):
        if j in matched_curr:
            continue
        dj = parse_date(date_of(t))
        best = None
        for i, p in enumerate(prev):
            if i in matched_prev:
                continue
            if fp(p, False) != fp(t, False):
                continue
            if _er(p) is not None and _er(t) is not None:
                continue
            di = parse_date(date_of(p))
            if dj is None or di is None:
                continue
            gap = abs((dj - di).days)
            if gap > MATCH_WINDOW_DAYS:
                continue
            if best is None or gap < best[0]:
                best = (gap, i)
        if best is not None:
            i = best[1]
            matched_prev[i] = j
            matched_curr[j] = i
            by[j] = "amount_window"

    compare = [
        "status",
        "booking_date",
        "value_date",
        "transaction_date",
        "entry_reference",
        "transaction_id",
    ]
    changed = []
    stats = {
        "pdng_to_book_by_entry_reference": 0,
        "pdng_to_book_by_amount_date": 0,
        "er_same": 0,
        "er_changed": 0,
        "er_appeared": 0,
        "er_prev_unmatched": 0,
    }
    for j, i in sorted(matched_curr.items()):
        p, c = prev[i], curr[j]
        fields = [f for f in compare if p.get(f) != c.get(f)]
        if text_of(p) != text_of(c):
            fields.append("text")
        if str(amount_of(p)) != str(amount_of(c)):
            fields.append("amount")
        if _er(p) is not None and _er(c) is not None:
            if _er(p) == _er(c):
                stats["er_same"] += 1
            else:
                stats["er_changed"] += 1
        elif _er(p) is None and _er(c) is not None:
            stats["er_appeared"] += 1
        if p.get("status") == "PDNG" and c.get("status") == "BOOK":
            if by[j] == "entry_reference":
                stats["pdng_to_book_by_entry_reference"] += 1
            else:
                stats["pdng_to_book_by_amount_date"] += 1
        if fields:
            changed.append({"idx": j, "prev_idx": i, "fields": fields, "by": by[j]})
    new = [j for j in range(len(curr)) if j not in matched_curr]
    gone = [i for i in range(len(prev)) if i not in matched_prev]
    stats["er_prev_unmatched"] = sum(1 for i in gone if _er(prev[i]) is not None)
    return {
        "new": new,
        "changed": changed,
        "disappeared": gone,
        "matched_by": {str(j): by[j] for j in by},
        "stats": stats,
        "had_previous": bool(prev),
    }


# ----------------------------------------------------------------------------- actions


def action_list_aspsps(st):
    res = api("GET", "/aspsps", "/aspsps", query={"country": "SE"})
    if not res.ok or not isinstance(res.data, dict):
        fail("Lista banker misslyckades", res, "banker")
        return
    aspsps = res.data.get("aspsps") or []
    for a in aspsps:
        a.pop("logo", None)
    st["aspsps"] = aspsps
    st["aspsps_fetched_at"] = now_iso()
    if not chosen_aspsp(st):
        st["settings"]["aspsp_name"] = default_aspsp_name(aspsps)
    set_last("ok", "Lista banker", message="%d banker i SE." % len(aspsps), section="banker")


def action_choose_aspsp(st, form):
    name = form.get("aspsp", "")
    if not any(a.get("name") == name for a in st.get("aspsps") or []):
        set_last("error", "Välj bank", message="Okänd bank.", section="banker")
        return
    st["settings"]["aspsp_name"] = name
    st["settings"]["auth_method"] = ""
    set_last("ok", "Välj bank", message="Vald: %s" % name, section="banker")


def save_settings(st, form):
    s = st["settings"]
    try:
        days = int(form.get("history_days", s["history_days"]))
        if days in HISTORY_CHOICES:
            s["history_days"] = days
    except ValueError:
        pass
    choice = form.get("redirect_choice", "")
    custom = form.get("redirect_custom", "").strip()
    if choice == "custom" and custom:
        s["redirect_url"] = custom
    elif choice in REDIRECT_CHOICES:
        s["redirect_url"] = choice
    s["auth_method"] = form.get("auth_method", "")
    if form.get("strategy") in ("default", "longest"):
        s["strategy"] = form["strategy"]


def consent_seconds(aspsp):
    if aspsp and isinstance(aspsp.get("maximum_consent_validity"), (int, float)):
        return min(int(aspsp["maximum_consent_validity"]), CONSENT_CAP_SECONDS)
    return CONSENT_FALLBACK_SECONDS


def build_auth_body(st, state_token, now=None):
    s = st["settings"]
    aspsp = chosen_aspsp(st)
    now = now or dt.datetime.now(dt.timezone.utc)
    # a few minutes' margin so the request never exceeds the ASPSP's maximum
    valid_until = now + dt.timedelta(seconds=consent_seconds(aspsp) - 300)
    body = {
        "access": {"valid_until": valid_until.replace(microsecond=0).isoformat()},
        "aspsp": {"name": s["aspsp_name"], "country": "SE"},
        "state": state_token,
        "redirect_url": s["redirect_url"],
        "psu_type": "personal",
    }
    if s.get("auth_method"):
        body["auth_method"] = s["auth_method"]
    return body


def action_auth(st):
    if not st["settings"].get("aspsp_name"):
        set_last("error", "Anslut", message="Välj en bank först (Lista banker).", section="anslut")
        return None
    token = secrets.token_urlsafe(24)
    body = build_auth_body(st, token)
    res = api("POST", "/auth", "/auth", body=body)
    if not res.ok or not isinstance(res.data, dict) or not res.data.get("url"):
        fail("POST /auth misslyckades", res, "anslut")
        return None
    st["pending_auth"] = {
        "state": token,
        "aspsp": body["aspsp"],
        "valid_until": body["access"]["valid_until"],
        "redirect_url": body["redirect_url"],
        "started_at": now_iso(),
        "authorization_id": res.data.get("authorization_id"),
        "url": res.data["url"],
    }
    set_last(
        "ok",
        "POST /auth (HTTP %d)" % res.status,
        message="Bankens sida öppnas i en ny flik. Scanna BankID-QR-koden med telefonen. " + PASTE_HINT,
        section="anslut",
    )
    return res.data["url"]


def extract_code_state(text):
    """Pull code, state, error and error_description out of any pasted URL or query."""
    text = (text or "").strip()
    found = {}
    candidates = []
    parsed = urllib.parse.urlparse(text)
    candidates.append(parsed.query)
    candidates.append(parsed.fragment)
    if "?" in text:
        candidates.append(text.split("?", 1)[1].split("#", 1)[0])
    candidates.append(text.lstrip("?#"))
    for c in candidates:
        if not c:
            continue
        qs = urllib.parse.parse_qs(c, keep_blank_values=False)
        for key in ("code", "state", "error", "error_description"):
            if key not in found and qs.get(key):
                found[key] = qs[key][0]
    return found


def complete_auth(st, params, source):
    if params.get("error"):
        set_last(
            "error",
            "Banken svarade med fel (%s)" % source,
            message="%s: %s" % (params.get("error"), params.get("error_description", "")),
            section="anslut",
        )
        return
    code = params.get("code")
    if not code:
        set_last("error", "Ingen code", message="Hittade ingen code i adressen.", section="anslut")
        return
    pending = st.get("pending_auth") or {}
    if pending.get("state") and params.get("state") != pending.get("state"):
        set_last(
            "error",
            "state stämmer inte",
            message="state i adressen matchar inte den senaste anslutningen. Starta om Anslut med BankID.",
            section="anslut",
        )
        return
    res = api("POST", "/sessions", "/sessions", body={"code": code})
    if not res.ok or not isinstance(res.data, dict):
        fail("POST /sessions misslyckades", res, "anslut")
        return
    d = res.data
    accounts = d.get("accounts") or []
    st["session"] = {
        "session_id": d.get("session_id"),
        "valid_until": (d.get("access") or {}).get("valid_until"),
        "aspsp": d.get("aspsp"),
        "psu_type": d.get("psu_type"),
        "accounts": accounts,
        "created_at": now_iso(),
    }
    st["pending_auth"] = None
    st["account_uid"] = default_account_uid(accounts)
    st["balances"] = None
    set_last(
        "ok",
        "Session skapad (%s)" % source,
        message="%d konto(n). Giltig till %s." % (len(accounts), st["session"]["valid_until"]),
        section="konton",
    )


def default_account_uid(accounts):
    if len(accounts) == 1:
        return accounts[0].get("uid")
    for a in accounts:
        if (a.get("cash_account_type") or "").upper() == "CARD":
            return a.get("uid")
    return None


def action_choose_account(st, form):
    uid = form.get("account_uid", "")
    accounts = (st.get("session") or {}).get("accounts") or []
    if not any(a.get("uid") == uid for a in accounts):
        set_last("error", "Välj konto", message="Okänt konto.", section="konton")
        return
    st["account_uid"] = uid
    set_last("ok", "Välj konto", message="Konto valt.", section="konton")


def action_check_session(st):
    sess = st.get("session") or {}
    if not sess.get("session_id"):
        set_last("error", "Session", message="Ingen session.", section="konton")
        return
    res = api(
        "GET",
        "/sessions/" + urllib.parse.quote(sess["session_id"], safe=""),
        "/sessions/{id}",
    )
    if not res.ok:
        fail("GET /sessions/{id} misslyckades", res, "konton")
        return
    data = res.data if isinstance(res.data, dict) else {}
    summary = {k: data.get(k) for k in ("status", "authorized", "created", "access", "psu_type")}
    set_last(
        "ok",
        "GET /sessions/{id}",
        message=json.dumps(summary, ensure_ascii=False),
        section="konton",
    )


def psu_headers(st, browser):
    """Headers that tell Enable Banking the user is present.

    Always Psu-Ip-Address and Psu-User-Agent, plus every header the ASPSP lists in
    required_psu_headers that the browser request lets us fill."""
    s = st["settings"]
    ip = (s.get("psu_ip") or "").strip()
    note = []
    if not ip:
        ip = public_ip() or ""
        if ip:
            s["psu_ip"] = ip
            note.append("PSU-IP hämtad från ipify")
    if not ip:
        ip = browser.get("client_ip") or "127.0.0.1"
        note.append("PSU-IP: använder %s (ingen publik IP hittad)" % ip)
    values = {
        "psu-ip-address": ip,
        "psu-user-agent": browser.get("user_agent") or "Mozilla/5.0",
        "psu-referer": "http://localhost:%d/" % CFG["port"],
        "psu-accept": browser.get("accept") or "text/html",
        "psu-accept-charset": browser.get("accept_charset") or "utf-8",
        "psu-accept-encoding": browser.get("accept_encoding") or "gzip, deflate",
        "psu-accept-language": browser.get("accept_language") or "sv-SE,sv;q=0.9",
    }
    wanted = ["psu-ip-address", "psu-user-agent"]
    aspsp = chosen_aspsp(st)
    for h in (aspsp or {}).get("required_psu_headers") or []:
        lh = str(h).lower()
        if lh not in wanted:
            wanted.append(lh)
    headers = {}
    for lh in wanted:
        if lh in values:
            headers[PSU_HEADER_CANON.get(lh, lh)] = values[lh]
        else:
            note.append("kan inte fylla %s" % lh)
    return headers, note


def error_code(res):
    if isinstance(res.data, dict):
        return res.data.get("error") or res.data.get("code")
    return res.error and "network"


def action_fetch(st, mode, browser):
    uid = st.get("account_uid")
    if not uid:
        set_last("error", "Hämta", message="Inget konto valt.", section="hamta")
        return
    s = st["settings"]
    date_from = (dt.date.today() - dt.timedelta(days=int(s["history_days"]))).isoformat()
    query = {"date_from": date_from}
    if s.get("strategy") == "longest":
        query["strategy"] = "longest"
    headers, notes = psu_headers(st, browser) if mode == "user" else ({}, [])
    txs = []
    pages = 0
    empty_pages = 0
    key = None
    truncated = False
    res = None
    path = "/accounts/%s/transactions" % urllib.parse.quote(uid, safe="")
    while True:
        q = dict(query)
        if key:
            q["continuation_key"] = key
        res = api("GET", path, "/accounts/{uid}/transactions", query=q, headers=headers)
        pages += 1
        if not res.ok or not isinstance(res.data, dict):
            break
        page = res.data.get("transactions") or []
        if not page:
            empty_pages += 1
        txs.extend(page)
        key = res.data.get("continuation_key")
        if not key:
            break
        if pages >= MAX_PAGES:
            truncated = True
            break
    entry = {
        "at": now_iso(),
        "mode": mode,
        "strategy": s.get("strategy"),
        "history_days": s["history_days"],
        "date_from": date_from,
        "psu_headers": sorted(headers.keys()),
        "status": res.status if res else 0,
        "pages": pages,
        "empty_pages": empty_pages,
        "truncated": truncated,
    }
    if res is None or not res.ok:
        entry["error"] = error_code(res) if res else "none"
        entry["ok"] = False
        st["fetch_log"].append(entry)
        label = "Uppdatera" if mode == "user" else "Bakgrundshämtning"
        fail("%s: sida %d misslyckades" % (label, pages), res, "hamta")
        if notes:
            LAST["message"] = "; ".join(notes)
        return
    prev = st.get("transactions") or []
    d = diff_transactions(prev, txs)
    dates = [parse_date(date_of(t)) for t in txs]
    dates = [x for x in dates if x]
    entry.update(
        {
            "ok": True,
            "count": len(txs),
            "oldest": min(dates).isoformat() if dates else None,
            "newest": max(dates).isoformat() if dates else None,
            "new": len(d["new"]),
            "changed": len(d["changed"]),
            "disappeared": len(d["disappeared"]),
            "had_previous": d["had_previous"],
        }
    )
    entry.update(d["stats"])
    st["fetch_log"].append(entry)
    st["previous_transactions"] = prev
    st["transactions"] = txs
    st["transactions_meta"] = {
        "at": entry["at"],
        "mode": mode,
        "date_from": date_from,
        "pages": pages,
        "strategy": s.get("strategy"),
    }
    st["last_diff"] = d
    msg = "%d transaktioner, %d sidor. Nya %d, ändrade %d, borta %d." % (
        len(txs),
        pages,
        len(d["new"]),
        len(d["changed"]),
        len(d["disappeared"]),
    )
    if notes:
        msg += " (" + "; ".join(notes) + ")"
    if truncated:
        msg += " Avbröt efter %d sidor." % MAX_PAGES
    label = "Uppdatera" if mode == "user" else "Bakgrundshämtning"
    set_last("ok", label + " (HTTP %d)" % res.status, message=msg, section="kop")


def action_balances(st, browser):
    uid = st.get("account_uid")
    if not uid:
        set_last("error", "Saldo", message="Inget konto valt.", section="hamta")
        return
    headers, notes = psu_headers(st, browser)
    res = api(
        "GET",
        "/accounts/%s/balances" % urllib.parse.quote(uid, safe=""),
        "/accounts/{uid}/balances",
        headers=headers,
    )
    entry = {
        "at": now_iso(),
        "mode": "balances",
        "status": res.status,
        "ok": res.ok,
        "psu_headers": sorted(headers.keys()),
    }
    if not res.ok or not isinstance(res.data, dict):
        entry["error"] = error_code(res)
        st["fetch_log"].append(entry)
        fail("Saldo misslyckades", res, "hamta")
        return
    bal = res.data.get("balances") or []
    entry["balance_types"] = sorted({str(b.get("balance_type")) for b in bal})
    st["fetch_log"].append(entry)
    st["balances"] = {"at": entry["at"], "balances": bal}
    set_last("ok", "Saldo (HTTP %d)" % res.status, message="%d saldon." % len(bal), section="hamta")
    if notes:
        LAST["message"] += " (" + "; ".join(notes) + ")"


def action_disconnect(st):
    sess = st.get("session") or {}
    msg = "Lokalt tillstånd rensat."
    if sess.get("session_id"):
        res = api(
            "DELETE",
            "/sessions/" + urllib.parse.quote(sess["session_id"], safe=""),
            "/sessions/{id}",
        )
        if not res.ok:
            fail("DELETE /sessions/{id} misslyckades (lokalt tillstånd rensat ändå)", res, "koppla")
        else:
            set_last("ok", "Koppla från (HTTP %d)" % res.status, message=msg, section="koppla")
    else:
        set_last("ok", "Koppla från", message=msg, section="koppla")
    # keep settings, the ASPSP list and the aggregate fetch log (no personal content)
    keep = {
        "settings": st["settings"],
        "aspsps": st.get("aspsps") or [],
        "aspsps_fetched_at": st.get("aspsps_fetched_at"),
        "fetch_log": st.get("fetch_log") or [],
    }
    st.clear()
    st.update(default_state())
    st.update(keep)
    st["fetch_log"].append({"at": now_iso(), "mode": "disconnect", "ok": True})


# ----------------------------------------------------------------------------- analysis


def pct(n, d):
    return "%.1f" % (100.0 * n / d) if d else "-"


def analysis_text(st):
    lines = []
    txs = st.get("transactions") or []
    meta = st.get("transactions_meta") or {}
    aspsp = chosen_aspsp(st) or {}
    sess = st.get("session") or {}
    lines.append("K0 analys, genererad %s" % now_iso())
    if aspsp:
        mcv = aspsp.get("maximum_consent_validity")
        lines.append(
            "ASPSP: %s (%s), beta %s, maximum_consent_validity %s s (%s d), psu_types %s"
            % (
                aspsp.get("name"),
                aspsp.get("country"),
                aspsp.get("beta"),
                mcv,
                round(mcv / 86400.0, 1) if isinstance(mcv, (int, float)) else "?",
                ",".join(aspsp.get("psu_types") or []),
            )
        )
        lines.append(
            "auth_methods: %s; required_psu_headers: %s"
            % (
                "; ".join(
                    "%s/%s/%s" % (m.get("name"), m.get("approach"), m.get("psu_type"))
                    for m in aspsp.get("auth_methods") or []
                )
                or "-",
                ",".join(aspsp.get("required_psu_headers") or []) or "-",
            )
        )
    else:
        lines.append("ASPSP: inte vald")
    if sess.get("session_id"):
        accts = sess.get("accounts") or []
        lines.append(
            "Session: giltig till %s; konton %d; valutor %s; cash_account_type %s"
            % (
                (sess.get("valid_until") or "?")[:10],
                len(accts),
                ",".join(sorted({str(a.get("currency")) for a in accts})),
                ",".join(sorted({str(a.get("cash_account_type")) for a in accts})),
            )
        )
    else:
        lines.append("Session: ingen")
    lines.append("")
    n = len(txs)
    lines.append(
        "Senaste lyckade hämtning: %s, läge %s, strategi %s, sidor %s"
        % (meta.get("at", "-"), meta.get("mode", "-"), meta.get("strategy", "-"), meta.get("pages", "-"))
    )
    lines.append("Transaktioner: %d" % n)
    if n:
        dates = sorted(x for x in (parse_date(date_of(t)) for t in txs) if x)
        bdates = sorted(x for x in (parse_date(t.get("booking_date")) for t in txs) if x)
        req = parse_date(meta.get("date_from"))
        oldest = dates[0] if dates else None
        lines.append(
            "Begärt från %s (%s d); äldsta nådda (transaction/booking/value) %s; äldsta booking_date %s; nyaste %s; täckning %s d av begärda %s d"
            % (
                meta.get("date_from"),
                st["settings"].get("history_days"),
                oldest,
                bdates[0] if bdates else "-",
                dates[-1] if dates else "-",
                (dt.date.today() - oldest).days if oldest else "-",
                (dt.date.today() - req).days if req else "-",
            )
        )
        by_status = {}
        by_dir = {}
        for t in txs:
            by_status[str(t.get("status"))] = by_status.get(str(t.get("status")), 0) + 1
            k = str(t.get("credit_debit_indicator"))
            by_dir[k] = by_dir.get(k, 0) + 1
        lines.append("Status: " + ", ".join("%s %d" % kv for kv in sorted(by_status.items())))
        lines.append("Riktning: " + ", ".join("%s %d" % kv for kv in sorted(by_dir.items())))
        non_sek = [t for t in txs if currency_of(t) and currency_of(t) != "SEK"]
        lines.append(
            "Icke-SEK transaction_amount: %d (exchange_rate ifyllt i %d; valutor %s)"
            % (
                len(non_sek),
                sum(1 for t in non_sek if present(t.get("exchange_rate"))),
                ",".join(sorted({currency_of(t) for t in non_sek})) or "-",
            )
        )
        with_er = [t for t in txs if _er(t) is not None]
        ers = [_er(t) for t in with_er]
        er_by_status = {}
        for t in txs:
            s = str(t.get("status"))
            tot, has = er_by_status.get(s, (0, 0))
            er_by_status[s] = (tot + 1, has + (1 if _er(t) is not None else 0))
        lines.append(
            "entry_reference: finns i %d/%d (%s); unika %d; dubbletter %d"
            % (
                len(with_er),
                n,
                ", ".join("%s %d/%d" % (s, h, tt) for s, (tt, h) in sorted(er_by_status.items())),
                len(set(ers)),
                len(ers) - len(set(ers)),
            )
        )
        lines.append(
            "transaction_id: finns i %d/%d"
            % (sum(1 for t in txs if present(t.get("transaction_id"))), n)
        )
        lines.append(
            "Text tom (creditor/debtor-namn, remittance_information, note): %d/%d"
            % (sum(1 for t in txs if not text_of(t)), n)
        )
        lines.append("")
        lines.append("Fältnärvaro (%% av %d rader, ifyllt):" % n)
        for k, c in sorted(field_presence(txs).items()):
            lines.append("  %-45s %6s%%" % (k, pct(c, n)))
    log_entries = st.get("fetch_log") or []
    tx_fetches = [e for e in log_entries if e.get("mode") in ("user", "background")]
    ok_fetches = [e for e in tx_fetches if e.get("ok")]
    lines.append("")
    lines.append(
        "Hämtningar: %d transaktionsanrop (%d lyckade), %d saldoanrop"
        % (
            len(tx_fetches),
            len(ok_fetches),
            sum(1 for e in log_entries if e.get("mode") == "balances"),
        )
    )
    tot = {
        "pdng_to_book_by_entry_reference": 0,
        "pdng_to_book_by_amount_date": 0,
        "er_same": 0,
        "er_changed": 0,
        "er_appeared": 0,
        "er_prev_unmatched": 0,
    }
    for e in ok_fetches:
        for k in tot:
            tot[k] += int(e.get(k) or 0)
    lines.append(
        "PDNG->BOOK (summerat över hämtningar): via entry_reference %d, via belopp+datum (exakt eller ±%d d) %d"
        % (
            tot["pdng_to_book_by_entry_reference"],
            MATCH_WINDOW_DAYS,
            tot["pdng_to_book_by_amount_date"],
        )
    )
    lines.append(
        "entry_reference mellan hämtningar: samma %d, ändrad %d, dök upp (saknades förut) %d, föregående med ER som försvann %d"
        % (tot["er_same"], tot["er_changed"], tot["er_appeared"], tot["er_prev_unmatched"])
    )
    for mode in ("user", "background"):
        last = [e for e in tx_fetches if e.get("mode") == mode]
        if last:
            e = last[-1]
            lines.append(
                "Senaste %s: HTTP %s%s, PSU-headers [%s]"
                % (
                    "Uppdatera" if mode == "user" else "Bakgrundshämtning",
                    e.get("status"),
                    (" " + str(e.get("error"))) if e.get("error") else "",
                    ",".join(e.get("psu_headers") or []),
                )
            )
    bal = [e for e in log_entries if e.get("mode") == "balances"]
    if bal:
        e = bal[-1]
        lines.append(
            "Senaste Saldo: HTTP %s%s, balance_type %s"
            % (
                e.get("status"),
                (" " + str(e.get("error"))) if e.get("error") else "",
                ",".join(e.get("balance_types") or []) or "-",
            )
        )
    if log_entries:
        lines.append("")
        lines.append("Logg (UTC; läge, HTTP, sidor/tomma, rader, från, äldsta, ny/ändr/borta, PDNG->BOOK er/datum):")
        for e in log_entries[-40:]:
            if e.get("mode") in ("user", "background"):
                lines.append(
                    "  %s %-10s %s sidor %s/%s rader %s från %s äldsta %s ny %s ändr %s borta %s pb %s/%s%s"
                    % (
                        e.get("at"),
                        e.get("mode"),
                        e.get("status"),
                        e.get("pages"),
                        e.get("empty_pages"),
                        e.get("count", "-"),
                        e.get("date_from"),
                        e.get("oldest", "-"),
                        e.get("new", "-"),
                        e.get("changed", "-"),
                        e.get("disappeared", "-"),
                        e.get("pdng_to_book_by_entry_reference", "-"),
                        e.get("pdng_to_book_by_amount_date", "-"),
                        (" fel " + str(e.get("error"))) if e.get("error") else "",
                    )
                )
            else:
                lines.append(
                    "  %s %-10s %s%s"
                    % (
                        e.get("at"),
                        e.get("mode"),
                        e.get("status", ""),
                        (" fel " + str(e.get("error"))) if e.get("error") else "",
                    )
                )
    return "\n".join(lines)


# ----------------------------------------------------------------------------- HTML

CSS = """
:root{--fg:#1b1b1b;--bg:#fafafa;--mut:#666;--hl:#fff3c4;--ok:#e6f4ea;--err:#fde7e7;--bd:#ddd}
@media (prefers-color-scheme: dark){:root{--fg:#eee;--bg:#161616;--mut:#999;--hl:#4a3f12;--ok:#173322;--err:#4a1c1c;--bd:#333}}
body{font:14px/1.45 system-ui,sans-serif;color:var(--fg);background:var(--bg);max-width:1200px;margin:0 auto;padding:16px}
h1{font-size:20px}h2{font-size:16px;margin-top:28px;border-bottom:1px solid var(--bd);padding-bottom:4px}
table{border-collapse:collapse;width:100%;margin:8px 0}td,th{border-bottom:1px solid var(--bd);padding:4px 6px;text-align:left;vertical-align:top}
th{font-weight:600}.num{text-align:right;font-variant-numeric:tabular-nums;white-space:nowrap}
.hl{background:var(--hl)}.ok{background:var(--ok);padding:8px}.err{background:var(--err);padding:8px}
.mut{color:var(--mut)}pre{white-space:pre-wrap;word-break:break-word;font-size:12px;margin:4px 0}
textarea{width:100%;font:12px ui-monospace,monospace}button{margin:2px 4px 2px 0}
.tag{font-size:11px;padding:1px 5px;border-radius:3px;border:1px solid var(--bd)}
.scroll{max-height:420px;overflow:auto}
.paste{border:2px solid var(--bd);padding:8px 12px;margin:12px 0}.paste.active{border-color:#c99a00;background:var(--hl)}
"""


def e(s):
    return html.escape("" if s is None else str(s), quote=True)


def render_last():
    if not LAST:
        return ""
    cls = "ok" if LAST.get("kind") == "ok" else "err"
    out = ['<div class="%s" id="svar"><b>%s</b>' % (cls, e(LAST.get("title")))]
    if "status" in LAST:
        out.append(" <span class=tag>HTTP %s</span>" % e(LAST.get("status")))
    out.append(" <span class=mut>%s</span>" % e(LAST.get("at")))
    if LAST.get("message"):
        out.append("<div>%s</div>" % e(LAST["message"]))
    if LAST.get("body") and LAST.get("kind") != "ok":
        out.append("<pre>%s</pre>" % e(LAST["body"]))
    out.append("</div>")
    return "".join(out)


def render_status(st):
    app_id, key_path, problems = load_config()
    sess = st.get("session") or {}
    rows = [
        ("Application id", e(app_id) if app_id else "<b>saknas</b>"),
        (
            "Nyckel",
            ("hittad (%s)" % e(os.path.basename(key_path)))
            if key_path and os.path.isfile(key_path)
            else "<b>saknas</b>",
        ),
        ("openssl", e(openssl_bin())),
        ("API", e(CFG["api_base"])),
        ("Bank", e(st["settings"].get("aspsp_name")) or "<b>inte vald</b>"),
    ]
    if sess.get("session_id"):
        vu = sess.get("valid_until") or ""
        left = ""
        try:
            vdt = dt.datetime.fromisoformat(vu.replace("Z", "+00:00"))
            left = " (%d dagar kvar)" % (vdt - dt.datetime.now(dt.timezone.utc)).days
        except ValueError:
            pass
        rows.append(("Session", "finns, giltig till %s%s" % (e(vu), e(left))))
    else:
        rows.append(("Session", "<b>ingen</b>"))
    acct = current_account(st)
    rows.append(
        (
            "Konto",
            e("%s %s (%s)" % (acct.get("name") or "", acct.get("product") or "", acct.get("currency") or ""))
            if acct
            else "<b>inte valt</b>",
        )
    )
    if st.get("pending_auth"):
        rows.append(("Väntar på", "callback (startad %s)" % e(st["pending_auth"].get("started_at"))))
    out = ["<table>"]
    for k, v in rows:
        out.append("<tr><th>%s</th><td>%s</td></tr>" % (e(k), v))
    out.append("</table>")
    if problems:
        out.append('<div class="err">%s</div>' % "<br>".join(e(p) for p in problems))
    return "".join(out)


def current_account(st):
    uid = st.get("account_uid")
    for a in (st.get("session") or {}).get("accounts") or []:
        if a.get("uid") == uid:
            return a
    return None


def render_aspsps(st):
    out = [
        '<form method=post action="/action/list_aspsps"><button>Lista banker</button>'
        ' <span class=mut>GET /aspsps?country=SE %s</span></form>'
        % (("(hämtad " + e(st.get("aspsps_fetched_at")) + ")") if st.get("aspsps_fetched_at") else "")
    ]
    aspsps = st.get("aspsps") or []
    if aspsps:
        chosen = st["settings"].get("aspsp_name")
        ordered = sorted(
            aspsps, key=lambda a: (0 if HIGHLIGHT.search(a.get("name") or "") else 1, a.get("name") or "")
        )
        out.append('<form method=post action="/action/choose_aspsp"><div class=scroll><table>')
        out.append(
            "<tr><th></th><th>Namn</th><th>beta</th><th>Max samtycke</th><th>psu_types</th>"
            "<th>auth_methods</th><th>required_psu_headers</th></tr>"
        )
        for a in ordered:
            name = a.get("name") or ""
            mcv = a.get("maximum_consent_validity")
            methods = "; ".join(
                "%s/%s/%s" % (m.get("name"), m.get("approach"), m.get("psu_type"))
                for m in a.get("auth_methods") or []
            )
            out.append(
                '<tr class="%s"><td><input type=radio name=aspsp value="%s"%s></td><td>%s</td><td>%s</td>'
                '<td class=num>%s</td><td>%s</td><td>%s</td><td>%s</td></tr>'
                % (
                    "hl" if HIGHLIGHT.search(name) else "",
                    e(name),
                    " checked" if name == chosen else "",
                    e(name),
                    e(a.get("beta")),
                    e("%.0f d" % (mcv / 86400.0)) if isinstance(mcv, (int, float)) else "?",
                    e(",".join(a.get("psu_types") or [])),
                    e(methods),
                    e(",".join(a.get("required_psu_headers") or [])),
                )
            )
        out.append("</table></div><button>Välj bank</button></form>")
    a = chosen_aspsp(st)
    if a:
        out.append("<p><b>Vald: %s</b></p><pre>%s</pre>" % (e(a.get("name")), e(json.dumps(a, ensure_ascii=False, indent=2))))
    return "".join(out)


def render_connect(st):
    s = st["settings"]
    a = chosen_aspsp(st)
    secs = consent_seconds(a)
    days_opts = "".join(
        '<option value="%d"%s>%d dagar</option>' % (d, " selected" if d == s["history_days"] else "", d)
        for d in HISTORY_CHOICES
    )
    redirect = s.get("redirect_url")
    radios = []
    for r in REDIRECT_CHOICES:
        radios.append(
            '<label><input type=radio name=redirect_choice value="%s"%s> %s</label><br>'
            % (e(r), " checked" if r == redirect else "", e(r))
        )
    custom = redirect not in REDIRECT_CHOICES
    radios.append(
        '<label><input type=radio name=redirect_choice value=custom%s> annan: </label>'
        '<input name=redirect_custom size=60 value="%s">'
        % (" checked" if custom else "", e(redirect if custom else ""))
    )
    methods = [m for m in (a or {}).get("auth_methods") or [] if m.get("psu_type") in (None, "personal")]
    m_opts = ['<option value="">(bankens standard)</option>']
    for m in methods:
        m_opts.append(
            '<option value="%s"%s>%s (%s)</option>'
            % (e(m.get("name")), " selected" if m.get("name") == s.get("auth_method") else "", e(m.get("name")), e(m.get("approach")))
        )
    strat = "".join(
        '<option value="%s"%s>%s</option>' % (v, " selected" if v == s.get("strategy") else "", v)
        for v in ("default", "longest")
    )
    out = [
        '<form method=post action="/action/settings">',
        "<p>Historik: <select name=history_days>%s</select> &nbsp; strategi: <select name=strategy>%s</select>"
        ' <span class=mut>(longest = Enable Bankings strategy=longest)</span></p>' % (days_opts, strat),
        "<p>Redirect URL (måste finnas registrerad i Enable Banking-appen):<br>%s</p>" % "".join(radios),
        "<p>auth_method: <select name=auth_method>%s</select></p>" % "".join(m_opts),
        "<p class=mut>access.valid_until = nu + %s dagar (%s, max 180) − 5 min; psu_type personal; aspsp {%s, SE}</p>"
        % (
            "%.1f" % (secs / 86400.0),
            "bankens maximum_consent_validity" if a else "reserv 90 d, bankens max okänt",
            e(s.get("aspsp_name") or "?"),
        ),
        '<button name=op value=save>Spara</button> <button name=op value=auth><b>Anslut med BankID</b></button>',
        "</form>",
    ]
    pending = st.get("pending_auth") or {}
    box_cls = "paste active" if pending else "paste"
    out.append('<div class="%s" id=klistra><h3>Klistra in adressen du hamnade på</h3>' % box_cls)
    out.append("<p><b>%s</b></p>" % e(PASTE_HINT))
    if pending.get("url"):
        out.append(
            '<p class=mut>Bankens sida öppnades i en ny flik (startad %s). '
            '<a href="%s" target=_blank rel=noopener>Öppna bankens sida igen</a></p>'
            % (e(pending.get("started_at")), e(pending["url"]))
        )
    out.append(
        '<form method=post action="/action/paste"><input name=url style="width:80%%" autofocus '
        'placeholder="https://kvitt.suorra.duckdns.org/cards/callback?code=...&amp;state=..."> '
        "<button><b>Använd adressen</b></button></form></div>"
    )
    out.append(
        "<p class=mut>Reserv: http://localhost:%d/callback hanteras också direkt om den adressen "
        "någon gång kan registreras (kontrollpanelen nekar http i dag).</p>" % CFG["port"]
    )
    return "".join(out)


def render_accounts(st):
    sess = st.get("session") or {}
    if not sess.get("session_id"):
        return "<p class=mut>Ingen session ännu.</p>"
    out = [
        "<p>Session %s, giltig till %s, aspsp %s, psu_type %s, skapad %s</p>"
        % (
            e(mask(sess.get("session_id"))),
            e(sess.get("valid_until")),
            e((sess.get("aspsp") or {}).get("name")),
            e(sess.get("psu_type")),
            e(sess.get("created_at")),
        ),
        '<form method=post action="/action/check_session"><button>Kontrollera session</button> <span class=mut>GET /sessions/{id}</span></form>',
        '<form method=post action="/action/choose_account"><table><tr><th></th><th>uid</th><th>name</th><th>product / details</th>'
        "<th>valuta</th><th>cash_account_type / usage</th><th>account_id (maskerat)</th><th>all_account_ids</th><th>identification_hash</th><th>credit_limit</th></tr>",
    ]
    for a in sess.get("accounts") or []:
        out.append(
            "<tr><td><input type=radio name=account_uid value=\"%s\"%s></td><td><code>%s</code></td><td>%s</td><td>%s</td>"
            "<td>%s</td><td>%s</td><td><pre>%s</pre></td><td><pre>%s</pre></td><td>%s</td><td>%s</td></tr>"
            % (
                e(a.get("uid")),
                " checked" if a.get("uid") == st.get("account_uid") else "",
                e(a.get("uid")),
                e(a.get("name")),
                e(" / ".join(x for x in (a.get("product"), a.get("details")) if x)),
                e(a.get("currency")),
                e(" / ".join(str(x) for x in (a.get("cash_account_type"), a.get("usage")) if x)),
                e(json.dumps(mask_ids(a.get("account_id")), ensure_ascii=False)),
                e(json.dumps(mask_ids(a.get("all_account_ids")), ensure_ascii=False)),
                e(mask(a.get("identification_hash") or "")),
                "finns" if present(a.get("credit_limit")) else "-",
            )
        )
    out.append("</table><button>Välj konto</button></form>")
    return "".join(out)


def render_fetch(st):
    s = st["settings"]
    meta = st.get("transactions_meta") or {}
    out = [
        '<form method=post action="/action/fetch">',
        "<p>PSU IP-adress: <input name=psu_ip size=20 value=\"%s\"> <span class=mut>(tom = hämtas från api.ipify.org vid första Uppdatera)</span></p>"
        % e(s.get("psu_ip")),
        "<p>Från %s (idag − %d dagar), strategi %s</p>"
        % (
            e((dt.date.today() - dt.timedelta(days=int(s["history_days"]))).isoformat()),
            int(s["history_days"]),
            e(s.get("strategy")),
        ),
        "<button name=op value=user><b>Uppdatera</b></button> <span class=mut>med PSU-headers (användaren närvarande)</span><br>",
        "<button name=op value=background>Bakgrundshämtning</button> <span class=mut>utan PSU-headers</span><br>",
        "<button name=op value=balances>Saldo</button> <span class=mut>GET /accounts/{uid}/balances med PSU-headers</span>",
        "</form>",
    ]
    if meta:
        out.append(
            "<p class=mut>Senast hämtat %s (%s, %s sidor).</p>"
            % (e(meta.get("at")), e(meta.get("mode")), e(meta.get("pages")))
        )
    bal = st.get("balances")
    if bal:
        out.append("<table><tr><th>name</th><th>balance_type</th><th class=num>belopp</th><th>valuta</th><th>reference_date</th><th>last_change_date_time</th></tr>")
        for b in bal.get("balances") or []:
            ba = b.get("balance_amount") or {}
            out.append(
                "<tr><td>%s</td><td>%s</td><td class=num>%s</td><td>%s</td><td>%s</td><td>%s</td></tr>"
                % (
                    e(b.get("name")),
                    e(b.get("balance_type")),
                    e(ba.get("amount")),
                    e(ba.get("currency")),
                    e(b.get("reference_date")),
                    e(b.get("last_change_date_time")),
                )
            )
        out.append("</table><p class=mut>Saldo hämtat %s</p>" % e(bal.get("at")))
    return "".join(out)


def tx_row(t, tag, raw_id):
    amt = amount_of(t)
    return (
        "<tr><td>%s</td><td>%s</td><td>%s</td><td class=num>%s</td><td>%s</td><td>%s</td><td>%s</td>"
        "<td><code>%s</code></td><td><details><summary>rå</summary><pre>%s</pre></details></td></tr>"
        % (
            e(t.get("booking_date") or "-") + "<br><span class=mut>" + e(t.get("transaction_date") or "-") + "</span>",
            tag,
            e(text_of(t) or "-"),
            e(fmt_kr(amt)) + " kr",
            e(currency_of(t)),
            e(t.get("credit_debit_indicator")),
            e(t.get("status")),
            e(t.get("entry_reference") or "-"),
            e(json.dumps(t, ensure_ascii=False, indent=2)),
        )
    )


def render_purchases(st):
    txs = st.get("transactions") or []
    prev = st.get("previous_transactions") or []
    d = st.get("last_diff") or {}
    if not txs and not d:
        return "<p class=mut>Inget hämtat ännu.</p>"
    new = set(d.get("new") or [])
    changed = {c["idx"]: c for c in d.get("changed") or []}
    out = []
    if d.get("had_previous"):
        out.append(
            "<p>Jämfört med förra hämtningen: <b>%d nya</b>, <b>%d ändrade</b>, <b>%d borta</b>.</p>"
            % (len(new), len(changed), len(d.get("disappeared") or []))
        )
    else:
        out.append("<p class=mut>Första hämtningen; inget att jämföra med.</p>")
    out.append(
        "<table><tr><th>booking_date<br><span class=mut>transaction_date</span></th><th>diff</th><th>text</th>"
        "<th class=num>belopp</th><th>valuta</th><th>riktning</th><th>status</th><th>entry_reference</th><th></th></tr>"
    )
    order = sorted(range(len(txs)), key=lambda j: (str(date_of(txs[j]) or ""), j), reverse=True)
    for j in order:
        t = txs[j]
        tag = ""
        if d.get("had_previous") and j in new:
            tag = "<span class=tag>ny</span>"
        elif j in changed:
            c = changed[j]
            p = prev[c["prev_idx"]] if c["prev_idx"] < len(prev) else {}
            st_change = ""
            if p.get("status") != t.get("status"):
                st_change = "%s→%s " % (p.get("status"), t.get("status"))
            tag = "<span class=tag>ändrad %s</span><br><span class=mut>%s; via %s</span>" % (
                e(st_change),
                e(",".join(c["fields"])),
                e(c["by"]),
            )
        out.append(tx_row(t, tag, j))
    out.append("</table>")
    gone = d.get("disappeared") or []
    if gone:
        out.append("<h3>Försvunna sedan förra hämtningen</h3><table>")
        for i in gone:
            if i < len(prev):
                out.append(tx_row(prev[i], "<span class=tag>borta</span>", "p%d" % i))
        out.append("</table>")
    return "".join(out)


def render_page(st):
    text = analysis_text(st)
    return """<!doctype html><html lang=sv><head><meta charset=utf-8>
<meta name=viewport content="width=device-width, initial-scale=1"><title>K0 Enable Banking</title><style>%s</style></head><body>
<h1>K0: Enable Banking-test</h1>
%s
<h2 id=status>1. Status</h2>%s
<h2 id=banker>2. Banker</h2>%s
<h2 id=anslut>3–4. Anslut</h2>%s
<h2 id=konton>5. Konton</h2>%s
<h2 id=hamta>6. Hämta</h2>%s
<h2 id=kop>7. Köp</h2>%s
<h2 id=analys>8. Analys</h2>
<p class=mut>Bara aggregat; inga texter, belopp eller id:n. <button type=button onclick="navigator.clipboard.writeText(document.getElementById('analys-text').value).then(function(){this.textContent='Kopierad'}.bind(this))">Kopiera analys</button></p>
<textarea id=analys-text rows=24 readonly>%s</textarea>
<h2 id=koppla>9. Koppla från</h2>
<form method=post action="/action/disconnect" onsubmit="return confirm('Koppla från och rensa sessionen?')"><button>Koppla från</button>
<span class=mut>DELETE /sessions/{id}; rensar session, konton och transaktioner lokalt (inställningar, bankernas lista och den aggregerade loggen behålls)</span></form>
</body></html>""" % (
        CSS,
        render_last(),
        render_status(st),
        render_aspsps(st),
        render_connect(st),
        render_accounts(st),
        render_fetch(st),
        render_purchases(st),
        e(text),
    )


# ----------------------------------------------------------------------------- server


class Handler(BaseHTTPRequestHandler):
    server_version = "K0/1"

    def log_request(self, code="-", size="-"):
        path = self.path.split("?", 1)[0]
        log("LOCAL %s %s %s" % (self.command, path, code))

    def log_message(self, fmt, *args):  # silence the default request-line logging
        return

    def send_html(self, body, status=200):
        raw = body.encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(raw)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(raw)

    def redirect(self, location):
        self.send_response(303)
        self.send_header("Location", location)
        self.send_header("Content-Length", "0")
        self.end_headers()

    def browser_info(self):
        return {
            "client_ip": self.client_address[0],
            "user_agent": self.headers.get("User-Agent"),
            "accept": self.headers.get("Accept"),
            "accept_charset": self.headers.get("Accept-Charset"),
            "accept_encoding": self.headers.get("Accept-Encoding"),
            "accept_language": self.headers.get("Accept-Language"),
        }

    def do_GET(self):
        parsed = urllib.parse.urlparse(self.path)
        if parsed.path == "/":
            with LOCK:
                st = load_state()
                self.send_html(render_page(st))
            return
        if parsed.path in ("/callback", "/cards/callback"):
            params = extract_code_state("?" + parsed.query)
            with LOCK:
                st = load_state()
                complete_auth(st, params, "callback")
                save_state(st)
            self.redirect("/#svar")
            return
        if parsed.path == "/favicon.ico":
            self.send_response(204)
            self.end_headers()
            return
        self.send_html("<p>Hittades inte.</p>", 404)

    def same_origin(self):
        origin = self.headers.get("Origin") or self.headers.get("Referer") or ""
        if not origin:
            return True  # curl and some browsers omit both on same-origin form posts
        host = urllib.parse.urlparse(origin).netloc
        return host in ("localhost:%d" % CFG["port"], "127.0.0.1:%d" % CFG["port"])

    def do_POST(self):
        parsed = urllib.parse.urlparse(self.path)
        if not parsed.path.startswith("/action/") or not self.same_origin():
            self.send_html("<p>Nekad.</p>", 403)
            return
        length = int(self.headers.get("Content-Length") or 0)
        raw = self.rfile.read(length).decode("utf-8", "replace") if length else ""
        form = {k: v[0] for k, v in urllib.parse.parse_qs(raw, keep_blank_values=True).items()}
        name = parsed.path[len("/action/"):]
        target = "/#svar"
        with LOCK:
            st = load_state()
            if name == "list_aspsps":
                action_list_aspsps(st)
            elif name == "choose_aspsp":
                action_choose_aspsp(st, form)
            elif name == "settings":
                save_settings(st, form)
                if form.get("op") == "auth":
                    url = action_auth(st)
                    if url:
                        # the bank page opens in its own tab; this tab shows the paste box
                        if CFG["open_browser"]:
                            webbrowser.open_new_tab(url)
                        target = "/#klistra"
                else:
                    set_last("ok", "Inställningar sparade", section="anslut")
            elif name == "paste":
                complete_auth(st, extract_code_state(form.get("url", "")), "inklistrad adress")
            elif name == "choose_account":
                action_choose_account(st, form)
            elif name == "check_session":
                action_check_session(st)
            elif name == "fetch":
                st["settings"]["psu_ip"] = form.get("psu_ip", "").strip()
                op = form.get("op")
                if op in ("user", "background"):
                    action_fetch(st, op, self.browser_info())
                elif op == "balances":
                    action_balances(st, self.browser_info())
            elif name == "disconnect":
                action_disconnect(st)
            else:
                self.send_html("<p>Okänd åtgärd.</p>", 404)
                return
            save_state(st)
        self.redirect(target)


class V6Server(ThreadingHTTPServer):
    address_family = socket.AF_INET6


def main(argv=None):
    p = argparse.ArgumentParser(description="K0 Enable Banking test app")
    p.add_argument("--port", type=int, default=DEFAULT_PORT)
    p.add_argument("--local-dir", default=DEFAULT_LOCAL_DIR, help="config and state folder")
    p.add_argument("--api-base", default=DEFAULT_API_BASE, help="for tests against a fake API")
    p.add_argument("--no-browser", action="store_true")
    args = p.parse_args(argv)
    CFG["port"] = args.port
    CFG["local_dir"] = os.path.abspath(args.local_dir)
    CFG["api_base"] = args.api_base
    CFG["open_browser"] = not args.no_browser
    os.makedirs(CFG["local_dir"], exist_ok=True)
    try:
        os.chmod(CFG["local_dir"], 0o700)
    except OSError:
        pass
    server = ThreadingHTTPServer(("127.0.0.1", args.port), Handler)
    # browsers may resolve localhost to ::1 first; serve the loopback there too if possible
    try:
        server6 = V6Server(("::1", args.port), Handler)
        threading.Thread(target=server6.serve_forever, daemon=True).start()
    except OSError:
        pass
    url = "http://localhost:%d/" % args.port
    log("K0 listening on %s (state in %s)" % (url, CFG["local_dir"]))
    if not args.no_browser:
        threading.Timer(0.5, lambda: webbrowser.open(url)).start()
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        log("stopped")
    return 0


if __name__ == "__main__":
    sys.exit(main())