178 lines
7.2 KiB
Python
178 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
|
"""q62_arxiv_verify.py -- grade `wizard-work/ledger/arxiv-papers.md` (Q62 leg c).
|
|
|
|
Q62 leg (c) asks for two things of that file: it must render as four aligned
|
|
columns, and it must carry a provenance line naming the query and the fetch
|
|
timestamp. Both used to be false -- a four-column header sat over rows carrying
|
|
6-11 unescaped pipes (the author lists), and there was no provenance line at
|
|
all.
|
|
|
|
WHAT IT GRADES
|
|
|
|
COLUMNS every table line has exactly 4 data cells once escaped pipes are
|
|
honoured, i.e. exactly 5 unescaped `|`. A row with a bare `|` in
|
|
a title or an author list splits into extra cells and is a FAULT.
|
|
PROVENANCE the document contains a line beginning `Provenance:` that names
|
|
the query URL and an ISO-8601 UTC timestamp.
|
|
RECORDS --fetch: the 12 ids are re-queried at export.arxiv.org in one
|
|
request and title / author list / `published` are compared as
|
|
exact strings against the table. Off by default, because a
|
|
network read is not something a gate should need; the pin is the
|
|
document, the fetch is the audit.
|
|
|
|
EXIT CODES (the tools/ contract)
|
|
0 every graded property holds
|
|
1 ran and graded a fault
|
|
2 could not grade (no table, no ids, --fetch could not reach the API)
|
|
|
|
python3 q62_arxiv_verify.py # offline: columns + provenance
|
|
python3 q62_arxiv_verify.py --fetch # plus the 12 records
|
|
python3 q62_arxiv_verify.py --json out.json
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
import urllib.request
|
|
import xml.etree.ElementTree as ET
|
|
from pathlib import Path
|
|
|
|
NS = {"a": "http://www.w3.org/2005/Atom"}
|
|
QUERY = "https://export.arxiv.org/api/query?id_list="
|
|
ROW_RE = re.compile(r"^\|\s*(?:\[(?P<link>[0-9]{4}\.[0-9]{4,5}v[0-9]+)\]\([^)]*\)"
|
|
r"|(?P<plain>https?://arxiv\.org/abs/[0-9]{4}\.[0-9]{4,5}v[0-9]+))\s*\|(.*)\|\s*$")
|
|
PROV_RE = re.compile(r"^Provenance:.*" + re.escape(QUERY) +
|
|
r".*[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z", re.M)
|
|
|
|
# The 12 versioned ids this document is about. A row that vanishes, or an id
|
|
# that changes version silently, is a FAULT rather than a smaller table.
|
|
EXPECTED_IDS = (
|
|
"2411.18583v1", "2502.00306v2", "2510.22344v1", "2601.05264v1",
|
|
"2503.16581v1", "2412.05838v1", "2504.08748v1", "2502.01113v3",
|
|
"2412.14457v1", "2412.15404v1", "2604.19899v1", "2607.01852v1",
|
|
)
|
|
|
|
|
|
def cells(line):
|
|
"""Split a markdown row on unescaped pipes only."""
|
|
body = line.strip()
|
|
if body.startswith("|"):
|
|
body = body[1:]
|
|
if body.endswith("|"):
|
|
body = body[:-1]
|
|
return [c.replace("\\|", "|").strip() for c in re.split(r"(?<!\\)\|", body)]
|
|
|
|
|
|
def unescaped_pipes(line):
|
|
return len(re.findall(r"(?<!\\)\|", line))
|
|
|
|
|
|
def parse(doc):
|
|
rows, faults = [], []
|
|
for n, line in enumerate(doc.splitlines(), 1):
|
|
if line.startswith("|---") or line.lstrip().startswith("| arXiv ID"):
|
|
continue
|
|
if not line.lstrip().startswith("|"):
|
|
continue
|
|
pipes = unescaped_pipes(line)
|
|
if pipes != 5:
|
|
faults.append(f"COLUMNS line {n}: {pipes} unescaped pipes, expected 5 -> {line[:80]}")
|
|
continue
|
|
c = cells(line)
|
|
if len(c) != 4:
|
|
faults.append(f"COLUMNS line {n}: {len(c)} cells, expected 4")
|
|
continue
|
|
m = ROW_RE.match(line)
|
|
if not m:
|
|
faults.append(f"ID line {n}: no arXiv id cell -> {line[:80]}")
|
|
continue
|
|
rows.append({"line": n, "id": m.group("link") or m.group("plain").rsplit("/", 1)[1],
|
|
"title": c[1], "authors": [a.strip() for a in c[2].split(",") if a.strip()],
|
|
"published": c[3]})
|
|
return rows, faults
|
|
|
|
|
|
def fetch(ids):
|
|
url = QUERY + ",".join(ids) + "&max_results=20"
|
|
raw = urllib.request.urlopen(url, timeout=60).read()
|
|
api = {}
|
|
for entry in ET.fromstring(raw).findall("a:entry", NS):
|
|
aid = entry.find("a:id", NS).text.rsplit("/abs/", 1)[1].strip()
|
|
api[aid] = {
|
|
"title": " ".join(entry.find("a:title", NS).text.split()),
|
|
"authors": [" ".join(a.find("a:name", NS).text.split())
|
|
for a in entry.findall("a:author", NS)],
|
|
"published": entry.find("a:published", NS).text.strip(),
|
|
}
|
|
return url, api
|
|
|
|
|
|
def grade(path, do_fetch):
|
|
doc = path.read_text()
|
|
rows, faults = parse(doc)
|
|
if not PROV_RE.search(doc):
|
|
faults.append("PROVENANCE missing: no `Provenance:` line naming " + QUERY + " and a UTC timestamp")
|
|
if not rows:
|
|
if faults:
|
|
return 1, faults, {"rows": 0}
|
|
return 2, ["no table rows parsed: cannot grade"], {}
|
|
got_ids = [r["id"] for r in rows]
|
|
if sorted(got_ids) != sorted(EXPECTED_IDS):
|
|
missing = [i for i in EXPECTED_IDS if i not in got_ids]
|
|
extra = [i for i in got_ids if i not in EXPECTED_IDS]
|
|
faults.append(f"IDS {len(got_ids)} rows for {len(EXPECTED_IDS)} pinned ids; "
|
|
f"missing {missing} unexpected {extra}")
|
|
detail = {"rows": len(rows), "ids": [r["id"] for r in rows]}
|
|
if do_fetch:
|
|
try:
|
|
url, api = fetch([r["id"] for r in rows])
|
|
except Exception as exc: # noqa: BLE001
|
|
return 2, [f"fetch failed: {exc.__class__.__name__}: {exc}"], detail
|
|
detail["query"] = url
|
|
missing = [i for i in detail["ids"] if i not in api]
|
|
for mid in missing:
|
|
faults.append(f"RECORDS {mid}: no record returned")
|
|
for row in rows:
|
|
a = api.get(row["id"])
|
|
if not a:
|
|
continue
|
|
for field in ("title", "published"):
|
|
if a[field] != row[field]:
|
|
faults.append(f"RECORDS {row['id']} {field}: table {row[field]!r} != api {a[field]!r}")
|
|
if [x for x in a["authors"]] != row["authors"]:
|
|
faults.append(f"RECORDS {row['id']} authors: table {row['authors']} != api {a['authors']}")
|
|
detail["records_returned"] = len(api)
|
|
return (1 if faults else 0), faults, detail
|
|
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
ap.add_argument("--doc", default=str(Path(__file__).resolve().parents[1] /
|
|
"ledger" / "arxiv-papers.md"))
|
|
ap.add_argument("--fetch", action="store_true")
|
|
ap.add_argument("--json", metavar="PATH")
|
|
args = ap.parse_args(argv)
|
|
|
|
path = Path(args.doc)
|
|
if not path.is_file():
|
|
print(f"could not grade: no such document {path}")
|
|
return 2
|
|
rc, faults, detail = grade(path, args.fetch)
|
|
for f in faults:
|
|
print(f"FAULT: {f}")
|
|
if rc == 0:
|
|
print(f"OK: {detail['rows']} rows, 4 columns each, provenance present"
|
|
+ (f", {detail.get('records_returned')} records re-fetched, 0 field mismatches"
|
|
if args.fetch else " (records not re-fetched; --fetch to audit)"))
|
|
elif rc == 2:
|
|
for f in faults:
|
|
print(f"could not grade: {f}")
|
|
if args.json:
|
|
Path(args.json).write_text(json.dumps({"rc": rc, "faults": faults, "detail": detail}, indent=2))
|
|
return rc
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|