koopa-admin-log/scripts/taler-landing/goa_amounts.py

#!/usr/bin/env python3
"""GOA (and generic CUR:amount) alt-unit formatting for landing stats.

Display uses compact SI-style names (kGOA, MGOA, GGOA, …) so tiles stay short.
Exchange /config may still send long names (Kilo-GOA); we compact them.
"""
from __future__ import annotations

from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
from typing import Any, Dict, Optional, Tuple

# Compact display ladder (preferred on landings)
DEFAULT_ALT: Dict[str, str] = {
    "24": "YGOA",
    "21": "ZGOA",
    "18": "EGOA",
    "15": "PGOA",
    "12": "TGOA",
    "9": "GGOA",
    "6": "MGOA",
    "3": "kGOA",
    "0": "GOA",
    "-1": "dGOA",
    "-2": "cGOA",
    "-3": "mGOA",
    "-6": "µGOA",
    "-7": "dµGOA",
    "-8": "aGOA",
}

# Long names from exchange /config → compact
_LONG_TO_COMPACT = {
    "yotta-goa": "YGOA",
    "zetta-goa": "ZGOA",
    "exa-goa": "EGOA",
    "peta-goa": "PGOA",
    "tera-goa": "TGOA",
    "giga-goa": "GGOA",
    "mega-goa": "MGOA",
    "kilo-goa": "kGOA",
    "goa": "GOA",
    "deci-goa": "dGOA",
    "centi-goa": "cGOA",
    "milli-goa": "mGOA",
    "micro-goa": "µGOA",
    "deci-micro-goa": "dµGOA",
    "atomic-goa": "aGOA",
}

# Use alt unit when |value| >= this (base units). Below: keep CUR:n.
ALT_THRESHOLD = Decimal("1000")


def compact_alt_map(alt: Optional[Dict[str, str]] = None) -> Dict[str, str]:
    """Normalize any alt_unit_names map to compact display labels."""
    base = dict(DEFAULT_ALT)
    if not alt:
        return base
    out: Dict[str, str] = {}
    for k, v in alt.items():
        sk = str(k)
        name = str(v).strip()
        low = name.lower().replace(" ", "")
        if low in _LONG_TO_COMPACT:
            out[sk] = _LONG_TO_COMPACT[low]
        elif sk in DEFAULT_ALT and (
            "goa" in low or name == DEFAULT_ALT.get(sk) or name in DEFAULT_ALT.values()
        ):
            # keep scale key, prefer compact for GOA family
            out[sk] = DEFAULT_ALT.get(sk, name)
        else:
            out[sk] = name
    # ensure full default ladder for missing scales
    for sk, name in DEFAULT_ALT.items():
        out.setdefault(sk, name)
    return out


def parse_amount(s: Any) -> Tuple[str, Decimal]:
    """Parse 'GOA:12.5' or bare number → (currency, value)."""
    if s is None:
        return "GOA", Decimal(0)
    if isinstance(s, (int, float, Decimal)):
        return "GOA", Decimal(str(s))
    text = str(s).strip()
    if not text:
        return "GOA", Decimal(0)
    if ":" in text:
        cur, rest = text.split(":", 1)
        return (cur or "GOA").strip(), Decimal(rest.strip() or "0")
    return "GOA", Decimal(text)


def fmt_coeff(v: Decimal) -> str:
    if v == v.to_integral_value():
        return format(int(v), "d")
    q = v.quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP)
    s = format(q, "f").rstrip("0").rstrip(".")
    return s or "0"


def fmt_base(cur: str, val: Decimal) -> str:
    if val == val.to_integral_value():
        return f"{cur}:{int(val)}"
    s = format(val, "f").rstrip("0").rstrip(".")
    return f"{cur}:{s}"


def format_amount_alt(
    amount: Any,
    alt: Optional[Dict[str, str]] = None,
    *,
    threshold: Decimal = ALT_THRESHOLD,
    with_base: bool = False,
) -> Dict[str, Any]:
    """Return display fields for one amount.

    Keys:
      amount       canonical CUR:n
      amount_alt   short display (e.g. "1.23 MGOA")
      amount_full  "1.23 MGOA (GOA:1230000)" when alt used
      value        float for JSON
      value_str    exact decimal string
    """
    alt = compact_alt_map(alt)
    try:
        cur, val = parse_amount(amount)
    except (InvalidOperation, ValueError, ArithmeticError):
        raw = str(amount or "")
        return {
            "amount": raw,
            "amount_alt": raw or "—",
            "amount_full": raw or "—",
            "value": 0.0,
            "value_str": "0",
        }

    base = fmt_base(cur, val)
    value_str = (
        format(val, "f").rstrip("0").rstrip(".")
        if val != val.to_integral_value()
        else str(int(val))
    )
    try:
        value_f = float(val)
    except Exception:
        value_f = 0.0

    base_name = alt.get("0") or cur
    cur_u = (cur or "").upper()
    base_u = str(base_name).upper()
    # Foreign currencies must not pick up kGOA / MGOA labels
    if cur_u and cur_u != "GOA" and ("GOA" in base_u or base_u.endswith("GOA")):
        return {
            "amount": base,
            "amount_alt": base,
            "amount_full": base,
            "value": value_f,
            "value_str": value_str,
        }

    if val == 0:
        return {
            "amount": base,
            "amount_alt": base,
            "amount_full": base,
            "value": 0.0,
            "value_str": "0",
        }

    scales = []
    for k, name in alt.items():
        try:
            scales.append((int(k), str(name)))
        except Exception:
            continue
    scales.sort(key=lambda x: -x[0])

    absval = abs(val)
    if absval < threshold:
        return {
            "amount": base,
            "amount_alt": base,
            "amount_full": base,
            "value": value_f,
            "value_str": value_str,
        }

    chosen = None
    for sc, name in scales:
        unit = Decimal(10) ** sc
        if unit <= 0:
            continue
        coeff = absval / unit
        if coeff >= 1:
            chosen = (sc, name, coeff if val >= 0 else -coeff)
            break

    if chosen is None or chosen[0] == 0:
        return {
            "amount": base,
            "amount_alt": base,
            "amount_full": base,
            "value": value_f,
            "value_str": value_str,
        }

    sc, name, coeff = chosen
    short = f"{fmt_coeff(coeff)} {name}"
    full = f"{short} ({base})"
    return {
        "amount": base,
        "amount_alt": short,
        "amount_full": full,
        "value": value_f,
        "value_str": value_str,
    }


def attach_alt(
    obj: Dict[str, Any], amount_key: str = "amount", alt: Optional[Dict[str, str]] = None
) -> Dict[str, Any]:
    """Mutate obj: ensure amount_alt / amount_full from amount_key."""
    if not isinstance(obj, dict):
        return obj
    src = obj.get(amount_key)
    if src is None:
        return obj
    f = format_amount_alt(src, alt)
    obj[amount_key] = f["amount"]
    obj["amount_alt"] = f["amount_alt"]
    obj["amount_full"] = f["amount_full"]
    if "value" not in obj:
        obj["value"] = f["value"]
    obj["value_str"] = f["value_str"]
    return obj


def load_alt_from_config(url: str, timeout: float = 8.0) -> Dict[str, str]:
    """GET exchange/bank /config → compact alt_unit_names (fallback DEFAULT_ALT)."""
    import json
    import urllib.request

    try:
        req = urllib.request.Request(url, headers={"Accept": "application/json"})
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            data = json.loads(resp.read().decode("utf-8", errors="replace"))
    except Exception:
        return dict(DEFAULT_ALT)

    au = None
    cs = data.get("currency_specification")
    if isinstance(cs, dict):
        au = cs.get("alt_unit_names")
    if not au and isinstance(data.get("currencies"), dict):
        for _code, spec in data["currencies"].items():
            if isinstance(spec, dict) and spec.get("alt_unit_names"):
                au = spec["alt_unit_names"]
                break
    if not isinstance(au, dict) or "0" not in au:
        return dict(DEFAULT_ALT)
    return compact_alt_map({str(k): str(v) for k, v in au.items()})


def enrich_stats_tree(data: Dict[str, Any], alt: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
    """Walk common landing stats.json shapes and add amount_alt fields."""
    alt = compact_alt_map(alt)
    if not isinstance(data, dict):
        return data

    # bank-style nested blocks
    for path in (
        ("flow", "incoming"),
        ("flow", "withdraw"),
        ("flow", "other_out"),
        ("withdraws",),
        ("withdraws", "last_24h"),
        ("withdraws", "last_7d"),
    ):
        cur: Any = data
        ok = True
        for p in path:
            if not isinstance(cur, dict) or p not in cur:
                ok = False
                break
            cur = cur[p]
        if ok and isinstance(cur, dict):
            if "amount" in cur:
                attach_alt(cur, "amount", alt)
            if "total_amount" in cur:
                f = format_amount_alt(cur["total_amount"], alt)
                cur["total_amount"] = f["amount"]
                cur["total_amount_alt"] = f["amount_alt"]
                cur["total_amount_full"] = f["amount_full"]
            if "last_amount" in cur and cur["last_amount"]:
                f = format_amount_alt(cur["last_amount"], alt)
                cur["last_amount"] = f["amount"]
                cur["last_amount_alt"] = f["amount_alt"]
                cur["last_amount_full"] = f["amount_full"]

    if isinstance(data.get("flow"), dict):
        fl = data["flow"]
        for k in ("total_in", "total_out"):
            if fl.get(k):
                f = format_amount_alt(fl[k], alt)
                fl[k] = f["amount"]
                fl[f"{k}_alt"] = f["amount_alt"]
                fl[f"{k}_full"] = f["amount_full"]

    if data.get("balance_explorer"):
        f = format_amount_alt(data["balance_explorer"], alt)
        data["balance_explorer"] = f["amount"]
        data["balance_explorer_alt"] = f["amount_alt"]
        data["balance_explorer_full"] = f["amount_full"]

    for key in ("recent_withdraws", "recent_incoming", "recent_activity"):
        rows = data.get(key)
        if isinstance(rows, list):
            for row in rows:
                if isinstance(row, dict) and row.get("amount"):
                    attach_alt(row, "amount", alt)

    # exchange-style top-level amounts
    for k in (
        "wire_in_amount",
        "withdraw_amount",
        "coins_remaining_amount",
    ):
        if data.get(k):
            f = format_amount_alt(data[k], alt)
            data[k] = f["amount"]
            data[f"{k}_alt"] = f["amount_alt"]
            data[f"{k}_full"] = f["amount_full"]

    for key in ("by_denom", "denom_ladder"):
        rows = data.get(key)
        if isinstance(rows, list):
            for row in rows:
                if isinstance(row, dict) and row.get("value"):
                    f = format_amount_alt(row["value"], alt)
                    row["value"] = f["amount"]
                    row["value_alt"] = f["amount_alt"]
                    row["value_full"] = f["amount_full"]

    # merchant dual currency
    for block in data.get("by_currency") or []:
        if not isinstance(block, dict):
            continue
        for k in ("amount_sum", "amount_paid_sum"):
            if block.get(k):
                f = format_amount_alt(block[k], alt)
                block[k] = f["amount"]
                block[f"{k}_alt"] = f["amount_alt"]
                block[f"{k}_full"] = f["amount_full"]

    for block in data.get("recent_activity_by_currency") or []:
        if not isinstance(block, dict):
            continue
        for row in block.get("items") or []:
            if isinstance(row, dict) and row.get("amount"):
                attach_alt(row, "amount", alt)

    data["alt_unit_names"] = alt
    # small legend for UI (positive scales only, compact)
    data["alt_unit_legend"] = [
        {"scale": int(k), "name": alt[k], "factor": f"10^{k}"}
        for k in sorted(alt.keys(), key=lambda x: -int(x))
        if int(k) >= 0
    ]
    return data