"""Shared helpers for hub data jobs. Stdlib only."""

from __future__ import annotations

import json, os, time, urllib.request, urllib.error, datetime, pathlib, concurrent.futures as cf

ROOT = pathlib.Path(__file__).resolve().parent.parent

DATA = ROOT / "data"

UA = "Mozilla/5.0 (compatible; BlockchainLabHub/1.0; +https://blockchains.github.io/)"

def now_iso() -> str:

return datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")

def http(url: str, data: bytes | None = None, headers: dict | None = None, timeout: float = 25, retries: int = 2) -> bytes:

h = {"User-Agent": UA, "Accept": "*/*"}

h.update(headers or {})

last: Exception | None = None

for i in range(retries + 1):

try:

req = urllib.request.Request(url, data=data, headers=h)

with urllib.request.urlopen(req, timeout=timeout) as r:

return r.read()

except urllib.error.HTTPError as e:

last = e

if e.code in (400, 401, 402, 403, 404):

break

except Exception as e: # noqa: BLE001

last = e

time.sleep(1.5 * (i + 1))

raise RuntimeError(f"GET {url} failed: {last}")

def get_json(url: str, **kw):

return json.loads(http(url, **kw))

def rpc(url: str, method: str, params: list | None = None, timeout: float = 12):

body = json.dumps({"jsonrpc": "2.0", "id": 1, "method": method, "params": params or []}).encode()

r = json.loads(http(url, data=body, headers={"Content-Type": "application/json"}, timeout=timeout, retries=0))

if "error" in r:

raise RuntimeError(str(r["error"]))

return r["result"]

def write(name: str, payload: dict, *, source: str, source_url: str, description: str) -> pathlib.Path:

out = DATA / name

out.parent.mkdir(parents=True, exist_ok=True)

env = {

"service": name.removesuffix(".json"),

"description": description,

"generated_at": now_iso(),

"source": source,

"source_url": source_url,

"built_by": "Blockchain Lab — https://blockchainlab.com",

}

env.update(payload)

out.write_text(json.dumps(env, indent=1, ensure_ascii=False) + "\n")

return out

def read(name: str, default=None):

p = DATA / name

return json.loads(p.read_text()) if p.exists() else default

def pmap(fn, items, workers: int = 16):

with cf.ThreadPoolExecutor(workers) as ex:

return list(ex.map(fn, items))

# ---------- Grok (xAI) — server-side only; key comes from the environment / Actions secret ----------

XAI_MODEL = os.environ.get("XAI_MODEL", "grok-4.3")

def grok(system: str, user: str, *, json_mode: bool = False, max_tokens: int = 1800, temperature: float = 0.3) -> tuple[str, dict]:

key = os.environ.get("XAI_API_KEY")

if not key:

raise RuntimeError("XAI_API_KEY not set")

body = {

"model": XAI_MODEL,

"messages": [{"role": "system", "content": system}, {"role": "user", "content": user}],

"max_tokens": max_tokens,

"temperature": temperature,

}

if json_mode:

body["response_format"] = {"type": "json_object"}

last = None

for i in range(3):

try:

r = json.loads(http("https://api.x.ai/v1/chat/completions", data=json.dumps(body).encode(),

headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},

timeout=180, retries=0))

text = r["choices"][0]["message"]["content"]

meta = {"model": r.get("model", XAI_MODEL), "tokens": r.get("usage", {}).get("total_tokens")}

if json_mode:

json.loads(text) # validate

return text, meta

except Exception as e: # noqa: BLE001

last = e

time.sleep(4 * (i + 1))

raise RuntimeError(f"grok failed: {last}")

AI_DISCLAIMER = ("AI-generated by Grok from the public data shown. It can be wrong; verify before acting. "

"Not financial, legal or security advice.")