beam: near-duplicate line detector via MinHash + union-find

wizard-work/beam/near_dup.py

@@ -0,0 +1,100 @@
+"""near_dup.py — detect near-duplicate lines in a text file using MinHash.
+
+Algorithm: character 3-shingles, k independent hash functions (random
+coefficients), MinHash signature per line, Jaccard estimate via signature
+agreement, union-find clustering above threshold. O(n * k) time.
+"""
+import argparse
+import hashlib
+import random
+import sys
+from collections import defaultdict
+
+
+def shingles(text: str, k: int = 3) -> set:
+    t = text.lower().strip()
+    if len(t) < k:
+        return {t} if t else set()
+    return {t[i:i + k] for i in range(len(t) - k + 1)}
+
+
+def _hashes(k: int, seed: int = 0x9E3779B9):
+    rng = random.Random(seed)
+    return [(rng.randrange(1 << 31), rng.randrange(1 << 31)) for _ in range(k)]
+
+
+def minhash_signature(sh: set, num_hashes: int = 64, seed: int = 0x9E3779B9) -> tuple:
+    if not sh:
+        return tuple([1 << 63] * num_hashes)
+    hashes = _hashes(num_hashes, seed)
+    sig = [1 << 63] * num_hashes
+    for s in sh:
+        h = int(hashlib.md5(s.encode()).hexdigest(), 16)
+        for i, (a, b) in enumerate(hashes):
+            val = (a * h + b) & ((1 << 63) - 1)
+            if val < sig[i]:
+                sig[i] = val
+    return tuple(sig)
+
+
+def jaccard_est(sig_a, sig_b) -> float:
+    if len(sig_a) != len(sig_b):
+        raise ValueError("signature length mismatch")
+    if not sig_a:
+        return 0.0
+    eq = sum(1 for a, b in zip(sig_a, sig_b) if a == b)
+    return eq / len(sig_a)
+
+
+class UnionFind:
+    def __init__(self, n):
+        self.parent = list(range(n))
+        self.rank = [0] * n
+
+    def find(self, x):
+        while self.parent[x] != x:
+            self.parent[x] = self.parent[self.parent[x]]
+            x = self.parent[x]
+        return x
+
+    def union(self, a, b):
+        ra, rb = self.find(a), self.find(b)
+        if ra == rb:
+            return
+        if self.rank[ra] < self.rank[rb]:
+            ra, rb = rb, ra
+        self.parent[rb] = ra
+        if self.rank[ra] == self.rank[rb]:
+            self.rank[ra] += 1
+
+
+def clusters(lines, threshold=0.8, num_hashes=64, seed=0x9E3779B9):
+    sigs = [minhash_signature(shingles(l), num_hashes, seed) for l in lines]
+    uf = UnionFind(len(lines))
+    # O(n^2) pairwise; fine for moderate files. Could LSH-band for scale.
+    for i in range(len(lines)):
+        for j in range(i + 1, len(lines)):
+            if jaccard_est(sigs[i], sigs[j]) >= threshold:
+                uf.union(i, j)
+    groups = defaultdict(list)
+    for idx in range(len(lines)):
+        groups[uf.find(idx)].append(idx)
+    return [g for g in groups.values() if len(g) > 1]
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("path")
+    ap.add_argument("--threshold", type=float, default=0.8)
+    ap.add_argument("--hashes", type=int, default=64)
+    args = ap.parse_args()
+    with open(args.path, encoding="utf-8") as f:
+        lines = [ln.rstrip("\n") for ln in f]
+    for g in clusters(lines, args.threshold, args.hashes):
+        print("GROUP:")
+        for idx in g:
+            print(f"  {idx + 1}: {lines[idx]}")
+
+
+if __name__ == "__main__":
+    main()

wizard-work/beam/test_near_dup.py

@@ -0,0 +1,38 @@
+import os, sys, tempfile
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from near_dup import shingles, minhash_signature, jaccard_est, clusters
+
+def test_shingles_basic():
+    s = shingles("abcde", 3)
+    assert s == {"abc", "bcd", "cde"}, s
+
+def test_identical_lines_cluster():
+    lines = ["the quick brown fox jumps", "the quick brown fox jumps over", "totally different sentence here"]
+    # first two should cluster at 0.5 threshold
+    g = clusters(lines, threshold=0.5)
+    assert any(0 in grp and 1 in grp for grp in g), g
+
+def test_exact_dup_cluster():
+    lines = ["hello world", "hello world", "goodbye"]
+    g = clusters(lines, threshold=0.99)
+    assert any(0 in grp and 1 in grp for grp in g), g
+    assert not any(2 in grp for grp in g if 0 in grp)
+
+def test_no_clusters_distinct():
+    lines = ["alpha beta gamma", "one two three four", "x y z"]
+    g = clusters(lines, threshold=0.9)
+    assert g == [], g
+
+def test_jaccard_identical_sigs():
+    s = minhash_signature(shingles("abcdef"))
+    assert jaccard_est(s, s) == 1.0
+
+def test_empty_line():
+    s = shingles("")
+    assert s == set() or s == {""}, s
+
+if __name__ == "__main__":
+    for name in [n for n in globals() if n.startswith("test_")]:
+        globals()[name]()
+        print("PASS", name)
+    print("ALL OK")