86 lines
2.9 KiB
Python
86 lines
2.9 KiB
Python
"""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()
|