#!/usr/bin/env python3"""Swap-level pull from the NEAR Intents Explorer API (1Click swaps). One request per 5.3 s, the documented limit.Needs a free read-only JWT from https://partners.near-intents.org (key type "Explorer"), passed as env INTENTS_JWTor as a path in INTENTS_JWT_FILE. Without it the pipeline uses the committed aggregates in raw/explorer/.Sampling is deterministic, so anyone re-running this draws the same swaps:  panel   one full UTC day per ISO week from 2025-12-01, the day chosen by sha256(week start). This is what the analysis uses.  whales  optional: every swap of at least $100k, month by month (the API times out on one long window)  recent  optional: every swap in the eight full weeks before --untilRows go to raw_private/explorer/ (ignored by git: 100s of MB). Addresses are stored only as a truncated sha256.analyze_explorer.py turns them into the committed aggregates."""import argparse, calendar, csv, datetime, gzip, hashlib, json, os, pathlib, re, sys, time, urllib.error, urllib.parse, urllib.requestsys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))from common import MAX_BYTES, OPENER, UA, StopSource  # noqa: E402ROOT = pathlib.Path(__file__).resolve().parent.parentOUT = ROOT / "raw_private" / "explorer"API = "https://explorer.near-intents.org/api/v0/transactions"FIELDS = ["ts", "status", "origin", "dest", "referral", "in_usd", "out_usd", "fees", "sender", "recipient", "kind"]GAP_S = 5.3def token():    tok = os.environ.get("INTENTS_JWT")    path = os.environ.get("INTENTS_JWT_FILE")    if not tok and path:        tok = pathlib.Path(path).read_text().strip()    if not tok:        raise SystemExit("INTENTS_JWT or INTENTS_JWT_FILE is required for a live Explorer pull")    return tokdef clean(tag, limit=60):    """Partner tags are free text from strangers: keep them plain before they reach any CSV or page."""    return re.sub(r"[^A-Za-z0-9 ._()@:-]", "_", tag or "")[:limit]def short_hash(addr):    return hashlib.sha256((addr or "").strip().lower().encode()).hexdigest()[:16] if addr else ""def compact(x):    fees = ";".join(f"{clean(str(f.get('recipient', '')), 70)}:{float(f.get('fee', 0) or 0):g}" for f in (x.get("appFees") or []))    senders = x.get("senders") or []    return {"ts": int(x.get("createdAtTimestamp") or 0), "status": x.get("status"), "origin": clean(x.get("originAsset"), 90),            "dest": clean(x.get("destinationAsset"), 90), "referral": clean(x.get("referral")),            "in_usd": x.get("amountInUsd") or "0", "out_usd": x.get("amountOutUsd") or "0", "fees": fees[:160],            "sender": short_hash(senders[0] if senders else ""), "recipient": short_hash(x.get("recipient")),            "kind": x.get("depositType") or ""}class Client:    def __init__(self):        self.tok = token()        self.last = 0.0    def get(self, query):        wait = self.last + GAP_S - time.time()        if wait > 0:            time.sleep(wait)        for attempt in range(6):            self.last = time.time()            req = urllib.request.Request(f"{API}?{query}", headers={"Authorization": f"Bearer {self.tok}", "User-Agent": UA})            try:                with OPENER.open(req, timeout=60) as resp:                    return json.loads(resp.read(MAX_BYTES + 1))            except urllib.error.HTTPError as exc:                if exc.code == 429:                    time.sleep(12 + attempt * 6)                    continue                if exc.code >= 500:                    time.sleep(10 + attempt * 5)                    continue                raise SystemExit(f"Explorer HTTP {exc.code}")            except (urllib.error.URLError, TimeoutError, OSError, ValueError, StopSource):                time.sleep(10 + attempt * 10)        raise SystemExit("Explorer: retries exhausted")def crawl(client, path, start, end, extra=""):    """Walks one window newest to oldest with the API's cursor and writes one gzip CSV. Skips finished files."""    done = path.with_suffix(path.suffix + ".done")    if done.exists():        return 0    tmp = path.with_suffix(path.suffix + ".tmp")    query = f"numberOfTransactions=1000&startTimestampUnix={start}&endTimestampUnix={end}&statuses=SUCCESS,REFUNDED,FAILED{extra}"    cursor, rows, pages = None, 0, 0    with gzip.open(tmp, "wt", newline="") as fh:        wr = csv.DictWriter(fh, fieldnames=FIELDS)        wr.writeheader()        while True:            q = query            if cursor:                q += f"&lastDepositAddress={urllib.parse.quote(cursor[0], safe='')}"                if cursor[1]:                    q += f"&lastDepositMemo={urllib.parse.quote(cursor[1], safe='')}"            batch = client.get(q)            for item in batch:                wr.writerow(compact(item))            rows += len(batch)            pages += 1            if len(batch) < 1000:                break            cursor = (batch[-1]["depositAddress"], batch[-1].get("depositMemo"))    tmp.rename(path)    done.write_text(json.dumps({"start": start, "end": end, "rows": rows, "pages": pages}))    return rowsdef utc(day):    return calendar.timegm(day.timetuple())def panel_days(first=datetime.date(2025, 1, 6), last=None):    last = last or datetime.datetime.now(datetime.timezone.utc).date()    day = first    while day + datetime.timedelta(days=7) <= last:        offset = hashlib.sha256(day.isoformat().encode()).digest()[0] % 7        yield day + datetime.timedelta(days=offset)        day += datetime.timedelta(days=7)def main():    ap = argparse.ArgumentParser()    ap.add_argument("--stage", choices=["panel", "whales", "recent", "all"], default="panel")    ap.add_argument("--until", default="2026-10-26", help="end (exclusive) of the eight recent weeks, a Monday")    ap.add_argument("--from-day", default="2025-12-01", help="panel stage: earliest sample day to fetch now")    ap.add_argument("--to-day", default="2099-01-01", help="panel stage: latest sample day to fetch now")    args = ap.parse_args()    client = Client()    OUT.mkdir(parents=True, exist_ok=True)    (OUT / "panel").mkdir(exist_ok=True)    (OUT / "recent").mkdir(exist_ok=True)    if args.stage in ("panel", "all"):        for day in sorted(panel_days(), reverse=True):            if not args.from_day <= day.isoformat() <= args.to_day:                continue            n = crawl(client, OUT / "panel" / f"{day.isoformat()}.csv.gz", utc(day), utc(day) + 86400 - 1)            print("panel", day, n, flush=True)    if args.stage in ("recent", "all"):        end = datetime.date.fromisoformat(args.until)        for i in range(56):            day = end - datetime.timedelta(days=i + 1)            n = crawl(client, OUT / "recent" / f"{day.isoformat()}.csv.gz", utc(day), utc(day) + 86400 - 1)            print("recent", day, n, flush=True)    if args.stage in ("whales", "all"):        (OUT / "whales").mkdir(exist_ok=True)        y, mon = 2025, 1        now = datetime.datetime.now(datetime.timezone.utc).date()        while (y, mon) <= (now.year, now.month):            nxt = (y + (mon == 12), mon % 12 + 1)            try:                n = crawl(client, OUT / "whales" / f"{y}-{mon:02d}.csv.gz", utc(datetime.date(y, mon, 1)), utc(datetime.date(*nxt, 1)) - 1, "&minUsdPrice=100000")                print("whales", y, mon, n, flush=True)            except SystemExit as exc:                print("whales failed", y, mon, exc, flush=True)            y, mon = nxtif __name__ == "__main__":    sys.exit(main())