""" Deduplication pipeline for the code corpus. Three stages of deduplication, applied in order: 1. EXACT DEDUP (document-level) - SHA-256 hash of normalized content (whitespace-collapsed, lowercased) - Drops byte-identical and whitespace-only-differing documents. 2. NEAR-DEDUP (document-level, MinHash + LSH) - Shingles each document into character n-grams. - Computes MinHash signatures. - Uses LSH banding to cluster near-duplicates (Jaccard >= threshold). - Keeps one representative per cluster (the longest, highest-quality). 3. LINE-LEVEL DEDUP (cross-document) - Builds a global set of exact line hashes. - Removes lines that appear in >threshold documents (boilerplate: license headers, auto-generated stubs, repeated imports, etc.). - Drops documents that become too short / too sparse after line removal. Input: data/raw_large/documents.jsonl Output: data/dedup/documents_dedup.jsonl Reports statistics at each stage so we can verify the pipeline is working. """ import hashlib import json import os import re import sys from collections import Counter, defaultdict from datasketch import MinHash, MinHashLSH PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DATA_DIR = os.path.join(PROJECT_DIR, "data") RAW_DOCS_PATH = os.path.join(DATA_DIR, "raw_large", "documents.jsonl") DEDUP_DIR = os.path.join(DATA_DIR, "dedup") DEDUP_DOCS_PATH = os.path.join(DEDUP_DIR, "documents_dedup.jsonl") STATS_PATH = os.path.join(DEDUP_DIR, "dedup_stats.json") # ─── MinHash / LSH parameters ──────────────────────────────────────────────── SHINGLE_SIZE = 5 # character 5-grams NUM_PERM = 128 # MinHash permutations (higher = more accurate) LSH_THRESHOLD = 0.5 # Jaccard similarity threshold for near-dup LSH_BANDS = 32 # LSH bands (NUM_PERM / LSH_ROWS should be integer) LSH_ROWS = NUM_PERM // LSH_BANDS # 128 / 32 = 4 # ─── Line-level dedup parameters ──────────────────────────────────────────── LINE_BOILERPLATE_THRESHOLD = 50 # lines appearing in >= this many docs are boilerplate MIN_LINES_AFTER_DEDUP = 5 # drop docs with fewer lines after line dedup MIN_CHARS_AFTER_DEDUP = 200 # drop docs shorter than this after line dedup def normalize_content(text: str) -> str: """Normalize content for exact-hash comparison: collapse whitespace, lowercase.""" # Collapse all whitespace sequences to single space text = re.sub(r"\s+", " ", text) return text.strip().lower() def exact_hash(text: str) -> str: """SHA-256 of normalized content.""" return hashlib.sha256(normalize_content(text).encode("utf-8")).hexdigest() def shingles(text: str, k: int = SHINGLE_SIZE) -> set[str]: """Character k-gram shingles of normalized text.""" text = re.sub(r"\s+", " ", text).strip().lower() if len(text) < k: return {text} return {text[i : i + k] for i in range(len(text) - k + 1)} def minhash_signature(text: str, num_perm: int = NUM_PERM) -> MinHash: """Compute MinHash signature for a document.""" m = MinHash(num_perm=num_perm) for s in shingles(text): m.update(s.encode("utf-8")) return m # ─── Stage 1: Exact dedup ──────────────────────────────────────────────────── def stage1_exact_dedup(docs: list[dict]) -> tuple[list[dict], dict]: """Remove exact (and whitespace-only) duplicates. Keep first occurrence.""" print("\n" + "=" * 60) print("STAGE 1: Exact document-level dedup (SHA-256 of normalized content)") print("=" * 60) seen_hashes = {} kept = [] dropped = 0 for i, doc in enumerate(docs): h = exact_hash(doc["content"]) if h in seen_hashes: dropped += 1 continue seen_hashes[h] = i kept.append(doc) if (i + 1) % 5000 == 0: print(f" Processed {i+1}/{len(docs)} | kept {len(kept)} | dropped {dropped}") stats = { "input": len(docs), "kept": len(kept), "dropped_exact": dropped, "drop_rate": dropped / max(len(docs), 1), } print(f" Input: {len(docs)} | Kept: {len(kept)} | Dropped: {dropped} " f"({stats['drop_rate']:.1%})") return kept, stats # ─── Stage 2: Near-dedup (MinHash LSH) ─────────────────────────────────────── def stage2_near_dedup(docs: list[dict]) -> tuple[list[dict], dict]: """Cluster near-duplicates via MinHash LSH, keep best representative per cluster.""" print("\n" + "=" * 60) print(f"STAGE 2: Near-dedup (MinHash LSH, threshold={LSH_THRESHOLD}, " f"perm={NUM_PERM}, bands={LSH_BANDS})") print("=" * 60) # Build LSH index lsh = MinHashLSH( threshold=LSH_THRESHOLD, num_perm=NUM_PERM, params=(LSH_BANDS, LSH_ROWS), ) signatures = {} kept_indices = set() for i, doc in enumerate(docs): content = doc["content"] if len(content) < 50: # Too short for meaningful MinHash — keep as-is kept_indices.add(i) continue mh = minhash_signature(content) signatures[i] = mh # Query for existing near-duplicates candidates = lsh.query(mh) if candidates: # This doc is a near-duplicate of an existing one. # Compare lengths and keep the longer one. existing_best = max(candidates, key=lambda j: len(docs[int(j)]["content"])) if len(content) > len(docs[int(existing_best)]["content"]): # New doc is longer — replace the existing representative kept_indices.discard(int(existing_best)) kept_indices.add(i) # We can't easily remove from LSH, so just add the new one. # The old one is dropped via kept_indices tracking. try: lsh.insert(str(i), mh) except Exception: pass # else: keep existing, drop this one else: # No near-duplicate found — keep this doc kept_indices.add(i) try: lsh.insert(str(i), mh) except Exception: pass if (i + 1) % 5000 == 0: print(f" Processed {i+1}/{len(docs)} | clusters so far: {len(kept_indices)}") kept = [docs[i] for i in sorted(kept_indices)] dropped = len(docs) - len(kept) stats = { "input": len(docs), "kept": len(kept), "dropped_near": dropped, "drop_rate": dropped / max(len(docs), 1), "threshold": LSH_THRESHOLD, "num_perm": NUM_PERM, } print(f" Input: {len(docs)} | Kept: {len(kept)} | Dropped: {dropped} " f"({stats['drop_rate']:.1%})") return kept, stats # ─── Stage 3: Line-level dedup ─────────────────────────────────────────────── def stage3_line_dedup(docs: list[dict]) -> tuple[list[dict], dict]: """Remove boilerplate lines that appear across many documents.""" print("\n" + "=" * 60) print(f"STAGE 3: Line-level dedup (boilerplate threshold={LINE_BOILERPLATE_THRESHOLD})") print("=" * 60) # First pass: count how many documents each normalized line appears in line_doc_count = Counter() for doc in docs: lines = doc["content"].split("\n") # Use normalized line hash to save memory seen_in_doc = set() for line in lines: norm = line.strip() if not norm: continue # Normalize: collapse internal whitespace, lowercase norm = re.sub(r"\s+", " ", norm).lower() if len(norm) < 5: # skip very short lines (braces, etc.) continue h = hashlib.md5(norm.encode("utf-8")).hexdigest() seen_in_doc.add(h) for h in seen_in_doc: line_doc_count[h] += 1 # Identify boilerplate lines (appear in >= threshold documents) boilerplate_hashes = {h for h, c in line_doc_count.items() if c >= LINE_BOILERPLATE_THRESHOLD} print(f" Boilerplate lines identified: {len(boilerplate_hashes)}") # Second pass: remove boilerplate lines from each document kept = [] dropped_too_short = 0 lines_removed_total = 0 for doc in docs: lines = doc["content"].split("\n") new_lines = [] for line in lines: norm = line.strip() if norm: norm_h = re.sub(r"\s+", " ", norm).lower() if len(norm_h) >= 5: h = hashlib.md5(norm_h.encode("utf-8")).hexdigest() if h in boilerplate_hashes: lines_removed_total += 1 continue new_lines.append(line) new_content = "\n".join(new_lines).strip() new_line_count = len([l for l in new_lines if l.strip()]) if new_line_count < MIN_LINES_AFTER_DEDUP or len(new_content) < MIN_CHARS_AFTER_DEDUP: dropped_too_short += 1 continue doc = dict(doc) doc["content"] = new_content kept.append(doc) stats = { "input": len(docs), "kept": len(kept), "dropped_too_short": dropped_too_short, "lines_removed": lines_removed_total, "boilerplate_hashes": len(boilerplate_hashes), } print(f" Input: {len(docs)} | Kept: {len(kept)} | " f"Dropped (too short): {dropped_too_short} | " f"Lines removed: {lines_removed_total}") return kept, stats # ─── Main ──────────────────────────────────────────────────────────────────── def main(): os.makedirs(DEDUP_DIR, exist_ok=True) print(f"Loading documents from {RAW_DOCS_PATH}...") docs = [] with open(RAW_DOCS_PATH, "r", encoding="utf-8") as f: for line in f: docs.append(json.loads(line)) print(f" Loaded {len(docs):,} documents") print(f" Total size: {sum(len(d['content']) for d in docs) / 1e6:.1f} MB") all_stats = {"input_count": len(docs)} # Stage 1: Exact dedup docs, s1 = stage1_exact_dedup(docs) all_stats["stage1_exact"] = s1 # Stage 2: Near-dedup docs, s2 = stage2_near_dedup(docs) all_stats["stage2_near"] = s2 # Stage 3: Line-level dedup docs, s3 = stage3_line_dedup(docs) all_stats["stage3_line"] = s3 # Final stats final_size = sum(len(d["content"]) for d in docs) all_stats["final_count"] = len(docs) all_stats["final_size_mb"] = final_size / 1e6 # Language distribution lang_counts = {} for d in docs: lang_counts[d["language"]] = lang_counts.get(d["language"], 0) + 1 all_stats["language_distribution"] = lang_counts print("\n" + "=" * 60) print("DEDUP COMPLETE") print("=" * 60) print(f" Input: {all_stats['input_count']:,} documents") print(f" Final: {len(docs):,} documents") print(f" Size: {final_size / 1e6:.1f} MB") print(f" Languages: {lang_counts}") # Write deduped documents with open(DEDUP_DOCS_PATH, "w", encoding="utf-8") as f: for doc in docs: f.write(json.dumps(doc, ensure_ascii=False) + "\n") print(f"\nDeduped documents written to {DEDUP_DOCS_PATH}") # Write stats with open(STATS_PATH, "w") as f: json.dump(all_stats, f, indent=2) print(f"Stats written to {STATS_PATH}") if __name__ == "__main__": main()