File size: 6,536 Bytes
7ed86c3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 | #!/usr/bin/env python3
"""Stage 10: adversarial double-check of the materialized folds.
Beyond re-running the independent verifier (stage 7), this asserts:
D1 MUTATION TEST -- the verifier is not vacuous. Records sampled from
fold_0 test, plus orthographically mutated copies (diacritics stripped,
punctuation replaced, sigma variants), MUST be flagged against the test
reference; genuine train records must not.
D2 DIGIT-RULE COMPLIANCE -- in every fold: every test inscription PHI id
ends in 3, every val one in 4; every test/val papyrus TM ends in 3/4;
no train record (any tier) carries a mapped TM/PHI ending in 3 or 4.
D3 ID-DISJOINTNESS -- within each fold, no base record id appears in more
than one of train/val/test.
Writes work/doublecheck_report.json; exits non-zero on any failure.
"""
import importlib.util
import io
import json
import os
import sys
from collections import defaultdict
from concurrent.futures import ProcessPoolExecutor
import orjson
import zstandard as zstd
HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
OUTDIR = os.path.expandvars(os.environ.get("FOLD_OUTDIR", "$STOICHEIA_DATA"))
spec = importlib.util.spec_from_file_location("verify", os.path.join(HERE, "07_verify.py"))
verify = importlib.util.module_from_spec(spec)
spec.loader.exec_module(verify)
DDBDP_JSONL = os.path.expandvars("$STOICHEIA_DATA/clean/ddbdp.jsonl")
PAPYRI_TM_JSONL = os.path.expandvars("$STOICHEIA_DATA/data/papyri_clean.jsonl")
def read_zst(path):
dctx = zstd.ZstdDecompressor()
with open(path, "rb") as f:
with dctx.stream_reader(f) as r:
for line in io.TextIOWrapper(r, encoding="utf-8"):
if line.strip():
yield line
def load_ddbdp_tm():
tm_by_base = {}
with open(PAPYRI_TM_JSONL, "rb") as f:
for line in f:
r = orjson.loads(line)
tm_by_base[r["file"]] = str(r["TM"])
id2tm = {}
with open(DDBDP_JSONL, "rb") as f:
for line in f:
r = orjson.loads(line)
tm = tm_by_base.get(r["file"].rsplit("/", 1)[-1])
if tm:
id2tm[r["id"]] = tm
return id2tm
def mutate(text):
"""Plausible 'other edition': strip diacritics, change punctuation/sigmas."""
import unicodedata
t = unicodedata.normalize("NFD", text)
t = "".join(c for c in t if not unicodedata.combining(c))
return t.replace(".", "·").replace(",", "").replace("ς", "σ")
def d1_mutation_test():
d = os.path.join(OUTDIR, "fold_0")
gk, sk = verify.build_reference([os.path.join(d, "test.jsonl.zst")])
verify.G["gk"], verify.G["sk"] = gk, sk
test_lines, train_lines = [], []
for line in read_zst(os.path.join(d, "test.jsonl.zst")):
test_lines.append(line)
if len(test_lines) >= 400:
break
for line in read_zst(os.path.join(d, "train.jsonl.zst")):
train_lines.append(line)
if len(train_lines) >= 4000:
break
g_plain, s_plain, _ = verify.check_batch(test_lines)
mutated = []
for line in test_lines:
r = orjson.loads(line)
for fld in ("text", "with_diacritics", "edition"):
if isinstance(r.get(fld), str):
r[fld] = mutate(r[fld])
mutated.append(orjson.dumps(r).decode())
g_mut, s_mut, _ = verify.check_batch(mutated)
g_train, s_train, _ = verify.check_batch(train_lines)
res = {"planted_verbatim": {"gram_hits": g_plain, "sent_hits": s_plain},
"planted_mutated_edition": {"gram_hits": g_mut, "sent_hits": s_mut},
"genuine_train_sample": {"gram_hits": g_train, "sent_hits": s_train},
"PASS": g_plain > 0 and s_plain > 0 and g_mut > 0
and g_train == 0 and s_train == 0}
return res
def check_fold(args):
k, id2tm = args
d = os.path.join(OUTDIR, "fold_%d" % k)
want = {"test": "3", "val": "4"}
bad_digit = []
seen = {}
dup_across = []
for split in ("train", "val", "test"):
for line in read_zst(os.path.join(d, split + ".jsonl.zst")):
r = orjson.loads(line)
rid = str(r["id"])
base = rid.split("#")[0]
if base.startswith("phi") and ":" in base:
base = base.split(":")[0]
if base.startswith("tlg") and base.count(".") >= 2:
base = ".".join(base.split(".")[:2]) # oga work granularity
prev = seen.get(base)
if prev is not None and prev != split:
if len(dup_across) < 5:
dup_across.append((base, prev, split))
seen[base] = split
# digit rule
tm = None
if r.get("tier") == "inscriptions":
tm = str(r["PHI_ID"]) if "PHI_ID" in r else base.replace("phi", "")
elif r.get("source") == "dclp":
tm = base.split("_")[0]
elif r.get("source") == "ddbdp":
tm = id2tm.get(base)
if tm is None:
continue
dig = tm.rstrip()[-1]
if split in ("val", "test"):
if dig != want[split] and len(bad_digit) < 5:
bad_digit.append((split, rid, tm))
else:
if dig in ("3", "4") and len(bad_digit) < 5:
bad_digit.append((split, rid, tm))
n_dup = len(dup_across)
return k, {"digit_violations": bad_digit, "cross_split_dup_ids": dup_across,
"PASS": not bad_digit and not dup_across}
def main():
report = {}
print("D1: mutation test of the verifier...", flush=True)
report["D1_mutation"] = d1_mutation_test()
print(json.dumps(report["D1_mutation"], indent=2), flush=True)
id2tm = load_ddbdp_tm()
print("D2+D3: digit-rule + id-disjointness over all folds...", flush=True)
with ProcessPoolExecutor(max_workers=10) as ex:
for k, res in ex.map(check_fold, [(k, id2tm) for k in range(10)]):
report["fold_%d" % k] = res
print("fold %d: %s" % (k, "PASS" if res["PASS"] else
"FAIL " + json.dumps(res)), flush=True)
ok = all(v["PASS"] for v in report.values())
report["ALL_PASS"] = ok
with open(os.path.join(ROOT, "work", "doublecheck_report.json"), "w") as f:
json.dump(report, f, indent=2, default=str)
print("DOUBLECHECK:", "ALL PASS" if ok else "FAILURES FOUND", flush=True)
sys.exit(0 if ok else 1)
if __name__ == "__main__":
main()
|