| """Sinhala continued-pretraining corpus -> data/corpus/clean/*.parquet (+ aggregate stats in reports/). |
| |
| Sources (downloaded to data/raw/corpus by this script, pinned revisions in DATA_LOG.md): |
| fineweb2 HuggingFaceFW/fineweb-2 sin_Sinh (ODC-By; already MinHash-deduplicated upstream) |
| wikipedia wikimedia/wikipedia 20231101.si (CC BY-SA 3.0) |
| nsina sinhala-nlp/NSINA news (CC BY-SA 4.0) |
| |
| Per document, in order (first failing rule is recorded): |
| 1. blocked_domain exam / past-paper / curriculum sites (SinhalaMMLU's sources and anything like them) |
| 2. exam_like question-paper text: "ප්රශ්න පත්ර", "වාර විභාග", "past paper" ... or many numbered option lines |
| 3. too_short < 200 characters |
| 4. not_sinhala < 50% of letters in the Sinhala block |
| 5. repetitive < 30% unique lines (boilerplate, menus) |
| 6. duplicate same normalised text already kept (exact, across sources; FineWeb-2 is near-deduped upstream) |
| 7. contaminated shares a word 8-gram with any SinhalaMMLU dev or Global-MMLU validation question |
| Kept documents get sl_score = Sri Lanka / curriculum-topic keyword hits per 1k characters (used to prioritise). |
| |
| Stages are resumable: finished shards are skipped on rerun. |
| Usage: python scripts/build_cpt_corpus.py [--limit N] [--workers 12] [--out data/corpus] |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import hashlib |
| import json |
| import re |
| import sys |
| import time |
| import unicodedata |
| from collections import Counter |
| from multiprocessing import Pool |
| from pathlib import Path |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parents[1])) |
| from eval.common import ROOT, load_env, read_jsonl |
|
|
| SOURCES = { |
| "fineweb2": ("HuggingFaceFW/fineweb-2", "data/sin_Sinh/train/000_00000.parquet"), |
| "wikipedia": ("wikimedia/wikipedia", "20231101.si/train-00000-of-00001.parquet"), |
| "nsina": ("sinhala-nlp/NSINA", "NSINA.json"), |
| } |
| BLOCKED_DOMAIN = re.compile( |
| r"pastpaper|past-paper|past_paper|thaksalawa|doenets|exam|paperbank|papersbank|model-?paper|term-?test|" |
| r"nie\.lk|edupub|e-?learning|tutor|guru\.lk|alevel|a-level|olevel|o-level|scholarship", re.I) |
| EXAM_PHRASES = re.compile( |
| r"ප්රශ්න\s*පත්ර|වාර\s*විභාග|පසුගිය\s*විභාග|ආදර්ශ\s*ප්රශ්න|පෙරහුරු\s*ප්රශ්න|" |
| r"past\s*papers?|model\s*papers?|term\s*test|marking\s*scheme|ලකුණු\s*දීමේ\s*පටිපාටිය", re.I) |
| OPTION_LINE = re.compile(r"^\s*(\(?[1-5][\).]|\(?[iv]{1,3}\)|[අආඇඉ]\))\s*\S", re.M) |
| SL_TOPICS = re.compile( |
| r"ශ්රී\s*ලංකා|ලංකාව|සිංහල|බුදු|බෞද්ධ|ධර්ම|විහාර|රජතුමා|රාජධානි|අනුරාධපුර|පොළොන්නරු|මහනුවර|ඉතිහාස|" |
| r"පුරාවිද්යා|සාහිත්ය|කාව්ය|නර්තන|නැටුම|සංගීත|රාග|චිත්ර|කලාව|නාට්ය|කතෝලික|ක්රිස්තියානි|ඉස්ලාම්|" |
| r"භූගෝල|පුරවැසි|ආණ්ඩුක්රම|සෞඛ්ය|විද්යාව") |
| SINHALA = re.compile(r"[-]") |
| LETTER = re.compile(r"[^\W\d_]") |
|
|
|
|
| def norm(text: str) -> str: |
| text = unicodedata.normalize("NFC", text).lower() |
| text = re.sub(r"[]", "", text) |
| |
| return re.sub(r"\s+", " ", re.sub(r"[^\w\s\u0D80-\u0DFF]", " ", text)).strip() |
|
|
|
|
| def word_grams(text: str, n: int = 8): |
| w = norm(text).split() |
| return {hash(" ".join(w[i:i + n])) for i in range(len(w) - n + 1)} |
|
|
|
|
| _CONTAM: set[int] = set() |
|
|
|
|
| def _init(contam: set[int]) -> None: |
| global _CONTAM |
| _CONTAM = contam |
|
|
|
|
| def check(doc: dict) -> tuple[str, dict]: |
| """Return (reason or 'kept', doc with stats). Duplicate check happens in the parent (needs global state).""" |
| text, url = doc["text"], doc.get("url") or "" |
| if url and BLOCKED_DOMAIN.search(url.split("/")[2] if "://" in url else url): |
| return "blocked_domain", doc |
| if EXAM_PHRASES.search(text[:5000]) or len(OPTION_LINE.findall(text)) >= 8: |
| return "exam_like", doc |
| if len(text) < 200: |
| return "too_short", doc |
| letters = LETTER.findall(text) |
| if not letters or len(SINHALA.findall(text)) / len(letters) < 0.5: |
| return "not_sinhala", doc |
| lines = [l.strip() for l in text.splitlines() if l.strip()] |
| if len(lines) >= 10 and len(set(lines)) / len(lines) < 0.3: |
| return "repetitive", doc |
| if _CONTAM and word_grams(text) & _CONTAM: |
| return "contaminated", doc |
| doc["n_chars"] = len(text) |
| doc["sl_score"] = round(1000 * len(SL_TOPICS.findall(text)) / len(text), 3) |
| doc["hash"] = hashlib.sha1(norm(text).encode()).hexdigest() |
| return "kept", doc |
|
|
|
|
| def iter_source(name: str, path: Path, limit: int | None, batch: int = 20000): |
| if name == "nsina": |
| rows = json.loads(path.read_text(encoding="utf-8")) |
| out = [] |
| for i, r in enumerate(rows[:limit] if limit else rows): |
| text = "\n".join(x for x in (r.get("Headline"), r.get("News Content")) if x) |
| out.append({"id": f"nsina/{i}", "source": name, "url": r.get("Source") or "", "text": text}) |
| if len(out) == batch: |
| yield out |
| out = [] |
| if out: |
| yield out |
| return |
| import pyarrow.parquet as pq |
|
|
| pf, n = pq.ParquetFile(path), 0 |
| for rb in pf.iter_batches(batch_size=batch, columns=["id", "url", "text"] + (["title"] if name == "wikipedia" else [])): |
| out = [] |
| for r in rb.to_pylist(): |
| text = f"{r['title']}\n{r['text']}" if name == "wikipedia" else r["text"] |
| out.append({"id": f"{name}/{r['id']}", "source": name, "url": r.get("url") or "", "text": text}) |
| if limit: |
| out = out[: max(limit - n, 0)] |
| n += len(out) |
| if out: |
| yield out |
| if limit and n >= limit: |
| return |
|
|
|
|
| def contamination_grams() -> set[int]: |
| grams = set() |
| for f in ("sinhalammlu_dev", "gmmlu_si_val"): |
| for r in read_jsonl(ROOT / "data" / "processed" / f"{f}.jsonl"): |
| grams |= word_grams(r["question"]) |
| grams |= word_grams(r["question"] + " " + r["options"][r["answer"] - 1]) |
| return grams |
|
|
|
|
| def main() -> None: |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--limit", type=int, help="documents per source (smoke test)") |
| ap.add_argument("--workers", type=int, default=12) |
| ap.add_argument("--out", default=str(ROOT / "data" / "corpus")) |
| args = ap.parse_args() |
| load_env() |
| import pandas as pd |
| from huggingface_hub import HfApi, hf_hub_download |
|
|
| out = Path(args.out) |
| (out / "clean").mkdir(parents=True, exist_ok=True) |
| api = HfApi() |
| revisions = {} |
| for name, (repo, f) in SOURCES.items(): |
| sha = api.dataset_info(repo).sha |
| revisions[name] = f"{repo}@{sha}" |
| hf_hub_download(repo, f, repo_type="dataset", revision=sha, |
| local_dir=ROOT / "data" / "raw" / "corpus" / repo.split("/")[1]) |
| contam = contamination_grams() |
| print(f"contamination 8-grams: {len(contam):,}", flush=True) |
|
|
| seen_hashes: set[str] = set() |
| stats = {"revisions": revisions, "sources": {}} |
| t0 = time.time() |
| with Pool(args.workers, initializer=_init, initargs=(contam,)) as pool: |
| for name, (repo, f) in SOURCES.items(): |
| path = ROOT / "data" / "raw" / "corpus" / repo.split("/")[1] / f |
| reasons, kept_chars, sl = Counter(), 0, [] |
| for shard, docs in enumerate(iter_source(name, path, args.limit)): |
| shard_file = out / "clean" / f"{name}-{shard:04d}.parquet" |
| kept = [] |
| for reason, doc in pool.imap(check, docs, chunksize=64): |
| if reason == "kept": |
| if doc["hash"] in seen_hashes: |
| reason = "duplicate" |
| else: |
| seen_hashes.add(doc["hash"]) |
| kept.append(doc) |
| reasons[reason] += 1 |
| if not shard_file.exists() and kept: |
| pd.DataFrame(kept).drop(columns=["hash"]).to_parquet(shard_file, index=False) |
| kept_chars += sum(d["n_chars"] for d in kept) |
| sl += [d["sl_score"] for d in kept] |
| print(f"{name} shard {shard}: {sum(reasons.values()):,} docs seen, {reasons['kept']:,} kept " |
| f"({time.time() - t0:.0f}s)", flush=True) |
| stats["sources"][name] = { |
| "docs_seen": sum(reasons.values()), "reasons": dict(reasons), "kept_chars": kept_chars, |
| "est_gemma_tokens": int(kept_chars * 0.40), |
| "sl_score_quantiles": {q: float(pd.Series(sl).quantile(q / 100)) for q in (50, 90, 99)} if sl else {}, |
| } |
| stats["total_est_gemma_tokens"] = sum(s["est_gemma_tokens"] for s in stats["sources"].values()) |
| stats["seconds"] = round(time.time() - t0) |
| name = "corpus_stats_smoke.json" if args.limit else "corpus_stats.json" |
| (ROOT / "reports" / name).write_text(json.dumps(stats, indent=2, ensure_ascii=False)) |
| print(json.dumps(stats, indent=2, ensure_ascii=False)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
