Something went wrong. Try again.
Something went wrong. Try again.
8.9 kB · 229 lines
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230"""Stages B-D (Tangled-API-free): fetch the actual records from PDSes and derive state.
B every PR author's sh.tangled.repo.pull records, straight from their PDSC every author's and repo owner's sh.tangled.repo.pull.status recordsD join them into pulls.jsonl in the same shape the original crawl produced, so analyze.py and build_html.py run unchanged
Derived state is the status of the newest status record pointing at a pull,which is exactly what Tangled's appview computes - we just compute it ourselves."""import json, os, queue, threading, time, urllib.parse, urllib.request, collections
SD = "/tmp/claude-1000/-home-jmm-Code-blogposts/7ba1f72e-807b-4749-bcd3-a065da793e6d/scratchpad"C = "https://constellation.microcosm.blue"UA = "tangled-stats-research/0.2 (permadeath.com; blog post stats)"
def get(url, tries=4, timeout=60): for i in range(tries): try: req = urllib.request.Request(url, headers={"User-Agent": UA}) with urllib.request.urlopen(req, timeout=timeout) as f: return json.load(f) except urllib.error.HTTPError as e: if e.code in (400, 404): return None if i == tries - 1: raise time.sleep(2 * (i + 1)) except Exception: if i == tries - 1: raise time.sleep(2 * (i + 1))
def par(items, fn, n=10, 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) if v is not None: with lock: res.append(v) except Exception as e: with lock: print(f"ERR {str(it)[:60]}: {str(e)[:60]}", flush=True) finally: with lock: c[0] += 1 if c[0] % 100 == 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
def cached(path, build): if os.path.exists(path): return json.load(open(path)) v = build() json.dump(v, open(path, "w")) return v
def pds_of(did): doc = get("https://" + did[8:] + "/.well-known/did.json") if did.startswith("did:web:") \ else get(f"https://plc.directory/{did}") if not doc: return (did, None) for s in doc.get("service", []): if s.get("id", "").endswith("atproto_pds"): return (did, s["serviceEndpoint"]) return (did, None)
def list_records(host, did, collection): out, cursor = [], None while True: u = (f"{host}/xrpc/com.atproto.repo.listRecords?repo={urllib.parse.quote(did)}" f"&collection={collection}&limit=100") if cursor: u += "&cursor=" + urllib.parse.quote(cursor) d = get(u) if not d: break out += d.get("records", []) cursor = d.get("cursor") if not cursor or not d.get("records"): break return out
def main(): # ---- who is involved linkrows = [json.loads(l) for l in open(f"{SD}/alt_links2.jsonl")] active = [r for r in linkrows if r["n"]] authors = {p["did"] for r in active for p in r["pulls"]} print(f"repos with pulls: {len(active)} pull authors: {len(authors)}", flush=True)
owners = cached(f"{SD}/alt_owners.json", lambda: dict(filter(None, par( [r["repoDid"] for r in active], lambda rd: (rd, (get(f"{C}/links?target={urllib.parse.quote(rd, safe='')}" f"&collection=sh.tangled.repo&path=.repoDid&limit=10") or {}).get("linking_records", [{}])[0].get("did")), n=8, label="owners")))) owner_dids = {v for v in owners.values() if v} print(f"repo owners: {len(owner_dids)}", flush=True)
everyone = sorted(authors | owner_dids) PM = cached(f"{SD}/pds_map.json", lambda: dict(par(everyone, pds_of, n=8, label="pds"))) missing = [d for d in everyone if d not in PM] if missing: PM.update(dict(par(missing, pds_of, n=8, label="pds+"))) json.dump(PM, open(f"{SD}/pds_map.json", "w")) print(f"PDS resolved: {sum(1 for v in PM.values() if v)}/{len(PM)}", flush=True)
# ---- B: pull records from each author's PDS def fetch_pulls(did): host = PM.get(did) if not host: return None return {"did": did, "records": list_records(host, did, "sh.tangled.repo.pull")}
if not os.path.exists(f"{SD}/alt_pull_records.jsonl"): rows = par(sorted(authors), fetch_pulls, n=10, label="pulls") with open(f"{SD}/alt_pull_records.jsonl", "w") as f: for r in rows: f.write(json.dumps(r) + "\n") pull_recs = {} for l in open(f"{SD}/alt_pull_records.jsonl"): r = json.loads(l) for rec in r["records"]: pull_recs[rec["uri"]] = rec print(f"pull records fetched: {len(pull_recs)}", flush=True)
# ---- C: status records from everyone who could have merged something def fetch_status(did): host = PM.get(did) if not host: return None return {"did": did, "records": list_records(host, did, "sh.tangled.repo.pull.status")}
if not os.path.exists(f"{SD}/alt_status_records.jsonl"): rows = par(everyone, fetch_status, n=10, label="status") with open(f"{SD}/alt_status_records.jsonl", "w") as f: for r in rows: f.write(json.dumps(r) + "\n") latest = {} nstat = 0 for l in open(f"{SD}/alt_status_records.jsonl"): r = json.loads(l) for rec in r["records"]: v = rec.get("value") or {} pull, st = v.get("pull"), v.get("status") if not pull or not st: continue nstat += 1 rkey = rec["uri"].rsplit("/", 1)[-1] # TID: monotonic, so max == newest cur = latest.get(pull) if not cur or rkey > cur[0]: latest[pull] = (rkey, st.rsplit(".", 1)[-1], v.get("createdAt")) print(f"status records: {nstat} covering {len(latest)} pulls", flush=True)
# ---- D: join into the legacy pulls.jsonl shape by_repo = collections.defaultdict(list) for row in active: for p in row["pulls"]: uri = f"at://{p['did']}/sh.tangled.repo.pull/{p['rkey']}" rec = pull_recs.get(uri) if not rec: continue v = rec.get("value") or {} st = latest.get(uri) rounds = v.get("rounds") or [] tgt = v.get("target") or {} by_repo[row["repoDid"]].append({ "uri": uri, "state": st[1] if st else "open", "createdAt": v.get("createdAt"), "stateUpdatedAt": st[2] if st else None, "title": (v.get("title") or "")[:200], "comments": 0, "rounds": len(rounds) if isinstance(rounds, list) else 0, "targetBranch": tgt.get("branch"), "targetRepo": tgt.get("repoDid") or tgt.get("repo"), "bodyLen": len(v.get("body") or ""), }) with open(f"{SD}/pulls.jsonl", "w") as f: for row in linkrows: items = by_repo.get(row["repoDid"], []) f.write(json.dumps({"repoDid": row["repoDid"], "n": len(items), "pulls": items}) + "\n") tot = sum(len(v) for v in by_repo.values()) states = collections.Counter(p["state"] for v in by_repo.values() for p in v) print(f"wrote pulls.jsonl: {tot} pulls {dict(states)}", flush=True)
# merged_pulls.json for the existing patch-blob stage mp = [] for repo, items in by_repo.items(): for p in items: if p["state"] != "merged": continue rec = pull_recs.get(p["uri"]) rounds = ((rec or {}).get("value") or {}).get("rounds") or [] pb = (rounds[-1].get("patchBlob") if rounds else None) or {} mp.append({"uri": p["uri"], "createdAt": p["createdAt"], "stateUpdatedAt": p["stateUpdatedAt"], "repoDid": repo, "rounds": len(rounds), "cid": (pb.get("ref") or {}).get("$link"), "size": pb.get("size")}) json.dump(mp, open(f"{SD}/merged_pulls.json", "w")) print(f"merged pulls with a patch: {sum(1 for x in mp if x['cid'])}/{len(mp)}", flush=True) print("DONE", flush=True)
if __name__ == "__main__": main()