import json, threading, queue, time, gzip, re, urllib.parse, urllib.request, os
A="https://api.tangled.org/xrpc"; UA="tangled-stats-research/0.1 (permadeath.com)"
SD="/tmp/claude-1000/-home-jmm-Code-blogposts/7ba1f72e-807b-4749-bcd3-a065da793e6d/scratchpad"
def _get(url,tries=4,raw=False,timeout=90):
    for i in range(tries):
        try:
            r=urllib.request.Request(url,headers={"User-Agent":UA})
            with urllib.request.urlopen(r,timeout=timeout) as f:
                return f.read() if raw else json.load(f)
        except Exception:
            if i==tries-1: raise
            time.sleep(1.5*(i+1))
def par(items,fn,n=12,label=""):
    q=queue.Queue(); [q.put(i) for i in items]; lock=threading.Lock(); res=[]; c=[0]
    def w():
        while True:
            try: it=q.get_nowait()
            except queue.Empty: return
            try:
                v=fn(it)
                with lock:
                    if v is not None: res.append(v)
            except Exception as e:
                with lock: print("ERR",str(it)[:80],e,flush=True)
            finally:
                with lock:
                    c[0]+=1
                    if c[0]%250==0: print(f"  {label} {c[0]}/{len(items)}",flush=True)
                q.task_done()
    ts=[threading.Thread(target=w,daemon=True) for _ in range(n)]
    [t.start() for t in ts]; [t.join() for t in ts]; return res

active=[json.loads(l)["repoDid"] for l in open(f"{SD}/pulls.jsonl") if json.loads(l)["n"]]
print("active repos",len(active),flush=True)
def merged(did):
    out=[];cur=None
    while True:
        u=f"{A}/sh.tangled.repo.listPulls?subject={did}&status=merged&limit=1000"+(f"&cursor={urllib.parse.quote(cur)}" if cur else "")
        d=_get(u); 
        for it in d.get("items",[]):
            v=it.get("value") or {}; rs=v.get("rounds") or []
            pb=(rs[-1].get("patchBlob") if rs else None) or {}
            out.append({"uri":it["uri"],"createdAt":v.get("createdAt"),
                        "stateUpdatedAt":it.get("stateUpdatedAt"),"repoDid":did,
                        "rounds":len(rs),
                        "cid":((pb.get("ref") or {}).get("$link")),"size":pb.get("size")})
        cur=d.get("cursor")
        if not cur or not d.get("items"): break
    return out
lists=par(active,merged,label="merged")
mp=[p for l in lists for p in l]
json.dump(mp,open(f"{SD}/merged_pulls.json","w"))
print("merged pulls",len(mp),flush=True)

dids=sorted({p["uri"].split("/")[2] for p in mp})
print("authors",len(dids),flush=True)
def pds(did):
    if did.startswith("did:web:"):
        doc=_get("https://"+did[8:]+"/.well-known/did.json")
    else:
        doc=_get(f"https://plc.directory/{did}")
    for s in doc.get("service",[]):
        if s.get("id","").endswith("atproto_pds"): return (did,s["serviceEndpoint"])
    return (did,None)
PM=dict(par(dids,pds,n=8,label="pds"))
json.dump(PM,open(f"{SD}/pds_map.json","w"))

FROM=re.compile(rb"(?m)^From [0-9a-f]{40} ")
DATE=re.compile(rb"(?m)^Date: (.+)$")
AUTH=re.compile(rb"(?m)^From: (.+)$")
out=open(f"{SD}/commits.jsonl","a")
seen=set()
if os.path.exists(f"{SD}/commits.jsonl"):
    for l in open(f"{SD}/commits.jsonl"):
        try: seen.add(json.loads(l)["uri"])
        except: pass
lock=threading.Lock()
def fetch(p):
    if p["uri"] in seen or not p["cid"]: return None
    did=p["uri"].split("/")[2]; host=PM.get(did)
    if not host: return None
    raw=_get(f"{host}/xrpc/com.atproto.sync.getBlob?did={did}&cid={p['cid']}",raw=True)
    try: txt=gzip.decompress(raw)
    except Exception: txt=raw
    n=len(FROM.findall(txt))
    dates=[d.decode("utf8","replace") for d in DATE.findall(txt)][:n]
    auths=[a.decode("utf8","replace") for a in AUTH.findall(txt)][:n]
    rec={"uri":p["uri"],"repoDid":p["repoDid"],"author":did,"commits":n,
         "createdAt":p["createdAt"],"stateUpdatedAt":p["stateUpdatedAt"],
         "rounds":p["rounds"],"patchBytes":len(raw),
         "commitDates":dates,"commitAuthors":auths}
    with lock: out.write(json.dumps(rec)+"\n")
    return None
par(mp,fetch,n=10,label="blob")
out.flush(); out.close(); print("DONE",flush=True)