#!/usr/bin/env python3 """Stage 9: per-fold manifests of the TEST split. For every fold writes: fold_k/test_manifest.tsv one row per test record: id, kind, source, group, author_title, nchars, preview fold_k/test_works.tsv aggregated by group (volume/work/TM/PHI): group, kind, source, records, chars, author_title, preview and TEST_MANIFEST.md at the root with per-fold summaries. "group" is the same work/volume granularity the pipeline used for splitting: oga tlgXXXX.tlgYYY, catholic/greek_pd/ia/pg/gutenberg volume prefix, TM: for papyri, PHI: for inscriptions. Texts listed here are guaranteed (see verify_report) to share no 8-word sequence and no sentence with the fold's train (or val) set -- safe for reconstruction evaluation. """ import io import os import re import sys from collections import defaultdict import orjson import zstandard as zstd sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) OUTDIR = os.path.expandvars(os.environ.get("FOLD_OUTDIR", "$STOICHEIA_DATA")) OGA_XML = os.path.expandvars("$STOICHEIA_DATA/raw/oga/" "opera_graeca_adnotata_v0.2.0/work_chronology/texts/" "chronology_greek_works.xml") DDBDP_JSONL = os.path.expandvars("$STOICHEIA_DATA/clean/ddbdp.jsonl") PAPYRI_TM_JSONL = os.path.expandvars("$STOICHEIA_DATA/data/papyri_clean.jsonl") def load_oga_names(): names = {} if not os.path.exists(OGA_XML): return names txt = io.open(OGA_XML, encoding="utf-8").read() for m in re.finditer(r"(.*?)", txt, re.S): blk = m.group(1) def g(tag): mm = re.search("<%s>(.*?)" % (tag, tag), blk, re.S) return mm.group(1).strip() if mm else "" urn = g("urn_cts") if urn: names[urn] = "%s — %s" % (g("author"), g("title_labels")) return names 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 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 orjson.loads(line) def clean(s, n=95): return re.sub(r"\s+", " ", s or "").strip()[:n] def main(): oga = load_oga_names() id2tm = load_ddbdp_tm() print("oga names: %d, ddbdp->TM: %d" % (len(oga), len(id2tm))) md = ["# Test-split manifests", "", "One row per test record in `fold_k/test_manifest.tsv`; aggregated", "by work/volume/document in `fold_k/test_works.tsv`.", "Every text listed is verified to share no 8-word sequence and no", "complete sentence with that fold's train and val sets (after", "orthographic normalization) — safe targets for text-reconstruction", "evaluation of a model trained on the same fold's train set.", ""] for k in range(10): d = os.path.join(OUTDIR, "fold_%d" % k) rows = [] for r in read_zst(os.path.join(d, "test.jsonl.zst")): rid = r["id"] src = r.get("source", "") if r.get("tier") == "inscriptions": kind = "inscription" group = "PHI:%s" % r.get("PHI_ID") name = clean(r.get("main_region", "") or "", 40) text = r.get("with_diacritics") or r.get("edition") or \ r.get("ithaca_text") or "" else: text = r.get("text", "") if src == "dclp": kind = "papyrus" group = "TM:" + rid.split("_")[0] name = "" elif src == "ddbdp": kind = "papyrus" tm = id2tm.get(rid) group = "TM:%s" % tm if tm else "ddbdp:" + rid name = rid else: kind = "literary" if src == "oga": group = ".".join(rid.split(".")[:2]) name = oga.get(group, "") elif "#" in rid: group = rid.split("#")[0] name = group if src == "catholic" else "" else: group = rid name = "" group = src + ":" + group rows.append((rid, kind, src, group, name, len(text), clean(text))) with io.open(os.path.join(d, "test_manifest.tsv"), "w", encoding="utf-8") as f: f.write("id\tkind\tsource\tgroup\tauthor_title\tnchars\tpreview\n") for row in rows: f.write("\t".join(str(x) for x in row) + "\n") groups = defaultdict(lambda: [0, 0, "", "", "", ""]) for rid, kind, src, group, name, nch, prev in rows: g = groups[group] g[0] += 1 g[1] += nch if nch >= len(g[5]): g[2], g[3], g[4], g[5] = kind, src, name or g[4], prev elif name and not g[4]: g[4] = name with io.open(os.path.join(d, "test_works.tsv"), "w", encoding="utf-8") as f: f.write("group\tkind\tsource\trecords\tchars\tauthor_title\tpreview\n") for group, g in sorted(groups.items(), key=lambda x: -x[1][1]): f.write("%s\t%s\t%s\t%d\t%d\t%s\t%s\n" % (group, g[2], g[3], g[0], g[1], g[4], g[5])) by_kind = defaultdict(lambda: [0, 0]) for _, kind, *_rest in rows: pass for rid, kind, src, group, name, nch, prev in rows: by_kind[kind][0] += 1 by_kind[kind][1] += nch ngroups = len(groups) md += ["## fold %d — %d records, %d works/documents" % (k, len(rows), ngroups), ""] md += ["| kind | records | Mchars |", "|---|---|---|"] for kind in ("literary", "papyrus", "inscription"): c = by_kind.get(kind, [0, 0]) md.append("| %s | %d | %.1f |" % (kind, c[0], c[1] / 1e6)) md += ["", "Largest test works:", ""] top = sorted(groups.items(), key=lambda x: -x[1][1])[:12] for group, g in top: label = g[4] or g[5][:60] md.append("- `%s` (%d recs, %.2f Mchars) %s" % (group, g[0], g[1] / 1e6, label)) md.append("") print("fold %d: %d records, %d groups" % (k, len(rows), ngroups)) with io.open(os.path.join(OUTDIR, "TEST_MANIFEST.md"), "w", encoding="utf-8") as f: f.write("\n".join(md)) print("wrote", os.path.join(OUTDIR, "TEST_MANIFEST.md")) if __name__ == "__main__": main()