Something went wrong. Try again.
Something went wrong. Try again.
5.8 kB · 170 lines
Python
at fixes-stack
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171"""Discovery expansion over the social graph, read from PDSes rather than an index.
A PR author is only counted if the crawl knows the person. Repo ownership plusConstellation backlinks missed at least one real author (juliet.tngl.sh: aself-hosted PDS, one repo, one PR, and no pull backlink in Constellation's index).
Follows, knot memberships and spindle memberships are all records a person keepsin their OWN repository, so reading them needs no index and has no coverage gap.Everyone those records point at is a Tangled user. Iterate to closure, then thepull crawl can be re-run over the enlarged set."""import json, os, queue, threading, time, urllib.parse, urllib.request
SD = "/tmp/claude-1000/-home-jmm-Code-blogposts/7ba1f72e-807b-4749-bcd3-a065da793e6d/scratchpad"UA = "tangled-stats-research/0.5 (permadeath.com; blog post stats)"COLLECTIONS = ["sh.tangled.graph.follow", "sh.tangled.knot.member", "sh.tangled.spindle.member", "sh.tangled.repo.collaborator"]MAX_ROUNDS = 5
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) except Exception: if i == tries - 1: return None time.sleep(2) return None
def par(items, fn, n=16, 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 for _ in range(500): 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
DID_KEYS = ("subject", "member", "did", "collaborator")
def dids_in(value): """Any DID this record points at.""" found = set()
def walk(v): if isinstance(v, str): if v.startswith("did:"): found.add(v) elif v.startswith("at://did:"): found.add(v[5:].split("/")[0]) elif isinstance(v, dict): for x in v.values(): walk(x) elif isinstance(v, list): for x in v: walk(x)
walk(value) return found
def main(): PM = json.load(open(f"{SD}/pds_map.json")) people = set(json.load(open(f"{SD}/closure_people.json"))) if os.path.exists(f"{SD}/closure_people.json") else set(PM) people |= set(PM) scanned = set(json.load(open(f"{SD}/social_scanned.json"))) if os.path.exists(f"{SD}/social_scanned.json") else set() print(f"start: {len(people)} people, {len(scanned)} already scanned", flush=True)
for rnd in range(1, MAX_ROUNDS + 1): todo = [d for d in people if d not in scanned and PM.get(d)] print(f"\n=== round {rnd}: scanning {len(todo)} people for social records", flush=True) if not todo: print("nothing to scan", flush=True) break lock = threading.Lock() found = set()
def scan(did): hits = set() for coll in COLLECTIONS: for rec in list_records(PM[did], did, coll): hits |= dids_in(rec.get("value") or {}) with lock: found.update(hits) scanned.add(did) return None
par(todo, scan, n=16, label="social") new = {d for d in found if d.startswith("did:") and d not in people} print(f" DIDs referenced: {len(found)} NEW people: {len(new)}", flush=True)
if new: print(f" resolving {len(new)} PDSes", flush=True) PM.update(dict(par(sorted(new), pds_of, n=10, label="pds"))) json.dump(PM, open(f"{SD}/pds_map.json", "w")) people |= new json.dump(sorted(people), open(f"{SD}/closure_people.json", "w")) json.dump(sorted(scanned), open(f"{SD}/social_scanned.json", "w")) if not new: print(" closed - no new people", flush=True) break
print(f"\nDONE people={len(people)}", flush=True)
if __name__ == "__main__": main()