"""AN4.4 (DC Analyst 4, 7 Oct 2026): funder roots for Uniswap v3 creators, crew.db method (first funder by archive balance gallop +
debug_traceBlockByNumber callTracer, send-to-many contracts resolve to the caller), walked up hop by hop until the funder is a contract,
an exchange-like wallet (nonce >= 2000, the crew.db exchange floor) or unknown. root = the top EOA reached. Reuses PP ffund.py logic.
Coins: A1 universe v3 rows (venue v3) + every pool traded by the FW mc5 / SW live books. Creator + creation block from harness gold idx.json.
Known funders are read from PP crew.db / ffund_snap.db first (read only). Writes ONLY /opt/tw/an4/crew_v3.db (owner: DC Analyst 4).
Paid dRPC key read from /opt/common/drpc_paid.json, never printed."""
import json, sys, time, sqlite3, urllib.request, threading, csv, gzip
from concurrent.futures import ThreadPoolExecutor
U = json.load(open("/opt/common/drpc_paid.json"))["robinhood"]
U = U if U.startswith("http") else "https://lb.drpc.org/ogrpc?network=robinhood&dkey=" + U
TH = int(sys.argv[1]) if len(sys.argv) > 1 else 8; MAXHOP = 5; EXCH = 2000
ST = {"calls": 0, "req": 0, "traced": 0, "reused": 0, "err": 0}; LK = threading.Lock()
def post(body):
    err = None
    for i in range(6):
        try:
            with LK: ST["req"] += 1; ST["calls"] += len(body) if isinstance(body, list) else 1
            return json.loads(urllib.request.urlopen(urllib.request.Request(U, json.dumps(body).encode(),
                {"Content-Type": "application/json", "User-Agent": "Mozilla/5.0"}), timeout=90).read())
        except Exception as e: err = e; time.sleep(1 + 2 * i)
    raise RuntimeError(err)
def rpc(m, p):
    r = post({"jsonrpc": "2.0", "id": 1, "method": m, "params": p})
    if "error" in r: raise RuntimeError(str(r["error"])[:160])
    return r["result"]
def rpcb(calls):
    r = post([{"jsonrpc": "2.0", "id": k, "method": m, "params": p} for k, (m, p) in enumerate(calls)])
    r = {x["id"]: x for x in r}
    if any("error" in r[k] for k in r): raise RuntimeError(str([r[k]["error"] for k in r if "error" in r[k]][0])[:160])
    return [r[k]["result"] for k in range(len(calls))]
def first_funding(a, before):
    """earliest block with balance > 0 before `before` (ffund.py). Returns (fb or None, is_contract, nonce_latest)."""
    hi = before - 1
    pts = [hi] + [max(0, hi - 256 * 8 ** k) for k in range(8)]
    res = rpcb([("eth_getBalance", [a, hex(b)]) for b in pts] + [("eth_getCode", [a, "latest"]), ("eth_getTransactionCount", [a, "latest"])])
    isc = bool(res[-2] and res[-2] != "0x"); nonce = int(res[-1], 16)
    if isc or nonce >= EXCH: return None, isc, nonce
    bals = [int(x, 16) for x in res[:len(pts)]]
    if bals[0] == 0: return None, False, nonce
    lo = None
    for b, v in zip(pts[1:], bals[1:]):
        if v == 0: lo = b; break
        hi = b
    if lo is None: return (0 if hi == 0 else hi), False, nonce
    while hi - lo > 1:
        step = max(1, (hi - lo) // 8); mids = sorted({lo + step * k for k in range(1, 8) if lo < lo + step * k < hi})
        if not mids: break
        bv = [int(x, 16) for x in rpcb([("eth_getBalance", [a, hex(m)]) for m in mids])]
        nlo, nhi = lo, hi
        for m, v in zip(mids, bv):
            if v > 0: nhi = m; break
            nlo = m
        lo, hi = nlo, nhi
    return hi, False, nonce
def sender_of(a, b):
    tr = rpc("debug_traceBlockByNumber", [hex(b), {"tracer": "callTracer"}]) or []
    for t in tr:
        res = t.get("result") or {}; top = (res.get("from") or "").lower(); hit = []
        def walk(c, parent):
            to = (c.get("to") or "").lower()
            if to == a and int(c.get("value") or "0x0", 16) > 0 and c.get("type", "CALL") in ("CALL", "CALLCODE"): hit.append(parent)
            for x in c.get("calls") or []: walk(x, to)
        walk(res, None)
        if hit: return top, (hit[0] if hit[0] and hit[0] != top else None), t.get("txHash")
    return None, None, None
DB = sqlite3.connect("/opt/tw/an4/crew_v3.db", timeout=120, check_same_thread=False)
DB.executescript("CREATE TABLE IF NOT EXISTS wallet(addr TEXT PRIMARY KEY, kind TEXT, funder TEXT, via TEXT, fund_b INT, fund_tx TEXT, nonce INT, src TEXT);"
                 "CREATE TABLE IF NOT EXISTS launch(pool TEXT PRIMARY KEY, token TEXT, creator TEXT, root TEXT, hops INT, stop TEXT, b INT, src TEXT);"
                 "CREATE INDEX IF NOT EXISTS launch_root ON launch(root); CREATE INDEX IF NOT EXISTS wallet_funder ON wallet(funder);")
WL = threading.Lock(); W = {a: (k, f, fb, n) for a, k, f, fb, n in DB.execute("SELECT addr, kind, funder, fund_b, nonce FROM wallet")}
PPC = sqlite3.connect("file:/opt/pricepredict-data/crew.db?mode=ro", uri=True, check_same_thread=False)
PPF = sqlite3.connect("file:/opt/pricepredict-data/ffund_snap.db?mode=ro", uri=True, check_same_thread=False)
PL = threading.Lock()
def pp_lookup(a):
    with PL:
        r = PPC.execute("SELECT kind, funder, fund_b, nonce FROM wallet WHERE addr=?", (a,)).fetchone()
        if not r: r = PPF.execute("SELECT kind, funder, fund_b, NULL FROM ff WHERE addr=?", (a,)).fetchone()
    return r
def save(a, kind, f, via, fb, tx, n, src):
    with WL:
        W[a] = (kind, f, fb, n)
        DB.execute("INSERT OR REPLACE INTO wallet VALUES(?,?,?,?,?,?,?,?)", (a, kind, f, via, fb, tx, n, src)); DB.commit()
def funder_of(a, before):
    if a in W: ST["reused"] += 1; return W[a]
    r = pp_lookup(a)
    if r and (r[0] in ("exchange", "contract") or r[1]):
        save(a, r[0], (r[1] or "").lower() or None, None, r[2], None, r[3], "pp"); ST["reused"] += 1; return W[a]
    fb, isc, n = first_funding(a, before)
    if isc: save(a, "contract", None, None, None, None, n, "trace"); return W[a]
    if n >= EXCH: save(a, "exchange", None, None, None, None, n, "trace"); return W[a]
    f, via, tx = sender_of(a, fb) if fb else (None, None, None)
    ST["traced"] += 1
    save(a, "eoa", f if f and f != a else None, via, fb, tx, n, "trace"); return W[a]
def root_of(creator, b):
    cur, before, chain = creator, b, []
    for hop in range(MAXHOP + 1):
        kind, f, fb, n = funder_of(cur, before)
        if kind in ("contract", "exchange"):
            return (chain[-1] if chain else cur), hop, kind
        chain.append(cur)
        if not f: return cur, hop, "nofunder"
        if hop == MAXHOP: return cur, hop, "maxhop"
        cur, before = f, (fb or before)
    return chain[-1], MAXHOP, "maxhop"
IDX = json.load(open("/opt/sw/harness/gold/idx.json"))
pools = {}
for r in csv.DictReader(gzip.open("/root/dca1/dc-a1-1-universe.csv.gz", "rt")):
    if r["venue"] == "v3": pools[r["pool"].lower()] = "a1v3"
for fn in ("/root/fwa3/lab/out/sw_accept/live_book_trades.csv", "/root/fwa3/lab/out/sw/fw_mc5_trades.csv"):
    for r in csv.DictReader(open(fn)): pools.setdefault(r["pool"].lower(), "book")
done = {p for (p,) in DB.execute("SELECT pool FROM launch")}
todo = [(p, s) for p, s in pools.items() if p not in done and IDX.get(p, {}).get("creator")][:int(sys.argv[2]) if len(sys.argv) > 2 else None]
print("pools", len(pools), "with creator", sum(1 for p in pools if IDX.get(p, {}).get("creator")), "todo", len(todo), flush=True)
def one(item):
    p, src = item; m = IDX[p]
    try:
        cr = m["creator"].lower(); b = int(m.get("blk") or m.get("first_b") or 0)
        root, hops, stop = root_of(cr, b)
        with WL:
            DB.execute("INSERT OR REPLACE INTO launch VALUES(?,?,?,?,?,?,?,?)", (p, m.get("tok"), cr, root, hops, stop, b, src)); DB.commit()
    except Exception as e:
        ST["err"] += 1; print("fail", p, repr(e)[:150], flush=True)
t0 = time.time()
def mon():
    while True:
        time.sleep(60); print(time.strftime("%H:%M:%S"), ST, "%.0f s" % (time.time() - t0), flush=True)
threading.Thread(target=mon, daemon=True).start()
with ThreadPoolExecutor(TH) as ex: list(ex.map(one, todo))
print("DONE", ST, "%.0f s" % (time.time() - t0), flush=True)