spb/wpai0002_uqo

1"""Exact and near-duplicate detection (MinHash LSH over word 5-gram shingles) for documents and passages."""2from __future__ import annotations34import re5from dataclasses import dataclass, field67from datasketch import MinHash, MinHashLSH89from htr.utils.hashing import normalize_for_hash, sha256_text1011_TOKEN = re.compile(r"\w+")121314def shingles(text: str, k: int = 5) -> set[str]:15    toks = _TOKEN.findall(normalize_for_hash(text))16    if len(toks) < k:17        return {" ".join(toks)} if toks else set()18    return {" ".join(toks[i : i + k]) for i in range(len(toks) - k + 1)}192021def minhash(text: str, num_perm: int = 128, k: int = 5) -> MinHash:22    m = MinHash(num_perm=num_perm)23    for s in shingles(text, k):24        m.update(s.encode("utf-8"))25    return m262728@dataclass29class DedupResult:30    exact_dupes: dict[str, str] = field(default_factory=dict)  # id -> canonical id31    near_dupes: dict[str, str] = field(default_factory=dict)  # id -> canonical id32    clusters: dict[str, str] = field(default_factory=dict)  # id -> cluster id (union of exact + near)333435class Deduper:36    """Incremental deduper. ``add(id, text)`` returns the canonical id (itself if new)."""3738    def __init__(self, threshold: float = 0.8, num_perm: int = 128, k: int = 5):39        self.lsh = MinHashLSH(threshold=threshold, num_perm=num_perm)40        self.num_perm = num_perm41        self.k = k42        self.exact: dict[str, str] = {}43        self.result = DedupResult()44        self._mh: dict[str, MinHash] = {}4546    def add(self, item_id: str, text: str) -> str:47        h = sha256_text(normalize_for_hash(text))48        if h in self.exact:49            canon = self.exact[h]50            self.result.exact_dupes[item_id] = canon51            self.result.clusters[item_id] = self.result.clusters.get(canon, canon)52            return canon53        self.exact[h] = item_id54        m = minhash(text, self.num_perm, self.k)55        cands = self.lsh.query(m)56        if cands:57            canon = sorted(cands)[0]58            self.result.near_dupes[item_id] = canon59            self.result.clusters[item_id] = self.result.clusters.get(canon, canon)60        else:61            self.result.clusters[item_id] = item_id62        self.lsh.insert(item_id, m)63        self._mh[item_id] = m64        return self.result.clusters[item_id]6566    def jaccard(self, a: str, b: str) -> float:67        return float(self._mh[a].jaccard(self._mh[b]))68