#!/usr/bin/env python3"""Keyless chain reads from public NEAR RPC: protocol parameters, validators, revenue-wallet balances and control.Weekly wallet history is incremental: archival RPC is slow, so only missing weeks are read."""import datetime, gzip, csv, pathlib, signal, sys, timesys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))from common import ARCHIVAL, ROOT, StopSource, load, now_utc, rpc_call, rpc_view, saveWALLETS = {    "buybacks.multisignature.near": "Buyback multisig",    "fefundsadmin.sputnik-dao.near": "Front-end fee DAO",    "1csfundsadmin.sputnik-dao.near": "1Click fee DAO",    "treasury.near": "Protocol treasury",}FEE_COLLECTOR_FALLBACK = "7066024d3f20f94de601c003163367873cca78507eeca4df66d9be645f197f05"def protocol():    cfg = rpc_call("EXPERIMENTAL_protocol_config", {"finality": "final"})    keep = ["protocol_version", "epoch_length", "max_inflation_rate", "protocol_reward_rate", "protocol_treasury_account",            "num_blocks_per_year", "num_block_producer_seats", "min_gas_price", "max_gas_price", "gas_price_adjustment_rate"]    out = {k: cfg.get(k) for k in keep}    out["burnt_gas_reward"] = cfg["runtime_config"]["transaction_costs"].get("burnt_gas_reward")    out["storage_amount_per_byte"] = cfg["runtime_config"].get("storage_amount_per_byte")    block = rpc_call("block", {"finality": "final"})["header"]    out["head"] = {"height": block["height"], "timestamp_ns": block["timestamp_nanosec"], "total_supply": block["total_supply"], "gas_price": block["gas_price"]}    return outdef validators():    res = rpc_call("validators", [None])    rows = []    for v in res["current_validators"]:        fee = None        try:            fee = rpc_view(v["account_id"], "get_reward_fee_fraction")        except StopSource:            pass        rows.append({"account": v["account_id"], "stake": v["stake"], "blocks_expected": v.get("num_expected_blocks"),                     "blocks_produced": v.get("num_produced_blocks"), "fee": fee})        time.sleep(0.08)    return {"epoch_start_height": res["epoch_start_height"], "epoch_height": res["epoch_height"], "validators": rows}def wallets_now():    out = {}    for acct in WALLETS:        res = rpc_call("query", {"request_type": "view_account", "finality": "final", "account_id": acct})        out[acct] = {"amount": res["amount"], "locked": res["locked"], "height": res["block_height"]}    return outdef control():    out = {"intents_fee_pips": rpc_view("intents.near", "fee"), "intents_fee_collector": rpc_view("intents.near", "fee_collector"),           "buyback_confirmations": rpc_view("buybacks.multisignature.near", "get_num_confirmations"),           "buyback_request_nonce": rpc_view("buybacks.multisignature.near", "get_request_nonce")}    for dao in ("fefundsadmin.sputnik-dao.near", "1csfundsadmin.sputnik-dao.near"):        pol = rpc_view(dao, "get_policy")        roles = [{"name": r["name"], "members": r["kind"].get("Group") if isinstance(r["kind"], dict) else r["kind"]} for r in pol["roles"]]        out[dao] = {"roles": roles, "last_proposal_id": rpc_view(dao, "get_last_proposal_id")}    out["venear_supply_near"] = int(rpc_view("venear.dao", "ft_total_supply")) / 1e24    out["venear_accounts"] = rpc_view("venear.dao", "get_num_accounts")    cfg = rpc_view("venear.dao", "get_config")    out["venear_unlock_days"] = int(cfg["unlock_duration_ns"]) / 86400e9    return outdef governance():    """House of Stake proposal 29 (issuance ramp to 1.6%) and every later proposal id, read from vote.dao."""    props = rpc_view("vote.dao", "get_proposals", {"from_index": 24, "limit": 12})    keep = ("id", "title", "status", "votes", "total_votes", "voting_start_time_ns", "voting_duration_ns", "approval_threshold_bps")    return [{k: p.get(k) for k in keep} for p in props]def heights_by_date():    out = {}    for name in ("01_supply_daily", "01b_supply_recent"):        path = ROOT / "raw" / "bigquery" / f"{name}.csv.gz"        if path.exists():            with gzip.open(path, "rt") as fh:                out.update({r["block_date"]: int(float(r["last_height"])) for r in csv.DictReader(fh) if r.get("last_height")})    return outdef view_at(account, height):    for delta in (0, -1, -2, -3, -5, -8, -13):        try:            res = rpc_call("query", {"request_type": "view_account", "block_id": height + delta, "account_id": account}, url=ARCHIVAL, timeout=30, retries=1)            return int(res["amount"])        except StopSource as exc:            text = str(exc)            if "UNKNOWN_ACCOUNT" in text or "does not exist" in text:                return 0            continue    return Nonedef wallet_history(start="2026-01-05", budget_s=420):    rows = load("wallet_history", []) or []    have = {(r["date"], r["account"]) for r in rows}    heights = heights_by_date()    day = datetime.date.fromisoformat(start)    today = datetime.datetime.now(datetime.timezone.utc).date()    t0 = time.time()    while day < today:        key = day.isoformat()        if key in heights:            for acct in WALLETS:                if (key, acct) in have:                    continue                if time.time() - t0 > budget_s:                    return sorted(rows, key=lambda r: (r["date"], r["account"]))                amt = view_at(acct, heights[key])                if amt is not None:                    rows.append({"date": key, "account": acct, "near": amt / 1e24, "height": heights[key]})                time.sleep(0.2)        day += datetime.timedelta(days=7)    return sorted(rows, key=lambda r: (r["date"], r["account"]))def supply_crosscheck():    """Compare BigQuery's supply with the header served by archival RPC at five dates: two independent readings of one number."""    heights = heights_by_date()    dates = sorted(heights)    picks = [dates[int(len(dates) * q)] for q in (0.3, 0.55, 0.8, 0.95, 0.999)]    out = []    for d in picks:        hdr = rpc_call("block", {"block_id": heights[d]}, url=ARCHIVAL, timeout=40)["header"]        out.append({"date": d, "height": heights[d], "rpc_supply_near": int(hdr["total_supply"]) / 1e24})        time.sleep(0.5)    return outJOBS = {"chain_protocol": protocol, "chain_validators": validators, "chain_wallets": wallets_now, "chain_control": control, "chain_governance": governance,        "wallet_history": wallet_history, "chain_supply_crosscheck": supply_crosscheck}def on_alarm(*_):    raise StopSource("source took too long")def main(argv):    only = set(argv[1:])    status = load("_fetch_status", {}) or {}    bad = 0    signal.signal(signal.SIGALRM, on_alarm)    for name, fn in JOBS.items():        if only and name not in only:            continue        signal.alarm(540)        try:            rec = save(name, fn(), "NEAR public RPC")            status[name] = {"ok": True, "pulled_at": rec["pulled_at"], "sha256": rec["sha256"][:16], "source": "NEAR public RPC"}            print(f"ok   {name}: {rec['bytes']/1024:.0f} KB")        except Exception as exc:  # the last good snapshot stays and the page names the stale source            bad += 1            status[name] = {"ok": False, "failed_at": now_utc(), "error": str(exc)[:160], "source": "NEAR public RPC",                            "pulled_at": (status.get(name) or {}).get("pulled_at")}            print(f"FAIL {name}: {str(exc)[:160]}")        finally:            signal.alarm(0)    save("_fetch_status", status, "pipeline")    return badif __name__ == "__main__":    sys.exit(1 if main(sys.argv) else 0)