| """Context-aware code / text deduplicator (PRIVATE to Data_Module). NO BLIND LINE HASHING. |
| |
| A single repeated line (`}`, `import os`, `<resources>`, a licence header, "thank you for watching") |
| is never a reason to delete anything. Duplication is only recognised as a RUN of consecutive |
| significant lines (a 3-line shingle window) that was already seen in earlier data: |
| |
| 1. every document is normalised to its significant lines (blank / punctuation-only lines are neutral); |
| 2. a line is "covered" when a 3-line shingle containing it was already indexed; |
| 3. FILE LEVEL overlap < 85 % -> keep the whole file intact |
| every significant line covered -> 100 % duplicate: drop the file |
| 4. SUB-SECTION for files with >= 85 % overlap, each covered run is inspected together with its |
| surrounding context: a run that has >= 3 unique significant lines right before/after it |
| is kept; a pure duplicate without unique context is excised (only that run). |
| |
| Two files with the same name (MainActivity.java, config.xml) but different content therefore both survive, |
| and every kept document is prefixed with its folder context: [PATH: relative/path/filename] |
| """ |
| from __future__ import annotations |
|
|
| import hashlib |
| import re |
| from typing import Dict, List, Optional, Set, Tuple |
|
|
| SHINGLE = 3 |
| OVERLAP_KEEP_BELOW = 0.85 |
| UNIQUE_CONTEXT_LINES = 3 |
| MAX_INDEX = 6_000_000 |
|
|
| _WS = re.compile(r"\s+") |
|
|
|
|
| def _norm(line: str) -> str: |
| return _WS.sub(" ", line.strip()).lower() |
|
|
|
|
| def _significant(norm_line: str) -> bool: |
| return len(norm_line) >= 3 and any(ch.isalnum() for ch in norm_line) |
|
|
|
|
| def _digest(parts: Tuple[str, ...]) -> int: |
| return int.from_bytes(hashlib.blake2b("\n".join(parts).encode("utf-8"), digest_size=8).digest(), "big") |
|
|
|
|
| class ContextDeduplicator: |
| def __init__(self) -> None: |
| self._index: Set[int] = set() |
| self.stats = {"dropped_files": 0, "partial_files": 0, "kept_intact": 0} |
|
|
| |
| @staticmethod |
| def _sig_lines(text: str) -> Tuple[List[str], List[int], List[str]]: |
| raw = text.split("\n") |
| sig_norm: List[str] = [] |
| sig_pos: List[int] = [] |
| for i, line in enumerate(raw): |
| n = _norm(line) |
| if _significant(n): |
| sig_norm.append(n) |
| sig_pos.append(i) |
| return raw, sig_pos, sig_norm |
|
|
| @staticmethod |
| def _shingles(sig_norm: List[str]) -> List[int]: |
| if not sig_norm: |
| return [] |
| if len(sig_norm) < SHINGLE: |
| return [_digest(tuple(sig_norm))] |
| return [_digest(tuple(sig_norm[i:i + SHINGLE])) for i in range(len(sig_norm) - SHINGLE + 1)] |
|
|
| def _remember(self, sig_norm: List[str]) -> None: |
| if len(self._index) >= MAX_INDEX: |
| return |
| self._index.update(self._shingles(sig_norm)) |
|
|
| def _coverage(self, sig_norm: List[str]) -> List[bool]: |
| covered = [False] * len(sig_norm) |
| shingles = self._shingles(sig_norm) |
| width = SHINGLE if len(sig_norm) >= SHINGLE else len(sig_norm) |
| for start, value in enumerate(shingles): |
| if value in self._index: |
| for k in range(start, start + width): |
| covered[k] = True |
| return covered |
|
|
| |
| def seed_text(self, text: str) -> None: |
| """Index text WITHOUT deduplicating it (used for chat.jsonl cross-deduplication).""" |
| _raw, _pos, sig_norm = self._sig_lines(text) |
| self._remember(sig_norm) |
|
|
| def process(self, path: str, text: str) -> Tuple[Optional[str], Dict]: |
| """Return (document_with_path_header or None, info). info['status'] in clean|partial|duplicate.""" |
| raw, sig_pos, sig_norm = self._sig_lines(text) |
| header = f"[PATH: {path.lstrip('/')}]\n" |
| if not sig_norm: |
| return None, {"status": "duplicate", "reason": "no-significant-lines"} |
|
|
| covered = self._coverage(sig_norm) |
| total, hits = len(sig_norm), sum(covered) |
| overlap = hits / total |
|
|
| if hits == total: |
| self.stats["dropped_files"] += 1 |
| return None, {"status": "duplicate", "overlap": 1.0} |
| if overlap < OVERLAP_KEEP_BELOW or hits == 0: |
| self._remember(sig_norm) |
| self.stats["kept_intact"] += 1 |
| return header + text, {"status": "clean", "overlap": overlap} |
|
|
| |
| drop_sig: List[Tuple[int, int]] = [] |
| i = 0 |
| while i < total: |
| if not covered[i]: |
| i += 1 |
| continue |
| j = i |
| while j + 1 < total and covered[j + 1]: |
| j += 1 |
| before = 0 |
| k = i - 1 |
| while k >= 0 and not covered[k]: |
| before += 1 |
| k -= 1 |
| after = 0 |
| k = j + 1 |
| while k < total and not covered[k]: |
| after += 1 |
| k += 1 |
| if before + after < UNIQUE_CONTEXT_LINES: |
| drop_sig.append((i, j)) |
| i = j + 1 |
|
|
| if not drop_sig: |
| self._remember(sig_norm) |
| self.stats["kept_intact"] += 1 |
| return header + text, {"status": "clean", "overlap": overlap} |
|
|
| dropped_lines = set() |
| for a, b in drop_sig: |
| for pos in range(sig_pos[a], sig_pos[b] + 1): |
| dropped_lines.add(pos) |
| kept = [line for pos, line in enumerate(raw) if pos not in dropped_lines] |
| kept_text = "\n".join(kept).strip("\n") |
| _r, _p, kept_norm = self._sig_lines(kept_text) |
| if not kept_norm: |
| self.stats["dropped_files"] += 1 |
| return None, {"status": "duplicate", "overlap": overlap} |
| self._remember(sig_norm) |
| self.stats["partial_files"] += 1 |
| return header + kept_text, {"status": "partial", "overlap": overlap, |
| "excised_lines": len(dropped_lines)} |
