"""MinHash-based near-duplicate line detector. Implements character n-gram shingling, k-hash MinHash signatures, and banding LSH for candidate pairs. Pure stdlib. Deterministic given seed. """ from __future__ import annotations import hashlib, random, sys def shingles(s: str, n: int = 5) -> set: s = " " + s.lower() + " " if len(s) < n: return {s} return {s[i:i+n] for i in range(len(s)-n+1)} def _hashes_for_shingle(sh: str, num_hashes: int) -> list: # Derive num_hashes independent hash values from one shingle via # salted SHA-256 (deterministic, no numpy needed). out = [] for i in range(num_hashes): h = hashlib.sha256((str(i) + "|" + sh).encode()).digest() out.append(int.from_bytes(h[:8], "big")) return out def minhash_signature(shingle_set: set, num_hashes: int = 64) -> list: if not shingle_set: return [2**64 - 1] * num_hashes sig = [2**64 - 1] * num_hashes for sh in shingle_set: hs = _hashes_for_shingle(sh, num_hashes) for i in range(num_hashes): if hs[i] < sig[i]: sig[i] = hs[i] return sig def jaccard_est(sig_a, sig_b) -> float: if not sig_a or len(sig_a) != len(sig_b): raise ValueError("signature length mismatch") eq = sum(1 for a,b in zip(sig_a, sig_b) if a == b) return eq / len(sig_a) def candidate_pairs(signatures: list, bands: int = 8) -> set: # Banding LSH: split signature into bands; rows per band = len(sig)/bands. if not signatures: return set() n = len(signatures[0]) if n % bands != 0: raise ValueError("num_hashes must be divisible by bands") rows = n // bands pairs = set() for b in range(bands): buckets = {} for idx, sig in enumerate(signatures): key = tuple(sig[b*rows:(b+1)*rows]) buckets.setdefault(key, []).append(idx) for grp in buckets.values(): for i in range(len(grp)): for j in range(i+1, len(grp)): pairs.add((min(grp[i], grp[j]), max(grp[i], grp[j]))) return pairs def find_duplicates(lines, threshold=0.5, n=5, num_hashes=64, bands=8): sigs = [minhash_signature(shingles(ln, n), num_hashes) for ln in lines] cands = candidate_pairs(sigs, bands) out = [] for (i,j) in sorted(cands): sim = jaccard_est(sigs[i], sigs[j]) if sim >= threshold: out.append((i, j, sim)) return out def main(): if len(sys.argv) < 2: print("usage: minhash_dedup.py [threshold]", file=sys.stderr) sys.exit(2) path = sys.argv[1] thr = float(sys.argv[2]) if len(sys.argv) > 2 else 0.5 with open(path) as f: lines = [ln.rstrip("\n") for ln in f] dups = find_duplicates(lines, threshold=thr) for i,j,sim in dups: print(f"{i}\t{j}\t{sim:.3f}\t{lines[i][:60]!r}") print(f"# {len(dups)} candidate duplicate pairs", file=sys.stderr) if __name__ == "__main__": main()