beam: MinHash near-duplicate line detector with tests

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 <file> [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()

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)