101 lines
3.0 KiB
Python
101 lines
3.0 KiB
Python
"""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()
|