File size: 2,171 Bytes
fd29131 | 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 | #!/usr/bin/env python3
"""Dump a sample of FineWeb-Edu docs that PASS our build-integrated decontam (KEPT docs)
as jsonl {"text": ...}, for Hart's independent decontam_gate.py verification.
Same decontam logic + index as build_v2_blend.py -> if Hart's gate finds >0 hits on these
KEPT docs, our build-decontam has a gap. Expected: 0 hits (parity / clean-by-construction).
"""
import argparse
import hashlib
import json
import re
from pathlib import Path
N_HARD = 13
def normalize(text):
text = text.lower()
text = re.sub(r"[^\w\s]", " ", text)
text = re.sub(r"\s+", " ", text).strip()
return text
def shingles(text, n):
words = normalize(text).split()
return {hashlib.blake2b(" ".join(words[i:i + n]).encode("utf-8"), digest_size=8).hexdigest()
for i in range(len(words) - n + 1)}
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--decontam-index", required=True)
ap.add_argument("--out", required=True)
ap.add_argument("--n-kept", type=int, default=100000)
a = ap.parse_args()
from datasets import load_dataset
idx = json.loads(Path(a.decontam_index).read_text())
hard = set(idx["hard_hashes"])
blimp = set(idx["blimp_hashes"])
def contaminated(text):
sh = shingles(text, N_HARD)
if sh & hard:
return True
if blimp and sh and len(sh & blimp) / len(sh) > 0.005:
return True
return False
ds = load_dataset("HuggingFaceFW/fineweb-edu", "default", split="train", streaming=True)
kept = seen = dropped = 0
with open(a.out, "w", encoding="utf-8") as f:
for ex in ds:
seen += 1
t = ex.get("text") or ""
if contaminated(t):
dropped += 1
else:
f.write(json.dumps({"text": t}) + "\n")
kept += 1
if kept >= a.n_kept:
break
if seen % 20000 == 0:
print(f" seen={seen:,} kept={kept:,} dropped={dropped:,}", flush=True)
print(f"DONE kept={kept:,}/{seen:,} dropped(my-decontam)={dropped:,} -> {a.out}", flush=True)
if __name__ == "__main__":
main()
|