#!/usr/bin/env python3 """Grade the 2026-11-01 per-user post census (Extract lane). WHAT IT GRADES -- the committed snapshot, not the host and not the calendar. json the JSON is internally consistent: exactly the four documented keys, rows carry exactly the six documented fields in order, ids are 1..N with no gaps, n_users == len(rows), and total_posts == sum(post_count). csv re-rendering the JSON rows as CSV is byte-identical to the committed 2026-11-01-posts-census.csv (header included). md re-rendering the JSON rows as the markdown table is byte-identical to the committed 2026-11-01-posts-census.md, summary line included (the headline carries the population and the post total, so a stale count is a fault here, not a footnote). live only with --live: fetch /users + /posts, derive the rows, and compare them with the snapshot. An unreachable source is rc 2 -- neither a pass nor a fault. WHY THE DEFAULT IS OFFLINE The artifact is dated 2026-11-01 and records one reading of a public API. A gate that re-fetches on every run grades the network's mood, not the tree: it cannot run in a clean room and it fails for reasons the artifact did not choose. The snapshot (JSON) is the record; CSV and MD are its two independent renderings, and agreeing with each other is the property that can be graded anywhere. --live re-checks the record against the source when a reader wants that, and says so. EXIT CODES 0 all graded legs pass, 1 a leg drifted, 2 cannot grade (missing or unparseable artifact, or --live with the source unreachable). USAGE python3 grade_posts_census.py # offline replay python3 grade_posts_census.py --live # replay + source re-check python3 grade_posts_census.py --dir DIR # grade DIR (probe batteries) python3 grade_posts_census.py --json-out FILE """ import argparse import csv import io import json import os import sys import urllib.request FIELDS = ["id", "name", "username", "email", "city", "post_count"] BASE = "https://jsonplaceholder.typicode.com" HERE = os.path.dirname(os.path.abspath(__file__)) STEM = "2026-11-01-posts-census" def artifact_dir(override=None): """Artifacts sit beside the CHECK that grades them. Not the host path and not the git root: a copy of the directory (a clone, a tarball, a probe) grades its own bytes. `--dir` points the grader at another directory for a probe battery. """ return os.path.abspath(override) if override else HERE def render_csv(rows): buf = io.StringIO() w = csv.writer(buf, lineterminator="\n") w.writerow(FIELDS) for r in rows: w.writerow([r[f] for f in FIELDS]) return buf.getvalue() def render_md(rows): total = sum(int(r["post_count"]) for r in rows) out = ["# Per-user Post Census (JSONPlaceholder)", "", "Table of each user with their post count " f"({len(rows)} users, {total} posts total).", "", "| id | name | username | email | city | post_count |", "|---:|---|---:|---:|---:|---:|"] for r in rows: out.append(f"| {r['id']} | {r['name']} | {r['username']} | " f"{r['email']} | {r['city']} | {r['post_count']} |") return "\n".join(out) + "\n" def load_json(path): with open(path, encoding="utf-8") as f: return json.load(f) def check_json_shape(j): problems = [] if list(j.keys()) != ["source", "n_users", "total_posts", "rows"]: problems.append(f"top-level keys {list(j.keys())} != " "['source', 'n_users', 'total_posts', 'rows']") rows = j.get("rows") if not isinstance(rows, list) or not rows: return ["rows is not a non-empty list"] for i, r in enumerate(rows): if list(r.keys()) != FIELDS: problems.append(f"row {i} keys {list(r.keys())} != {FIELDS}") elif not isinstance(r["id"], int) or not isinstance(r["post_count"], int): problems.append(f"row {i} id/post_count are not integers") ids = [r.get("id") for r in rows] if ids != list(range(1, len(rows) + 1)): problems.append(f"ids {ids[:3]}... are not 1..{len(rows)}") if j.get("n_users") != len(rows): problems.append(f"n_users {j.get('n_users')} != {len(rows)}") total = sum(int(r["post_count"]) for r in rows if isinstance(r, dict)) if j.get("total_posts") != total: problems.append(f"total_posts {j.get('total_posts')} != {total}") return problems def fetch(path): with urllib.request.urlopen(BASE + path, timeout=20) as r: return json.loads(r.read().decode("utf-8")) def live_rows(): users = fetch("/users") posts = fetch("/posts") by_id = {u["id"]: u for u in users} counts = {} for p in posts: counts[p["userId"]] = counts.get(p["userId"], 0) + 1 rows = [] for uid in sorted(by_id): u = by_id[uid] rows.append({"id": uid, "name": u["name"], "username": u["username"], "email": u["email"], "city": u["address"]["city"], "post_count": counts.get(uid, 0)}) return rows def main(): ap = argparse.ArgumentParser() ap.add_argument("--live", action="store_true", help="also re-derive from JSONPlaceholder") ap.add_argument("--json-out", default=None) ap.add_argument("--dir", default=None, help="grade artifacts in this directory instead of beside " "the script (probe batteries)") args = ap.parse_args() outdir = artifact_dir(args.dir) jpath = os.path.join(outdir, STEM + ".json") cpath = os.path.join(outdir, STEM + ".csv") mpath = os.path.join(outdir, STEM + ".md") missing = [p for p in (jpath, cpath, mpath) if not os.path.isfile(p)] if missing: print("CANNOT GRADE missing artifact(s):") for p in missing: print(" -", p) return 2 try: snap = load_json(jpath) except Exception as e: print(f"CANNOT GRADE json unreadable: {e}") return 2 results = [] legs = {"json": 0, "csv": 0, "md": 0} shape = check_json_shape(snap) if shape: legs["json"] = 1 results.append(("json", shape)) rows = snap["rows"] with open(cpath, encoding="utf-8", newline="") as f: got_csv = f.read() exp_csv = render_csv(rows) if got_csv != exp_csv: legs["csv"] = 1 results.append(("csv", [f"csv {len(got_csv)} B != re-rendered " f"{len(exp_csv)} B"])) with open(mpath, encoding="utf-8") as f: got_md = f.read() exp_md = render_md(rows) if got_md != exp_md: legs["md"] = 1 results.append(("md", [f"md {len(got_md)} B != re-rendered " f"{len(exp_md)} B"])) live_note = "" if args.live: try: derived = live_rows() except Exception as e: print(f"CANNOT GRADE data source unreachable: {e}") return 2 if derived != rows: legs["live"] = 1 diff = [i for i in range(max(len(derived), len(rows))) if i >= len(rows) or i >= len(derived) or derived[i] != rows[i]] results.append(("live", [f"{len(diff)} row(s) differ from the " f"live API, first at index {diff[0]}"])) else: live_note = f", live {len(derived)} rows match the API" for leg in ("json", "csv", "md", ("live" if args.live else None)): if leg is None: continue state = "PASS" if legs.get(leg, 0) == 0 else "FAIL" print(f"{state} {leg}") bad = [name for name, _ in results] if bad: print("POSTS CENSUS MISMATCH") for name, probs in results: for p in probs: print(f" - {name}: {p}") return 1 total = sum(int(r["post_count"]) for r in rows) print(f"VERDICT=PASS posts-census-2026-11-01 users={len(rows)} " f"posts={total} rows={len(rows)}{live_note}") if args.json_out: with open(args.json_out, "w", encoding="utf-8") as f: json.dump({"verdict": "PASS", "users": len(rows), "posts": total, "live": bool(args.live)}, f, indent=2) return 0 if __name__ == "__main__": sys.exit(main())