#!/usr/bin/env python3 """ score.py -- run a JSON-recovery parser against the MALFORMED-300 conformance suite (or the free 30-case sample) and print what it actually does. Standard library only. No network. No telemetry. Public domain (CC0). USAGE python3 score.py --corpus sample30.jsonl --parser json python3 score.py --corpus sample30.jsonl --parser mymodule:recover python3 score.py --corpus sample30.jsonl --parser-cmd "./myparser --stdin" python3 score.py --corpus sample30.jsonl --parser json --baseline base.json python3 score.py --selftest python3 score.py --spec PARSER PROTOCOL --parser module:callable the callable takes one str and returns the parsed value, OR returns None to refuse, OR raises to refuse. No case in this suite has a top-level expected value of null, so None is unambiguous. --parser json the standard library json.loads, as a control. --parser-cmd "..." a subprocess: the malformed text goes in on stdin; a recovered value comes back on stdout as JSON; a non-zero exit or empty stdout counts as a refusal. EXIT CODES 0 scored, and no regression against --baseline (if one was given) 1 usage or harness error 2 a regression against --baseline: fewer exact matches, or more invented values on unrecoverable cases. Wire this into CI. """ import argparse, json, importlib, io, os, subprocess, sys, time SPEC = """MALFORMED-300 grading spec 1. expected_kind "value": pass only by returning exactly that value. Comparison is on json.dumps(v, sort_keys=True, separators=(",",":")) -- key order and whitespace do not matter, types and values do. 2. expected_kind "unrecoverable": pass only by REFUSING. Any returned value, including {} or [] or "", fails. Inventing an empty object where the model produced nothing is the failure this suite exists to measure. 3. Truncated cases: keep every pair or element that was completely written before the cut, drop the incomplete tail, close the open containers, invent nothing. 4. True->true, False->false, None/undefined->null. NaN and Infinity are absent from the suite on purpose: they have no JSON equivalent. 5. Typographic quotes around a key or string map to the ASCII form. 6. No case has a top-level expected value of null. """ def canon(v): return json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False) class Refused(Exception): pass def load_corpus(path): cases = [] with io.open(path, encoding="utf-8") as f: for ln, line in enumerate(f, 1): line = line.strip() if not line: continue c = json.loads(line) for k in ("id", "category", "input", "expected_kind"): if k not in c: raise SystemExit("corpus line %d is missing %r" % (ln, k)) if c["expected_kind"] == "value" and "expected" not in c: raise SystemExit("corpus line %d claims a value and has none" % ln) cases.append(c) if not cases: raise SystemExit("corpus %s is empty" % path) return cases def make_parser(spec, cmd): if cmd: def run(text): p = subprocess.run(cmd, shell=True, input=text.encode("utf-8"), stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) if p.returncode != 0 or not p.stdout.strip(): raise Refused() return json.loads(p.stdout.decode("utf-8")) return run if spec == "json": return json.loads if ":" not in spec: raise SystemExit("--parser wants module:callable, or the word json") mod, fn = spec.split(":", 1) sys.path.insert(0, os.getcwd()) m = importlib.import_module(mod) f = getattr(m, fn) if not callable(f): raise SystemExit("%s is not callable" % spec) return f def score(cases, parser): rows, cats = [], {} exact = invented = refused_right = refused_wrong = errors = 0 t0 = time.time() for c in cases: got, refused, err = None, False, None try: got = parser(c["input"]) if got is None: refused = True except Refused: refused = True except Exception as e: refused = True err = "%s: %s" % (type(e).__name__, e) if c["expected_kind"] == "unrecoverable": ok = refused if refused: refused_right += 1 else: invented += 1 else: ok = (not refused) and canon(got) == canon(c["expected"]) if refused: refused_wrong += 1 if ok: exact += 1 if err: errors += 1 st = cats.setdefault(c["category"], {"n": 0, "ok": 0}) st["n"] += 1 st["ok"] += 1 if ok else 0 rows.append({"id": c["id"], "category": c["category"], "ok": ok, "refused": refused, "error": err, "got": None if refused else canon(got), "expected": "" if c["expected_kind"] == "unrecoverable" else canon(c["expected"])}) n = len(cases) n_unre = sum(1 for c in cases if c["expected_kind"] == "unrecoverable") return { "cases": n, "exact_match": exact, "exact_match_rate": round(exact / n, 4), "unrecoverable_cases": n_unre, "correctly_refused": refused_right, "invented_values_on_unrecoverable": invented, "false_refusals_on_recoverable": refused_wrong, "parser_exceptions": errors, "by_category": {k: {"n": v["n"], "ok": v["ok"], "rate": round(v["ok"] / v["n"], 4)} for k, v in sorted(cats.items())}, "seconds": round(time.time() - t0, 3), }, rows def report(res, name): w = sys.stdout.write w("\nMALFORMED-300 parser: %s\n" % name) w("%-18s %s\n" % ("cases", res["cases"])) w("%-18s %d / %d (%.1f%%)\n" % ("exact match", res["exact_match"], res["cases"], 100 * res["exact_match_rate"])) w("%-18s %d / %d\n" % ("refused correctly", res["correctly_refused"], res["unrecoverable_cases"])) w("%-18s %d <- values invented where the model produced none\n" % ("invented", res["invented_values_on_unrecoverable"])) w("%-18s %d <- gave up on output that was recoverable\n" % ("false refusals", res["false_refusals_on_recoverable"])) w("\n %-16s %5s %5s %7s\n" % ("category", "n", "ok", "rate")) for k, v in res["by_category"].items(): w(" %-16s %5d %5d %6.1f%%\n" % (k, v["n"], v["ok"], 100 * v["rate"])) w("\n") def compare(res, baseline): fails = [] if res["exact_match"] < baseline["exact_match"]: fails.append("exact match fell from %d to %d" % (baseline["exact_match"], res["exact_match"])) if res["invented_values_on_unrecoverable"] > baseline["invented_values_on_unrecoverable"]: fails.append("invented values rose from %d to %d" % (baseline["invented_values_on_unrecoverable"], res["invented_values_on_unrecoverable"])) for k, v in res["by_category"].items(): b = baseline["by_category"].get(k) if b and v["ok"] < b["ok"]: fails.append("%s fell from %d to %d" % (k, b["ok"], v["ok"])) return fails # ------------------------------------------------------------------ selftest def selftest(): """Every number below is derived by hand first, then checked.""" ok = [] def check(name, cond): ok.append((name, bool(cond))) check("canon ignores key order", canon({"b": 1, "a": 2}) == canon({"a": 2, "b": 1}) == '{"a":2,"b":1}') check("canon separates types", canon(1) != canon("1") and canon(True) != canon(1)) corpus = [ {"id": "t1", "category": "x", "input": "a", "expected_kind": "value", "expected": {"a": 1}}, {"id": "t2", "category": "x", "input": "b", "expected_kind": "value", "expected": [1, 2]}, {"id": "t3", "category": "y", "input": "c", "expected_kind": "value", "expected": {"k": "v"}}, {"id": "t4", "category": "z", "input": "d", "expected_kind": "unrecoverable"}, {"id": "t5", "category": "z", "input": "e", "expected_kind": "unrecoverable"}, ] table = {"a": {"a": 1}, "b": [1, 2], "c": {"k": "WRONG"}, "d": None, "e": {}} res, rows = score(corpus, lambda t: table[t]) # hand-derived: t1 ok, t2 ok, t3 wrong value, t4 refused correctly, t5 invented {} check("exact match is 3/5", res["exact_match"] == 3 and res["exact_match_rate"] == 0.6) check("one invented value", res["invented_values_on_unrecoverable"] == 1) check("one correct refusal", res["correctly_refused"] == 1) check("no false refusals", res["false_refusals_on_recoverable"] == 0) check("category x is 2/2", res["by_category"]["x"] == {"n": 2, "ok": 2, "rate": 1.0}) check("category z is 1/2", res["by_category"]["z"] == {"n": 2, "ok": 1, "rate": 0.5}) # an empty object is NOT a pass on an unrecoverable case check("{} does not satisfy a refusal", rows[4]["ok"] is False) # raising is a refusal def raiser(t): raise ValueError("nope") res2, _ = score(corpus, raiser) check("raising refuses everything", res2["correctly_refused"] == 2 and res2["false_refusals_on_recoverable"] == 3 and res2["exact_match"] == 2) check("exceptions counted", res2["parser_exceptions"] == 5) # regression detection base = dict(res) worse = json.loads(json.dumps(res)) worse["exact_match"] = 2 check("a drop in exact match is a regression", compare(worse, base)) better = json.loads(json.dumps(res)) better["exact_match"] = 4 better["by_category"]["y"]["ok"] = 1 check("an improvement is not a regression", compare(better, base) == []) inv = json.loads(json.dumps(res)) inv["invented_values_on_unrecoverable"] = 2 check("more invented values is a regression", compare(inv, base)) for name, good in ok: print("%-42s %s" % (name, "PASS" if good else "FAIL")) bad = [n for n, g in ok if not g] print("\n%d/%d selftest checks passed" % (len(ok) - len(bad), len(ok))) return 0 if not bad else 1 def main(): ap = argparse.ArgumentParser(add_help=True) ap.add_argument("--corpus") ap.add_argument("--parser") ap.add_argument("--parser-cmd") ap.add_argument("--baseline") ap.add_argument("--write-baseline") ap.add_argument("--jsonl-out") ap.add_argument("--json", action="store_true", help="print the result object only") ap.add_argument("--selftest", action="store_true") ap.add_argument("--spec", action="store_true") a = ap.parse_args() if a.spec: print(SPEC) return 0 if a.selftest: return selftest() if not a.corpus or not (a.parser or a.parser_cmd): ap.print_help() return 1 cases = load_corpus(a.corpus) parser = make_parser(a.parser, a.parser_cmd) res, rows = score(cases, parser) name = a.parser_cmd or a.parser res["parser"] = name res["corpus"] = os.path.basename(a.corpus) if a.json: print(json.dumps(res, indent=2, sort_keys=True)) else: report(res, name) if a.jsonl_out: with open(a.jsonl_out, "w") as f: for r in rows: f.write(json.dumps(r, ensure_ascii=False) + "\n") if a.write_baseline: with open(a.write_baseline, "w") as f: json.dump(res, f, indent=2, sort_keys=True) print("baseline written to %s" % a.write_baseline) if a.baseline: with open(a.baseline) as f: base = json.load(f) fails = compare(res, base) if fails: print("REGRESSION against %s:" % a.baseline) for x in fails: print(" - %s" % x) return 2 print("no regression against %s" % a.baseline) return 0 # M-13: the corpus this scorer grades against is 30 of 300 cases. A normal run ends with # one line on STDERR naming where the full corpus lives, so that a copy of this file that # has travelled away from the site still says where it came from. STDOUT is never touched, # so --json output stays byte-parseable; --json also suppresses the line entirely. FULL_CORPUS_URL = "https://buy.stripe.com/4gMfZi5KS7TS6B86JT5Ne09?client_reference_id=artifact-scorer" POINTER_LINE = "MALFORMED-300 full corpus, 300 cases, EUR 29: " + FULL_CORPUS_URL if __name__ == "__main__": _rc = main() if "--json" not in sys.argv: sys.stderr.write(POINTER_LINE + "\n") sys.exit(_rc)