Something went wrong. Try again.
Something went wrong. Try again.
4.1 kB · 95 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596import json, threading, queue, time, gzip, re, urllib.parse, urllib.request, osA="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 outlists=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: passlock=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 Nonepar(mp,fetch,n=10,label="blob")out.flush(); out.close(); print("DONE",flush=True)