esh-pfi-infrastructure/scripts/r49-corpus/verify_corpus.py

"""R49 D1 acceptance gate for a built corpus.

The design doc's D1 acceptance is "clean UTF-8, chapter-segmented, zero
boilerplate lines, stable tokenization". Each is checked here as something that
can actually go RED -- a gate that cannot fail is the third failure mode this
target has already recorded, and it is not repeated here.

  python verify_corpus.py <corpus-dir> [--tokenizer PATH]
"""
from __future__ import annotations
import argparse, collections, json, re, sys, unicodedata
from pathlib import Path

#: ⚠ Anchored to line start, and that is not cosmetic. The first draft matched
#: `Produced by` anywhere and went RED on four hits that were all Charlotte
#: Bronte's own prose -- "a chilling effect produced by his steady announcement",
#: "how such a result was produced by such means". A hard rule on a phrase with a
#: common non-boilerplate sense manufactures failures; same shape as the drift
#: detector that fired on the adjective "minor" and stopped work three times.
#: Gutenberg credits always begin a line, so require that.
BOILER = [r"^.*PROJECT GUTENBERG.*$", r"^.*gutenberg\.org.*$", r"^\s*Produced by\b",
          r"^\s*E-text prepared by\b", r"^\s*Transcribed from\b",
          r"^\s*Distributed Proofread", r"^\*\*\*\s*(?:START|END) OF"]

ap = argparse.ArgumentParser()
ap.add_argument("corpus")
ap.add_argument("--tokenizer", default=None)
a = ap.parse_args()
root = Path(a.corpus)
man = json.loads((root / "manifest.json").read_text())
alpha = json.loads((root / "corpus_alphabet.json").read_text())

records = []
for w in man["works"]:
    for line in (root / w["path"]).read_text(encoding="utf-8").splitlines():
        records.append(json.loads(line))
text = "\n\n".join(r["text"] for r in records)
fails = []

def check(name, ok, detail=""):
    print(f"  [{'PASS' if ok else 'FAIL'}] {name}{('  -- ' + detail) if detail else ''}")
    if not ok:
        fails.append(name)

print(f"== {len(records)} chapters, {sum(r['words'] for r in records):,} words, {len(text):,} chars\n")

# 1. boilerplate
hits = {p: len(re.findall(p, text, re.I | re.M)) for p in BOILER}
bad = {p: n for p, n in hits.items() if n}
check("zero Gutenberg boilerplate", not bad, f"found {bad}" if bad else "7 patterns, 0 hits")

# 2. structure
per_work = collections.Counter(r["work"] for r in records)
seq_ok = all(
    [r["chapter"] for r in records if r["work"] == w] == list(range(1, per_work[w] + 1))
    for w in per_work)
check("chapters number 1..N with no gaps", seq_ok, ", ".join(f"{w}:{n}" for w, n in per_work.items()))
check("no empty chapters", all(r["words"] > 100 for r in records),
      f"min {min(r['words'] for r in records)} words")

# 3. typography consistency AFTER normalisation -- the reason normalisation exists
counts = collections.Counter(text)
straight = counts['"'] + counts["'"]
dbl_hyphen = len(re.findall(r"(?<!-)--(?!-)", text))
check("no straight quotes survive", straight == 0, f'" {counts[chr(34)]}, \' {counts[chr(39)]}')
check("no `--` survives", dbl_hyphen == 0, f"{dbl_hyphen} occurrences")
#: An open/close COUNT mismatch is not an error here and asserting equality was
#: a bad gate. Nineteenth-century convention runs a speech across paragraphs by
#: opening each one and closing only the last, so every work carries a surplus of
#: opens -- measured +46 / +49 / +51 on the three works whose quotes were never
#: touched. The real error signature is a paragraph that BEGINS with a closing
#: quote, which convention never produces and a bad conversion does.
paras = [p.strip() for p in text.split("\n\n") if p.strip()]
lead_close = [p[:60] for p in paras if p.lstrip()[:1] == chr(0x201d)]
check("no paragraph opens with a closing quote", not lead_close,
      f"{len(lead_close)} of {len(paras):,} paragraphs" + (f" e.g. {lead_close[0]!r}" if lead_close else ""))
surplus = counts[chr(0x201c)] - counts[chr(0x201d)]
print(f"         open-quote surplus {surplus:+} of {counts[chr(0x201c)]:,} "
      f"(multi-paragraph speech; expected, not a failure)")

# 4. alphabet is the real inventory
observed = {c for c in text if c.isalpha()}
check("alphabet matches the text exactly", observed == set(alpha["letters"]),
      f"declared {len(alpha['letters'])}, observed {len(observed)}, "
      f"diff {sorted(observed ^ set(alpha['letters']))}")

# 5. no control / exotic codepoints
weird = {c for c in text if unicodedata.category(c) in ("Cc", "Cf", "Co", "Cs") and c != "\n"}
check("no control or private-use codepoints", not weird, repr(sorted(weird)))

# 6. tokenizer stability -- F02's byte-fallback lesson, on the real carrier
if a.tokenizer:
    from transformers import AutoTokenizer
    tok = AutoTokenizer.from_pretrained(a.tokenizer)
    sample = text[:400000]
    ids = tok.encode(sample, add_special_tokens=False)
    back = tok.decode(ids)
    check("tokenizer round-trip is lossless", back == sample,
          f"{len(ids):,} tokens from {len(sample):,} chars")
    pieces = tok.convert_ids_to_tokens(ids)
    fallback = [p for p in pieces if "�" in p]
    check("no byte-fallback pieces", not fallback,
          f"{len(fallback)} of {len(pieces):,} pieces")
    total = len(tok.encode(text, add_special_tokens=False))
    print(f"\n  full corpus = {total:,} tokens ({total/sum(r['words'] for r in records):.2f} tok/word)")

print(f"\n== {'ALL CHECKS PASSED' if not fails else 'FAILED: ' + ', '.join(fails)}")
sys.exit(1 if fails else 0)