84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
# Vendored from thoth_sqldb2 (thoth-dbmanager 0.7.4) — lsh/core.py
|
|
# Copyright 2025 Marco Pancotti — Apache License 2.0
|
|
# Modifiche locali documentate in VENDORED.md.
|
|
"""Core LSH (MinHash) per la ricerca di valori simili nei campi del database."""
|
|
|
|
import logging
|
|
|
|
from datasketch import MinHash, MinHashLSH
|
|
from tqdm import tqdm
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def create_minhash(signature_size: int, string: str, n_gram: int) -> MinHash:
|
|
m = MinHash(num_perm=signature_size)
|
|
for d in [string[i : i + n_gram] for i in range(len(string) - n_gram + 1)]:
|
|
m.update(d.encode("utf8"))
|
|
return m
|
|
|
|
|
|
# Token che, se presenti nel nome di una colonna, la marcano come "name-like":
|
|
# nomi propri di persone/enti, i cui valori sono utili al value-matching e quindi
|
|
# mai da escludere dall'indice LSH. Bilingue EN/IT (modifica locale, vedi VENDORED.md).
|
|
# Per substring coprono le forme flesse: "nome" -> cognome/soprannome,
|
|
# "name" -> surname/username, "denominazion" -> denominazione, ecc.
|
|
NAME_LIKE_TOKENS: tuple[str, ...] = (
|
|
"name",
|
|
"nome",
|
|
"nominativ",
|
|
"denominazion",
|
|
"ragione_sociale",
|
|
)
|
|
|
|
|
|
def skip_column(
|
|
column_name: str,
|
|
column_values: list[str],
|
|
max_total_chars: int = 50000,
|
|
max_avg_length: int = 20,
|
|
name_tokens: tuple[str, ...] = NAME_LIKE_TOKENS,
|
|
) -> bool:
|
|
lowered = column_name.lower()
|
|
if any(token in lowered for token in name_tokens):
|
|
return False
|
|
sum_of_lengths = sum(len(value) for value in column_values)
|
|
average_length = sum_of_lengths / len(column_values)
|
|
return (sum_of_lengths > max_total_chars) and (average_length > max_avg_length)
|
|
|
|
|
|
def jaccard_similarity(m1: MinHash, m2: MinHash) -> float:
|
|
return m1.jaccard(m2)
|
|
|
|
|
|
def create_lsh_index(
|
|
unique_values: dict[str, dict[str, list[str]]],
|
|
signature_size: int,
|
|
n_gram: int,
|
|
threshold: float,
|
|
verbose: bool = True,
|
|
) -> tuple[MinHashLSH, dict[str, tuple[MinHash, str, str, str]]]:
|
|
lsh = MinHashLSH(threshold=threshold, num_perm=signature_size)
|
|
minhashes: dict[str, tuple[MinHash, str, str, str]] = {}
|
|
total = sum(
|
|
len(column_values)
|
|
for table_values in unique_values.values()
|
|
for column_values in table_values.values()
|
|
)
|
|
logger.info("Total unique values: %s", total)
|
|
progress_bar = tqdm(total=total, desc="Creating LSH") if verbose else None
|
|
|
|
for table_name, table_values in unique_values.items():
|
|
for column_name, column_values in table_values.items():
|
|
for idx, value in enumerate(column_values):
|
|
minhash = create_minhash(signature_size, value, n_gram)
|
|
minhash_key = f"{table_name}_{column_name}_{idx}"
|
|
minhashes[minhash_key] = (minhash, table_name, column_name, value)
|
|
lsh.insert(minhash_key, minhash)
|
|
if progress_bar:
|
|
progress_bar.update(1)
|
|
|
|
if progress_bar:
|
|
progress_bar.close()
|
|
return lsh, minhashes
|