From 6fe684f8c0acc8ef2fb483a8532f60738120f4bd Mon Sep 17 00:00:00 2001 From: beam Date: Thu, 29 Oct 2026 18:46:59 +0000 Subject: [PATCH] beam: MinHash near-duplicate line detector with tests --- wizard-work/beam/minhash_dedup.py | 85 +++++++++++++++++++++++++++++++ wizard-work/beam/test_minhash.py | 69 +++++++++++++++++++++++++ 2 files changed, 154 insertions(+) create mode 100644 wizard-work/beam/minhash_dedup.py create mode 100644 wizard-work/beam/test_minhash.py diff --git a/wizard-work/beam/minhash_dedup.py b/wizard-work/beam/minhash_dedup.py new file mode 100644 index 0000000..286a207 --- /dev/null +++ b/wizard-work/beam/minhash_dedup.py @@ -0,0 +1,85 @@ +"""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() diff --git a/wizard-work/beam/test_minhash.py b/wizard-work/beam/test_minhash.py new file mode 100644 index 0000000..6195269 --- /dev/null +++ b/wizard-work/beam/test_minhash.py @@ -0,0 +1,69 @@ +import os, sys +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from minhash_dedup import * + +def test_shingles_basic(): + sh = shingles("abcde", 3) + assert " abc" in sh or "abc" in sh or len(sh) > 0 + assert len(shingles("hello world", 4)) > 0 + +def test_identical_lines_high_similarity(): + a = "the quick brown fox jumps over the lazy dog" + b = "the quick brown fox jumps over the lazy dog" + sa = minhash_signature(shingles(a)) + sb = minhash_signature(shingles(b)) + assert jaccard_est(sa, sb) == 1.0 + +def test_distinct_lines_low_similarity(): + a = "the quick brown fox jumps over the lazy dog" + b = "completely unrelated sentence about databases and servers" + sa = minhash_signature(shingles(a)) + sb = minhash_signature(shingles(b)) + assert jaccard_est(sa, sb) < 0.5 + +def test_find_duplicates_detects_near_dup(): + lines = [ + "error: connection refused to host 10.0.0.1 port 5432", + "error: connection refused to host 10.0.0.2 port 5432", + "totally different line", + ] + dups = find_duplicates(lines, threshold=0.3) + assert any((i,j)==(0,1) for i,j,_ in dups), dups + +def test_candidate_pairs_self(): + sigs = [minhash_signature(shingles("abc"*10)), minhash_signature(shingles("abc"*10))] + pairs = candidate_pairs(sigs, bands=8) + assert (0,1) in pairs + +def test_empty_shingle_set(): + sig = minhash_signature(set()) + assert all(x == 2**64-1 for x in sig) + +def test_jaccard_length_mismatch(): + try: + jaccard_est([1,2],[1,2,3]) + raise AssertionError("expected ValueError") + except ValueError: + pass + +def test_threshold_filter(): + lines = ["aaaa", "bbbb", "cccc"] + dups = find_duplicates(lines, threshold=0.99) + # distinct short strings may or may not collide; just ensure no crash and type + assert isinstance(dups, list) + +if __name__ == "__main__": + # minimal runner without pytest + import traceback + tests = [v for k,v in globals().items() if k.startswith("test_")] + fails = 0 + for t in tests: + try: + t() + print("PASS", t.__name__) + except Exception as e: + fails += 1 + print("FAIL", t.__name__, repr(e)) + traceback.print_exc() + print("FAILS:", fails) + sys.exit(1 if fails else 0)