Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
4.2 kB · 115 lines
Python
at typos-fix-1
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116"""Validate derived PR state against Tangled's own web UI.
Samples real PR pages, reads the at:// URI and the state off each page, andcompares with what the crawl derived for that exact URI. Matching is by URI, notby title - titles repeat, and matching on them produced a false mismatch earlier."""import json, random, re, html, urllib.parse, urllib.request, collections, sys
SD = "/tmp/claude-1000/-home-jmm-Code-blogposts/7ba1f72e-807b-4749-bcd3-a065da793e6d/scratchpad"C = "https://constellation.microcosm.blue"UA = "tangled-stats-research/0.3 (permadeath.com; blog post stats)"random.seed(11)
def fetch(url, timeout=25): req = urllib.request.Request(url, headers={"User-Agent": UA}) with urllib.request.urlopen(req, timeout=timeout) as f: return f.read().decode("utf8", "replace")
def jget(url): return json.loads(fetch(url))
def repo_ident(repo_did, handles): """repoDid -> (owner handle, repo name) using the sh.tangled.repo record.""" d = jget(f"{C}/links?target={urllib.parse.quote(repo_did, safe='')}" f"&collection=sh.tangled.repo&path=.repoDid&limit=5") recs = d.get("linking_records") or [] if not recs: return None owner, name = recs[0]["did"], recs[0]["rkey"] h = handles.get(owner) if not h: try: doc = jget(f"https://plc.directory/{owner}") aka = [a[5:] for a in doc.get("alsoKnownAs", []) if a.startswith("at://")] h = aka[0] if aka else None except Exception: h = None return (h, name) if h else None
STATE_RE = re.compile(r'\b(Merged|Closed|Open)\b')URI_RE = re.compile(r'at://did:[a-z0-9:._%-]+/sh\.tangled\.repo\.pull/[a-z0-9]+')
def page_state(url): h = fetch(url) uris = URI_RE.findall(h) t = re.sub(r"<script.*?</script>|<style.*?</style>", "", h, flags=re.S) t = " ".join(html.unescape(re.sub(r"<[^>]+>", " ", t)).split()) m = STATE_RE.search(t[:1500]) return (uris[0] if uris else None), (m.group(1).lower() if m else None)
def main(): handles = json.load(open(f"{SD}/handles.json")) derived = {} per_repo = collections.defaultdict(list) for l in open(f"{SD}/pulls.jsonl"): r = json.loads(l) for p in r["pulls"]: derived[p["uri"]] = p["state"] per_repo[r["repoDid"]].append(p)
# sample repos: the busiest few plus random mid-sized ones cands = [(rd, len(v)) for rd, v in per_repo.items() if rd != "unknown" and len(v) >= 8] cands.sort(key=lambda x: -x[1]) picks = cands[:3] + random.sample(cands[3:], min(6, max(0, len(cands) - 3)))
checked = agree = 0 misses = [] for rd, cnt in picks: ident = repo_ident(rd, handles) if not ident: continue h, name = ident base = f"https://tangled.org/@{h}/{name}" try: listing = fetch(f"{base}/pulls") except Exception as e: print(f" skip {h}/{name}: {str(e)[:50]}") continue nums = sorted({int(x) for x in re.findall(rf"/{re.escape(h)}/{re.escape(name)}/pulls/(\d+)", listing)}) if not nums: nums = sorted({int(x) for x in re.findall(r"/pulls/(\d+)", listing)}) if not nums: print(f" skip {h}/{name}: no PR numbers on listing") continue lo, hi = min(nums), max(nums) sample = random.sample(range(lo, hi + 1), min(3, hi - lo + 1)) for num in sample: try: uri, web = page_state(f"{base}/pulls/{num}") except Exception: continue if not uri or not web: continue mine = derived.get(uri) checked += 1 if mine == web: agree += 1 else: misses.append((f"{h}/{name}#{num}", web, mine or "NOT CRAWLED", uri)) print(f" {'OK ' if mine == web else 'DIFF'} {h}/{name}#{num:<5} web={web:7} mine={mine}") print(f"\n{agree}/{checked} agree ({100*agree/checked:.0f}%)" if checked else "\nnothing checked") if misses: print("\ndisagreements:") for m in misses: print(" ", m)
if __name__ == "__main__": main()