Something went wrong. Try again.
Something went wrong. Try again.
8.7 kB · 235 lines
Python
at typo-stack
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236"""Transitive closure over knots and users, so coverage isn't limited to knot1.
Repo owners were discovered by asking constellation who owns each repo DID on themain knot. That has two holes: it only ever looks at one knot, and constellation'sindex is firehose-derived (no backfill yet), so it misses repos whose recordspredate its indexing or have since been deleted.
This closes both. A person's own sh.tangled.repo records name the knot each repolives on AND give the repoDid -> owner mapping directly, with no index in themiddle. So: read every known person's repo records, collect the knots, list everyrepo on every knot, resolve owners of the new ones, and repeat until no new peopleturn up. Issue authors, stargazers and collaborators are pulled in too, to catchpeople who never created a repo of their own."""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.4 (permadeath.com; blog post stats)"MAX_ROUNDS = 6
def get(url, tries=2, timeout=20): 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: return None time.sleep(2 * (i + 1)) except Exception: if i == tries - 1: return None time.sleep(2 * (i + 1)) return None
def par(items, fn, n=14, label="", every=500): 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: pass finally: with lock: c[0] += 1 if c[0] % every == 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 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 knot_repos(knot): """Every repo DID a knot serves.""" out, cursor = [], None for _ in range(200): u = f"https://{knot}/xrpc/sh.tangled.sync.listRepos?limit=1000" if cursor: u += "&cursor=" + urllib.parse.quote(str(cursor)) d = get(u, timeout=45) if not d: break rs = d.get("repos") or [] out += [r["repo"] for r in rs if r.get("repo")] cursor = d.get("cursor") if not cursor or not rs: break return out
def jload(p, default): return json.load(open(p)) if os.path.exists(p) else default
def main(): PM = jload(f"{SD}/pds_map.json", {}) people = set(PM) for f in ("alt_pull_records.jsonl", "alt_status_records.jsonl"): if os.path.exists(f"{SD}/{f}"): for l in open(f"{SD}/{f}"): try: people.add(json.loads(l)["did"]) except Exception: pass owners = jload(f"{SD}/closure_owners.json", {}) # repoDid -> owner did knots = set(jload(f"{SD}/closure_knots.json", ["knot1.tangled.sh"])) repo_seen = set(jload(f"{SD}/closure_repos.json", [])) have_repo_recs = set(jload(f"{SD}/closure_fetched.json", []))
for rnd in range(1, MAX_ROUNDS + 1): print(f"\n===== round {rnd}: {len(people)} people, {len(knots)} knots, " f"{len(repo_seen)} repos, {len(owners)} owner links", flush=True)
# 1. resolve PDSes for anyone new missing = [d for d in people if d not in PM] if missing: print(f" resolving {len(missing)} PDSes", flush=True) PM.update(dict(par(missing, pds_of, n=10, label="pds"))) json.dump(PM, open(f"{SD}/pds_map.json", "w"))
# 2. read each new person's own repo records -> owner map + knot names todo = [d for d in people if d not in have_repo_recs and PM.get(d)] print(f" reading sh.tangled.repo for {len(todo)} people", flush=True) newknots = set()
def grab(did): recs = list_records(PM[did], did, "sh.tangled.repo") out = [] for r in recs: v = r.get("value") or {} rd, kn = v.get("repoDid"), v.get("knot") if rd: out.append((rd, did)) if kn: newknots.add(kn.strip().lower()) return (did, out)
for did, pairs in par(todo, grab, n=14, label="repos"): have_repo_recs.add(did) for rd, o in pairs: owners[rd] = o repo_seen.add(rd) knots |= {k for k in newknots if k and "." in k} print(f" knots known: {len(knots)}", flush=True)
# 3. list every repo on every knot def one_knot(k): return (k, knot_repos(k))
knotres = par(sorted(knots), one_knot, n=6, label="knots", every=5) for k, rs in knotres: if rs: repo_seen |= set(rs) print(f" repos known: {len(repo_seen)}", flush=True)
# 4. owners for repos we still can't attribute, plus other people who # touched them (issue authors, stargazers, collaborators) unknown = [r for r in repo_seen if r not in owners] print(f" looking up {len(unknown)} unattributed repos", flush=True) found = {}
def lookup(rd): hits = set() t = urllib.parse.quote(rd, safe='') for coll, path in (("sh.tangled.repo", ".repoDid"), ("sh.tangled.repo.issue", ".repo"), ("sh.tangled.repo.collaborator", ".repo"), ("sh.tangled.feed.star", ".subject.did")): d = get(f"{C}/links?target={t}&collection={coll}&path={path}&limit=100") for r in (d or {}).get("linking_records", []): hits.add(r["did"]) if coll == "sh.tangled.repo": found[rd] = r["did"] return hits
newpeople = set() for hits in par(unknown, lookup, n=10, label="attrib"): newpeople |= hits owners.update(found)
# pull authors we already know about count as people too for l in open(f"{SD}/alt_pull_records.jsonl"): try: newpeople.add(json.loads(l)["did"]) except Exception: pass
before = len(people) people |= newpeople json.dump(owners, open(f"{SD}/closure_owners.json", "w")) json.dump(sorted(knots), open(f"{SD}/closure_knots.json", "w")) json.dump(sorted(repo_seen), open(f"{SD}/closure_repos.json", "w")) json.dump(sorted(have_repo_recs), open(f"{SD}/closure_fetched.json", "w")) print(f" people {before} -> {len(people)} (+{len(people)-before})", flush=True) if len(people) == before: print(" closed - no new people", flush=True) break
json.dump(sorted(people), open(f"{SD}/closure_people.json", "w")) print(f"\nDONE people={len(people)} knots={len(knots)} repos={len(repo_seen)}", flush=True) print("knots:", ", ".join(sorted(knots)), flush=True)
if __name__ == "__main__": main()