| |
| """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]) |
| 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 |
| |
| 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() |
|
|
