101 lines
3.8 KiB
Python
101 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Verification for the arXiv cs.AI extraction artifact (ledger / Extract lane).
|
|
|
|
Re-fetches the same arXiv API query the artifact was built from and checks that
|
|
the on-disk data matches a fresh extraction (re-extract + diff), validates the
|
|
schema, and checks the row count.
|
|
|
|
House exit contract: 0 = artifact matches a fresh fetch; nonzero = mismatch.
|
|
"""
|
|
import json
|
|
import sys
|
|
import urllib.request
|
|
from datetime import datetime, timezone
|
|
|
|
OUT_DIR = "/root/.hermes/wizard-outputs/ledger"
|
|
ARTIFACT_FILES = ["arxiv-csai-10.csv", "arxiv-csai-10.json", "arxiv-csai-10.md"]
|
|
QUERY = "https://export.arxiv.org/api/query?search_query=cat:cs.AI&max_results=10"
|
|
MAX_RESULTS = 10
|
|
|
|
fails = []
|
|
|
|
|
|
def check(name, cond, detail=""):
|
|
print("[%s] %s %s" % ("PASS" if cond else "FAIL", name, detail))
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
def local(tag):
|
|
return tag.split("}", 1)[-1]
|
|
|
|
|
|
def fetch_atom(url):
|
|
req = urllib.request.Request(url, headers={"User-Agent": "ledger/1.0"})
|
|
with urllib.request.urlopen(req, timeout=40) as r:
|
|
raw = r.read()
|
|
if r.status != 200:
|
|
sys.exit("arXiv API returned HTTP %s" % r.status)
|
|
return raw
|
|
|
|
|
|
def parse_atom(xml_bytes):
|
|
import xml.etree.ElementTree as ET
|
|
root = ET.fromstring(xml_bytes)
|
|
entries = []
|
|
for e in root.iter():
|
|
if local(e.tag) != "entry":
|
|
continue
|
|
id_ = next((c.text for c in e.iter() if local(c.tag) == "id"), "").strip()
|
|
title = next((c.text for c in e.iter() if local(c.tag) == "title"), "").strip()
|
|
submitted = next((c.text for c in e.iter() if local(c.tag) == "published"), "").strip()
|
|
authors = [a.text.strip() for a in e.iter() if local(a.tag) == "name"]
|
|
terms = [c.get("term") for c in e.iter() if local(c.tag) == "category" and c.get("term")]
|
|
summary = next((c.text for c in e.iter() if local(c.tag) == "summary"), "").strip()
|
|
entries.append({"id": id_, "submitted": submitted, "title": title,
|
|
"authors": "; ".join(authors[:3]),
|
|
"categories": ", ".join(terms), "summary": summary})
|
|
return entries
|
|
|
|
|
|
def main():
|
|
# 1) fresh fetch + re-extract
|
|
fresh_xml = fetch_atom(QUERY)
|
|
fresh = parse_atom(fresh_xml)
|
|
check("fresh row count == max_results", len(fresh) == MAX_RESULTS, "got %d" % len(fresh))
|
|
|
|
# 2) on-disk artifact present and readable
|
|
for fn in ARTIFACT_FILES:
|
|
try:
|
|
open(OUT_DIR + "/" + fn, encoding="utf-8").read()
|
|
except OSError as exc:
|
|
check("artifact present " + fn, False, str(exc))
|
|
return
|
|
with open(OUT_DIR + "/arxiv-csai-10.json", encoding="utf-8") as f:
|
|
artifact_rows = json.load(f)
|
|
check("artifact row count == max_results", len(artifact_rows) == MAX_RESULTS,
|
|
"got %d" % len(artifact_rows))
|
|
|
|
# 3) compare against a fresh extraction
|
|
fresh.sort(key=lambda r: r["submitted"], reverse=True)
|
|
artifact = sorted(artifact_rows, key=lambda r: r["submitted"], reverse=True)
|
|
check("id set matches a fresh fetch",
|
|
[r["id"] for r in artifact] == [r["id"] for r in fresh],
|
|
"artifact %d rows vs fresh %d rows" % (len(artifact), len(fresh)))
|
|
|
|
# 4) per-row schema + content integrity
|
|
schema_ok = all(set(r.keys()) == {"id", "submitted", "title", "authors", "categories", "summary"}
|
|
for r in artifact)
|
|
check("schema keys exact", schema_ok)
|
|
iso_ok = all(isinstance(r["submitted"], str) and "T" in r["submitted"] for r in artifact)
|
|
check("submitted is ISO-8601 datetime", iso_ok)
|
|
nonblank = all(r["title"] and r["id"] and r["submitted"] and r["categories"] for r in artifact)
|
|
check("no blank title/id/submitted/categories", nonblank)
|
|
|
|
print("verdict: %s" % ("PASS" if not fails else "FAIL"))
|
|
sys.exit(1 if fails else 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|