"""Cluster Lens fetcher: keyless, cached, polite GETs only. Sources (all public, no credentials): * POAScan archive index of challenge txs https://poascan.io/api/v1/txs?type=challenge (ids, authors, epochs, timestamps; no answer text) * minepoa.com explorer, one tx at a time https://minepoa.com/api/explorer/tx/ (answer text + seed). Answers are immutable, so each is fetched once and cached forever. A per-pass budget and spacing keep load low, and the pass stops early when the node says it is busy (503 + retry_after). * LabLeak forum introductions https://minepoa.com/lableak/api/posts?sort=new&community=introductions """ import json import os import time import urllib.error import urllib.request UA = "poa-cluster-lens/1.0 (+https://embot-lantern.codeberg.page/poa-field-guide/clusters/)" POASCAN_TXS = "https://poascan.io/api/v1/txs?type=challenge&limit=1000&offset={off}" EXPLORER_TX = "https://minepoa.com/api/explorer/tx/{txid}" FORUM_INTROS = "https://minepoa.com/lableak/api/posts?sort=new&community=introductions&limit=200" class _NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, *a, **k): return None _OPENER = urllib.request.build_opener(_NoRedirect) class Busy(Exception): def __init__(self, retry_after): super().__init__("busy") self.retry_after = retry_after def http_get_json(url, timeout=120): req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "application/json"}) try: with _OPENER.open(req, timeout=timeout) as r: return json.loads(r.read().decode("utf-8")) except urllib.error.HTTPError as e: # The node signals "busy" with 429/503, and sometimes with a 307 carrying the same # JSON body. Redirects are never followed; a redirect without that body is an error. if e.code in (429, 503, 307): body = {} try: body = json.loads(e.read().decode("utf-8", "replace")) except Exception: pass if e.code == 307 and "retry_after_seconds" not in body: raise ra = body.get("retry_after_seconds") if ra is None and e.headers is not None: ra = e.headers.get("Retry-After") try: ra = int(ra) except (TypeError, ValueError): ra = 60 raise Busy(ra) raise # Chain segments (genesis cut 2026-10-24): epoch numbers restart in each segment, so every index row carries the # segment it belongs to, assigned by timestamp. Segment 1's keyed routes were halted from well before 20:00Z on # Oct 24 until the cut (notice issued 21:01Z) and its newest indexed challenge tx is from 13:28Z, so 20:00Z is a # safe boundary. Add a row here at the next cut. SEGMENTS = [("poa-mainnet-1-segment1", 0), ("poa-mainnet-1-segment2", 1792872000)] def segment_of(ts): seg = SEGMENTS[0][0] for sid, start in SEGMENTS: if isinstance(ts, (int, float)) and ts >= start: seg = sid return seg def tag_segments(index): """Give every index row a `seg` (in place). Returns how many rows were tagged.""" n = 0 for v in index.values(): if isinstance(v, dict) and "seg" not in v: v["seg"] = segment_of(v.get("ts")) n += 1 return n def update_index(index, get=http_get_json, sleep=time.sleep, max_pages=30, pause=2.0): """index: dict txid -> {from, epoch, ts, seg}. Pages newest-first until a page adds nothing new.""" added = 0 for page in range(max_pages): rows = get(POASCAN_TXS.format(off=page * 1000)) new_here = 0 for x in rows: if x.get("type") != "challenge" or not x.get("id"): continue if x["id"] not in index: new_here += 1 index[x["id"]] = {"from": x.get("from_addr"), "epoch": x.get("epoch"), "ts": x.get("ts"), "seg": segment_of(x.get("ts"))} added += new_here if len(rows) < 1000 or (new_here == 0 and page > 0): break sleep(pause) return added def fill_bodies(index, bodies, budget=600, get=http_get_json, sleep=time.sleep, spacing=1.2, max_busy=3, deadline_s=300, now=time.time, max_errors=3): """Fetch answer text for up to `budget` uncached txids, newest epochs first. Returns (fetched, missing_after, stopped_reason).""" todo = [t for t in index if t not in bodies] todo.sort(key=lambda t: (-(index[t].get("epoch") if index[t].get("epoch") is not None else -1), -(index[t].get("ts") or 0))) fetched, busy, errors, start = 0, 0, 0, now() reason = "done" for txid in todo: if fetched >= budget: reason = "budget" break if now() - start > deadline_s: reason = "deadline" break try: d = get(EXPLORER_TX.format(txid=txid), timeout=60) except Busy as b: busy += 1 if busy >= max_busy: reason = "node busy" break sleep(min(max(b.retry_after, 5), 120)) continue except Exception: errors += 1 # light-client rule: a few consecutive failures end the pass if errors >= max_errors: reason = "errors" break sleep(spacing) continue errors = 0 body = d.get("body") or {} if d.get("type") == "challenge" and isinstance(body.get("answer"), str): bodies[txid] = {"answer": body["answer"], "seed": body.get("seed"), "from": d.get("from_addr"), "epoch": d.get("epoch"), "ts": d.get("ts"), "seg": (index.get(txid) or {}).get("seg") or segment_of(d.get("ts"))} fetched += 1 elif d.get("type") and d.get("type") != "challenge": bodies[txid] = None # not an answer; never refetch sleep(spacing) return fetched, len([t for t in index if t not in bodies]), reason def fetch_intros(get=http_get_json): rows = get(FORUM_INTROS) out = [] for p in rows: if str(p.get("deleted")) == "1" or (p.get("body") or "").strip() == "[deleted by author]": continue out.append({"id": int(p["id"]), "author": p.get("author"), "title": p.get("title") or "", "body": p.get("body") or "", "created_at": p.get("created_at")}) return out def load(path, default): try: with open(path) as f: return json.load(f) except (OSError, ValueError): return default def save(path, obj): tmp = path + ".tmp" with open(tmp, "w") as f: json.dump(obj, f, separators=(",", ":")) os.replace(tmp, path)