File size: 4,409 Bytes
5952424 | 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 | """Build memmap shards from bronze.jsonl (synthetic translated Greek).
Same output format/planes as build_shards.py so the training loader unions gold/silver/bronze
transparently. tier='bronze', clean=0.5 nominal (synthetic). The <95%-Greek filter in
normalize drops encoding-garbage / Latin-contaminated generations automatically.
"""
from __future__ import annotations
import argparse, json, sys
from collections import Counter
from concurrent.futures import ProcessPoolExecutor
from dataclasses import asdict
from pathlib import Path
import numpy as np
import pyarrow as pa
import pyarrow.parquet as pq
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from data.normalize import Stats, normalize_record
PLANES = ("chars", "boundary", "dia", "cap", "punct")
def process_chunk(args):
lines, tmpdir, cid = args
tmpdir = Path(tmpdir); tmpdir.mkdir(parents=True, exist_ok=True)
stats = Stats()
bufs = {p: [] for p in PLANES}
rows = {"offset": [], "length": [], "tier": [], "clean": [], "source": [], "id": []}
off = 0
for line in lines:
try:
rec = json.loads(line)
except Exception:
continue
r = normalize_record(rec.get("text", ""), stats, with_punct=True)
if r is None:
continue
for p, a in zip(PLANES, r):
bufs[p].append(a)
rows["offset"].append(off)
rows["length"].append(len(r[0]))
rows["tier"].append("bronze")
rows["clean"].append(0.5)
rows["source"].append(rec.get("corpus", "bronze"))
rows["id"].append(rec.get("id", ""))
off += len(r[0])
for p in PLANES:
if bufs[p]:
np.concatenate(bufs[p]).tofile(tmpdir / f"{p}.bin")
else:
np.array([], np.uint8).tofile(tmpdir / f"{p}.bin")
pq.write_table(pa.table(rows), tmpdir / "index.parquet")
st = asdict(stats)
st["stripped"] = dict(stats.stripped); st["archaic"] = dict(stats.archaic)
st["other_marks"] = dict(stats.other_marks)
(tmpdir / "stats.json").write_text(json.dumps(st))
return cid, off
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--jsonl", required=True)
ap.add_argument("--out", required=True)
ap.add_argument("--workers", type=int, default=16)
ap.add_argument("--chunk", type=int, default=40000)
a = ap.parse_args()
out = Path(a.out); out.mkdir(parents=True, exist_ok=True)
tmp = out / "tmp"
# split file into line chunks
chunks, buf, cid = [], [], 0
with open(a.jsonl) as f:
for line in f:
buf.append(line)
if len(buf) >= a.chunk:
chunks.append((buf, str(tmp / f"c{cid:04d}"), cid)); buf = []; cid += 1
if buf:
chunks.append((buf, str(tmp / f"c{cid:04d}"), cid))
print(f"{len(chunks)} chunks")
with ProcessPoolExecutor(max_workers=a.workers) as ex:
done = sorted(ex.map(process_chunk, chunks))
order = [c[2] for c in chunks]
total = Stats(); tables, cum = [], 0
for cid in order:
d = tmp / f"c{cid:04d}"
t = pq.read_table(d / "index.parquet")
if t.num_rows:
t = t.set_column(0, "offset", pa.array(t.column("offset").to_numpy() + cum))
tables.append(t)
cum += int(np.sum(t.column("length").to_numpy()))
s = json.loads((d / "stats.json").read_text())
st = Stats(**{k: s[k] for k in ("records_in", "records_kept",
"records_dropped_nongreek", "records_dropped_empty", "letters",
"words", "sentences", "mark_conflicts", "orphan_marks")})
st.stripped = Counter({int(k): v for k, v in s["stripped"].items()})
total.merge(st)
for p in PLANES:
with open(out / f"{p}.bin", "wb") as fo:
for cid in order:
fo.write((tmp / f"c{cid:04d}" / f"{p}.bin").read_bytes())
pq.write_table(pa.concat_tables(tables), out / "index.parquet")
st = asdict(total); st["stripped"] = {str(k): v for k, v in total.stripped.items()}
st["archaic"] = dict(total.archaic); st["other_marks"] = dict(total.other_marks)
(out / "stats.json").write_text(json.dumps(st, indent=2, ensure_ascii=False))
print(f"bronze letters: {cum/1e9:.3f}B kept: {total.records_kept}/{total.records_in} "
f"words: {total.words/1e6:.1f}M")
if __name__ == "__main__":
main()
|