File size: 6,742 Bytes
9d614f1 | 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 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | """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 # consecutive significant lines that make up one "context window"
OVERLAP_KEEP_BELOW = 0.85 # < 85 % full-block overlap -> keep the file intact
UNIQUE_CONTEXT_LINES = 3 # >= 3 unique surrounding lines -> keep the run intact
MAX_INDEX = 6_000_000 # memory guard (shingle hashes are 8 bytes each)
_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}
# ---------------------------------------------------------------- helpers
@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
# ------------------------------------------------------------------ public
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: # 100 % duplicate -> drop the file
self.stats["dropped_files"] += 1
return None, {"status": "duplicate", "overlap": 1.0}
if overlap < OVERLAP_KEEP_BELOW or hits == 0: # enough unique context -> keep intact
self._remember(sig_norm)
self.stats["kept_intact"] += 1
return header + text, {"status": "clean", "overlap": overlap}
# >= 85 % overlap: look at each covered run together with its surrounding context
drop_sig: List[Tuple[int, int]] = [] # inclusive ranges in significant-line space
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: # pure duplicate without unique context
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: # excise the whole span, neutral lines inside included
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)} |