taler-monitoring/ladder_extract_rpubs.py

#!/usr/bin/env python3
"""Extract reserve_pub candidates from wallet accept + transactions dumps.

Used by check_amount_ladder.sh force-select. Kept as a file (not a bash
heredoc) so shell refactors cannot turn Python ``continue`` into ``return``.

Usage:
  ladder_extract_rpubs.py WID AMT accept.out tx.json used-rpubs.txt
Prints one reserve pub per line, preferred first.
"""
from __future__ import annotations

import json
import re
import sys
from pathlib import Path


def main(argv: list[str]) -> int:
    if len(argv) < 6:
        print(
            "usage: ladder_extract_rpubs.py WID AMT accept.out tx.json used-rpubs.txt",
            file=sys.stderr,
        )
        return 2

    wid = argv[1]
    amt = argv[2]
    paths = argv[3:5]
    used_path = argv[5]
    used: set[str] = set()
    if Path(used_path).is_file():
        used = {ln.strip() for ln in open(used_path) if ln.strip()}

    amt_num = amt.split(":", 1)[-1] if amt else ""
    dec = json.JSONDecoder()

    def walk_collect(o, bag, ctx=None):
        ctx = dict(ctx or {})
        if isinstance(o, dict):
            for k, v in o.items():
                if not isinstance(v, str):
                    continue
                kl = str(k).lower().replace("-", "_")
                if kl in (
                    "withdrawal_id",
                    "withdraw_id",
                    "wopid",
                    "id",
                    "transactionid",
                    "transaction_id",
                ):
                    ctx["id"] = v
                elif kl in (
                    "amount",
                    "rawamount",
                    "instructedamount",
                    "amounteffective",
                    "amountraw",
                    "transferamount",
                ):
                    ctx["amount"] = v
                elif kl in (
                    "taler_withdraw_uri",
                    "talerwithdrawuri",
                    "uri",
                    "bankconfirmationurl",
                    "confirmtransferurl",
                ):
                    ctx["uri"] = v
            for k, v in o.items():
                kl = str(k).lower().replace("-", "_")
                if kl in ("reserve_pub", "reservepub") and isinstance(v, str) and len(v) >= 40:
                    bag.append((v, dict(ctx)))
                elif not isinstance(v, (str, int, float, bool)) and v is not None:
                    walk_collect(v, bag, ctx)
        elif isinstance(o, list):
            for i in o:
                walk_collect(i, bag, ctx)

    def score_ctx(pub, ctx, window=""):
        score = 0
        cid = str(ctx.get("id") or "")
        camt = str(ctx.get("amount") or "")
        curi = str(ctx.get("uri") or "")
        if wid and wid in curi:
            score += 120
        if wid and wid in cid:
            score += 100
        if wid and wid in window:
            score += 80
        if amt and (camt == amt or camt.endswith(amt_num)):
            score += 50
        if amt and amt in window:
            score += 30
        if pub in used:
            score -= 200
        return score

    def iter_json_objects(text: str):
        i, n = 0, len(text)
        while i < n:
            if text[i] == "{":
                try:
                    o, end = dec.raw_decode(text, i)
                    yield o
                    i = end
                    continue
                except Exception:
                    pass
            i += 1

    raw_pubs: list[str] = []
    scored: list[tuple[int, str]] = []

    for p in paths:
        try:
            t = open(p, errors="replace").read()
        except Exception:
            continue
        for o in iter_json_objects(t):
            bag = []
            walk_collect(o, bag)
            for pub, ctx in bag:
                raw_pubs.append(pub)
                scored.append((score_ctx(pub, ctx), pub))

    rpub_pats = (
        r"\"reservePub\"\s*:\s*\"([A-Z0-9]{40,})\"",
        r"\"reserve_pub\"\s*:\s*\"([A-Z0-9]{40,})\"",
        r"reserve[_ ]?pub[\"\s:=]+([A-Z0-9]{40,})",
    )
    for p in paths:
        try:
            blob = open(p, errors="replace").read()
        except Exception:
            continue
        for pat in rpub_pats:
            for m in re.finditer(pat, blob, re.I):
                pub = m.group(1)
                raw_pubs.append(pub)
                window = blob[max(0, m.start() - 600) : m.end() + 600]
                scored.append((score_ctx(pub, {}, window) + 10, pub))

    order: list[str] = []
    seen: set[str] = set()
    for score, pub in sorted(scored, key=lambda x: x[0], reverse=True):
        if pub in seen or pub in used:
            continue
        seen.add(pub)
        order.append(pub)
    for pub in reversed(raw_pubs):
        if pub in seen or pub in used:
            continue
        seen.add(pub)
        order.append(pub)
    for pub in order:
        print(pub)
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))