#!/usr/bin/env python3 """Score every claim on the bounty and rewrite the leaderboard. python3 judge_claims.py # fetch, score, write results.json python3 judge_claims.py --page # ...and regenerate the page and the card The on-chain claim list is the truth (claims_onchain.js); poidh's indexer supplies the image URL and is allowed to lag. A claim that exists on chain but has no image in the indexer gets a row saying so rather than being dropped -- publishing "every score, win or lose" is a promise in immutable text, and the way to break it silently is to judge a subset and never notice. Scoring is score_image() from the sieve-test scorer: the same function that produced the published false-positive numbers, no second copy of the rule. """ import argparse, hashlib, io, json, os, subprocess, sys, time, urllib.request sys.path.insert(0, "/home/agent/work/sieve-test") from PIL import Image from score import load_ctx, score_image, MODEL, MANIFEST # noqa: E402 HERE = os.path.dirname(os.path.abspath(__file__)) SUBS = f"{HERE}/subs" B = json.load(open(f"{HERE}/bounty-result.json")) ISSUER = B["issuer"].lower() if "issuer" in B else "0x1c7afa67130ee637765a8281e83342e307409d57" MAX_BYTES = 40 << 20 UA = {"User-Agent": "agentatwork-fooldet/1.0 (+https://agentatwork.xyz/fool-the-detector/)"} def get(url, cap=MAX_BYTES): """Fetch, refusing to read an unbounded body into memory.""" req = urllib.request.Request(url, headers=UA) with urllib.request.urlopen(req, timeout=60) as r: ctype = (r.headers.get("content-type") or "").split(";")[0].strip() data = r.read(cap + 1) if len(data) > cap: raise ValueError(f"larger than {cap} bytes") return data, ctype def fetch_image(url): """Download the submission. One level of metadata indirection is followed. A poidh claim uri points at metadata JSON with an `image` key -- some clients hand that url straight through, so what the indexer calls an imageUrl is sometimes the json. Follow it once; twice would be someone playing games. """ data, ctype = get(url) if ctype in ("application/json", "text/json") or data[:1] in (b"{", b"["): try: meta = json.loads(data.decode("utf-8", "replace")) except ValueError: meta = None if isinstance(meta, dict) and meta.get("image"): data, ctype = get(meta["image"]) url = meta["image"] return data, ctype, url def match_all(chain_claims, site_claims): """Pair each on-chain claim with poidh's indexed copy of it. The index does not publish the on-chain claim id -- its `claimId` is the site's own database row (7743 for chain claim 2427). So match on what both sides carry: the submitter's address, the title, and the description. Two passes, and no third. Taking "the submitter's first unused row" as a fallback is what would quietly attribute one entrant's photograph to another of their claims: on bounty 320 that fallback scored claim 2176 against a row belonging to 2152. An unmatched claim is published as unmatched -- a missing score is visible, a wrong one is not. """ PASSES = [("name", "title", 0), ("description", "description", 120)] used, out = set(), {} for chain_field, site_field, trim in PASSES: for c in chain_claims: if c["id"] in out: continue want = (c.get(chain_field) or "").strip() want = want[:trim] if trim else want if not want: continue for s in site_claims: if int(s["claimId"]) in used or s["issuerAddress"].lower() != c["issuer"]: continue have = (s.get(site_field) or "").strip() have = have[:trim] if trim else have if have == want: out[c["id"]] = s used.add(int(s["claimId"])) break return out, [int(s["claimId"]) for s in site_claims if int(s["claimId"]) not in used] def score_claim(c, ctx, site): """One claim -> one published row. Never raises: an unscoreable entry is a result.""" rec = {"claim_id": c["id"], "claim_url": B["site"], "site_claim_id": site.get("claimId"), "by": c.get("name") or "—", "submitter": c["issuer"], "created_at": c.get("created_at"), "accepted": c.get("accepted", False)} if site.get("farcasterHandle"): rec["farcaster"] = site["farcasterHandle"] if c["issuer"] == ISSUER: rec["skip"] = "the bounty issuer's own claim — not eligible" return rec url = site.get("imageUrl") if not url: rec["skip"] = "on chain, no image in poidh's index yet — rescored on the next run" return rec rec["image_url"] = url path = None try: data, ctype, final = fetch_image(url) rec["image_url"] = final rec["content_type"] = ctype rec["sha256"] = hashlib.sha256(data).hexdigest() rec["bytes"] = len(data) img = Image.open(io.BytesIO(data)) rec["format"] = img.format img = img.convert("RGB") rec["w"], rec["h"] = img.size os.makedirs(SUBS, exist_ok=True) # kept so a score is re-checkable path = f'{SUBS}/claim{c["id"]}.{(img.format or "bin").lower()}' open(path, "wb").write(data) rec.update(score_image(img, ctx)) except Exception as e: rec["error"] = f"{type(e).__name__}: {e}" if path: rec["local"] = os.path.basename(path) return rec def main(): ap = argparse.ArgumentParser() ap.add_argument("--page", action="store_true", help="regenerate the page and card after") ap.add_argument("--test", type=int, metavar="BOUNTY_ID", help="score another bounty's claims and print them; writes nothing") ap.add_argument("--test-site", metavar="URL", help="that bounty's poidh page, for image urls") a = ap.parse_args() if a.test: # keep exercise downloads out of the real board globals()["SUBS"] = f"{HERE}/subs/test" argv = ["node", f"{HERE}/claims_onchain.js"] + ([str(a.test)] if a.test else []) chain = json.loads(subprocess.check_output(argv, text=True)) site_url = a.test_site or B["site"] data, _ = get(f'{site_url}/data') site = json.loads(data) site_claims = site.get("claims", []) site_claims.sort(key=lambda s: int(s["claimId"])) # the site's own db ids, ascending ctx = load_ctx() paired, stray = match_all(chain["claims"], site_claims) rows, missing = [], [] for c in chain["claims"]: s = paired.get(c["id"]) if not s: missing.append(c["id"]) rows.append(score_claim(c, ctx, s or {})) out = { "bounty_id": chain["bounty_id"], "site_id": B["site_id"], "site": B["site"], "chain": "base", "open": chain["open"], "accepted_claim_id": chain["accepted_claim_id"], "scored_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "read_at_block": chain["block"], "model": {"path": os.path.basename(MODEL), "sha256": hashlib.sha256(open(MODEL, "rb").read()).hexdigest(), "manifest": json.load(open(MANIFEST))["version"]}, "indexer_missing": missing, # on chain, absent from poidh's index "indexer_only": stray, # in the index, not in the contract's list "claims": rows, } if not a.test: json.dump(out, open(f"{HERE}/results.json", "w"), indent=1, ensure_ascii=False) for r in rows: s = r.get("score") print(f'claim {r["claim_id"]:<6} ' + (f'{s:.4f} {"AI-FLAGGED" if s >= 0.65 else "read as real"}' f'{" (tta)" if r.get("tta") else ""}' if s is not None else r.get("skip") or r.get("error", "not scored"))) scored = [r for r in rows if r.get("score") is not None] print(f'\n{len(rows)} claims on chain, {len(scored)} scored' + (f', best {max(r["score"] for r in scored):.4f}' if scored else '')) if missing: print(f'WARN: {len(missing)} claim(s) on chain with no match in the index: {missing}') if stray: print(f'WARN: {len(stray)} indexed claim(s) matched to nothing on chain: {stray}') if a.page: for script in ("gen_page.py", "gen_card.py", "gen_cardjson.py"): subprocess.check_call(["python3", script], cwd=HERE) subprocess.check_call(["rsvg-convert", "-w", "1200", "card.svg", "-o", "card-image.png"], cwd=HERE) for f in ("card.json", "card-image.png", "card.svg", "judge_claims.py", "claims_onchain.js"): subprocess.check_call(["cp", f"{HERE}/{f}", "/var/www/aaw/fool-the-detector/"]) print("page, card and scripts refreshed") return 0 if __name__ == "__main__": sys.exit(main())