Find duplicate records in a file

"""
Find duplicate records by comparing spelling, not strings.

Rung N1. The blocking key of N0 decides in advance which pairs deserve a look,
and everything it puts in two different groups stays invisible. This rung
drops the key: every record becomes a vector of its character n-grams, and
neighbours are looked up by cosine.

Why character n-grams rather than words: they survive a typo, a swapped word
order and a truncated field, because a misspelt word still shares most of its
three-letter slices with the correct one. Nothing here is learnt from a corpus
and there is no model to train. It is a weighting scheme and a distance.

The price is the search itself: no key means every pair of the file is
scored, so the cost grows with the square of the file. The latency class shown
for this rung is measured on the nominal file of its tests; twenty thousand
records are a matter of seconds, not of milliseconds.
"""

import unicodedata

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.neighbors import NearestNeighbors


def normalise(text: str | None) -> str:
    """Lower case, strip accents, invisible characters and punctuation, collapse spaces."""
    decomposed = unicodedata.normalize("NFKD", (text or "").lower())
    # A zero-width space is a format character: dropped, it does not split a word.
    letters = "".join(c for c in decomposed if not unicodedata.combining(c) and unicodedata.category(c) != "Cf")
    return " ".join("".join(c if c.isalnum() else " " for c in letters).split())


def record_text(record: dict) -> str:
    """
    One comparable string per record, columns in a stable order.

    Sorted by column name, so that two exports of the same data give the same
    text whatever order their columns come in: n-grams taken across a column
    boundary would otherwise differ. An empty cell adds nothing.
    """
    return normalise(" ".join(str(record[key]) for key in sorted(record) if record[key] is not None))


def find_duplicates(records: list[dict], threshold: float = 0.6) -> list[tuple]:
    """
    Return the pairs `(i, j, score)` that look like the same record.

    `char_wb` keeps n-grams inside word boundaries, so "dupont" and "dupond"
    share most of their slices while "dupont paris" borrows nothing from the
    join between the two words.
    """
    texts = [record_text(record) for record in records]
    if sum(1 for text in texts if text) < 2:
        return []  # fewer than two records with something to compare

    vectors = TfidfVectorizer(analyzer="char_wb", ngram_range=(2, 4)).fit_transform(texts)

    # Every neighbour within the radius, for every record. The radius takes a
    # hair more than 1 - threshold: a cosine computed in floating point can land
    # just under 1.0 for two identical records.
    index = NearestNeighbors(metric="cosine").fit(vectors)
    distances, neighbours = index.radius_neighbors(vectors, radius=1 - threshold + 1e-9)

    pairs = []
    for i, (row_distances, row_neighbours) in enumerate(zip(distances, neighbours)):
        for distance, j in zip(row_distances, row_neighbours):
            if i < j:
                pairs.append((i, int(j), round(1 - float(distance), 3)))
    return sorted(pairs, key=lambda pair: (-pair[2], pair[0], pair[1]))

/**
 * Find duplicate records by comparing spelling, not strings.
 *
 * Rung N1. The blocking key of N0 decides in advance which pairs deserve a
 * look, and everything it puts in two different groups stays invisible. This
 * rung drops the key: every record becomes a vector of its character n-grams,
 * and pairs are ranked by cosine.
 *
 * Why character n-grams rather than words: they survive a typo, a swapped
 * word order and a truncated field, because a misspelt word still shares most
 * of its three-letter slices with the correct one. Nothing here is learnt
 * from a corpus and there is no model to train. It is a weighting scheme and
 * a distance, written here with no dependency.
 *
 * The price is the search itself: no key means every pair of the file is
 * scored, so the cost grows with the square of the file. The latency class
 * shown for this rung is measured on the nominal file of its tests; twenty
 * thousand records are a matter of seconds, not of milliseconds.
 */

const NGRAM_SIZES = [2, 3, 4];

/** Lower case, strip accents, invisible characters and punctuation, collapse spaces. */
export function normalise(text) {
  // A zero-width space is a format character: dropped, it does not split a word.
  return String(text ?? '').toLowerCase().normalize('NFKD').replace(/\p{Cf}/gu, '')
    .replace(/\p{Diacritic}/gu, '').replace(/[^\p{L}\p{N}]+/gu, ' ').trim();
}

/**
 * One comparable string per record, columns in a stable order.
 *
 * Sorted by column name, so that two exports of the same data give the same
 * text whatever order their columns come in: n-grams taken across a column
 * boundary would otherwise differ. An empty cell adds nothing.
 */
export function recordText(record) {
  const keys = Object.keys(record).sort();
  return normalise(keys.filter((key) => record[key] !== null && record[key] !== undefined)
    .map((key) => String(record[key])).join(' '));
}

/**
 * Character n-grams, counted, taken inside word boundaries.
 *
 * Padding each word with spaces is what keeps "dupont" from borrowing an
 * n-gram from the join with the word that follows it.
 */
function ngrams(text) {
  const counts = new Map();
  for (const word of text.split(' ').filter(Boolean)) {
    const padded = ` ${word} `;
    for (const size of NGRAM_SIZES) {
      for (let i = 0; i + size <= padded.length; i += 1) {
        const gram = padded.slice(i, i + size);
        counts.set(gram, (counts.get(gram) ?? 0) + 1);
      }
    }
  }
  return counts;
}

/**
 * TF-IDF weights, L2 normalised, one sparse vector per record.
 *
 * The inverse document frequency is what stops "paris", present in every
 * record of a Paris file, from making every pair look alike.
 */
function vectorise(texts) {
  const counted = texts.map(ngrams);
  const documentFrequency = new Map();
  for (const counts of counted) {
    for (const gram of counts.keys()) {
      documentFrequency.set(gram, (documentFrequency.get(gram) ?? 0) + 1);
    }
  }
  return counted.map((counts) => {
    const vector = new Map();
    for (const [gram, count] of counts) {
      const idf = Math.log((1 + texts.length) / (1 + documentFrequency.get(gram))) + 1;
      vector.set(gram, count * idf);
    }
    const norm = Math.hypot(...vector.values());
    for (const [gram, weight] of vector) vector.set(gram, norm ? weight / norm : 0);
    return vector;
  });
}

/**
 * Return the pairs `[i, j, score]` that look like the same record.
 *
 * The threshold is yours to set: towards 1.0 if merging two different
 * customers is the worse outcome, towards 0.0 if missing a duplicate is.
 */
export function findDuplicates(records, threshold = 0.6) {
  const vectors = vectorise(records.map(recordText));
  const pairs = [];
  for (let i = 0; i < vectors.length; i += 1) {
    for (let j = i + 1; j < vectors.length; j += 1) {
      // Walking the shorter vector: only the grams the two records share
      // contribute anything to a cosine.
      const [small, large] = vectors[i].size <= vectors[j].size ? [vectors[i], vectors[j]] : [vectors[j], vectors[i]];
      let score = 0;
      for (const [gram, weight] of small) score += weight * (large.get(gram) ?? 0);
      // A hair of tolerance: in floating point, two identical records can score just under 1.0.
      if (score >= threshold - 1e-9) pairs.push([i, j, Math.round(score * 1000) / 1000]);
    }
  }
  return pairs.sort((a, b) => b[2] - a[2] || a[0] - b[0] || a[1] - b[1]);
}

"""
Find duplicate records with a self-hosted sentence encoder.

Rung N2. N1 compares spellings; two records that say the same thing in
different words share no character n-grams and stay invisible. A sentence
encoder maps each record to a dense vector, the mean of the vectors of its
tokens, and records are compared on those vectors rather than on shared
letters.

Two prices, and the second is the one that grows. The fixed one is the
deployment: weights to download and load, and a process to keep warm. The
other is the comparison, and it is quadratic — every pair of the file is
scored, a thousand records make half a million dot products of three hundred
and eighty-four numbers, ten thousand make fifty million. Past a few tens of
thousands of records, block first with the key of rung N0, or reach for an
approximate index; this snippet does neither, on purpose, because it shows the
distance and not the search.

The encoder is a parameter with a real default, so the reader sees the loading
code while the test injects a local double.
"""

from __future__ import annotations

import math
import unicodedata

import numpy as np

# A multilingual model, because a customer file is rarely in one language.
MODEL_NAME = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"


class EncodingFailed(RuntimeError):
    """The encoder could not be run, or returned something unusable."""


def normalise(text: str | None) -> str:
    """Lower case, strip accents, invisible characters and punctuation, collapse spaces."""
    decomposed = unicodedata.normalize("NFKD", (text or "").lower())
    # A zero-width space is a format character: dropped, it does not split a word.
    letters = "".join(c for c in decomposed if not unicodedata.combining(c) and unicodedata.category(c) != "Cf")
    return " ".join("".join(c if c.isalnum() else " " for c in letters).split())


def record_text(record: dict) -> str:
    """
    One comparable string per record, columns in a stable order.

    Sorted by column name, so that two exports of the same data give the same
    text whatever order their columns come in: n-grams taken across a column
    boundary would otherwise differ. An empty cell adds nothing.
    """
    return normalise(" ".join(str(record[key]) for key in sorted(record) if record[key] is not None))


def unit(vector) -> list[float]:
    """Normalise once, so that a cosine is a dot product afterwards."""
    values = [float(v) for v in vector]
    norm = sum(v * v for v in values) ** 0.5
    return [v / norm for v in values] if norm else values


def find_duplicates(records: list[dict], encoder=None, threshold: float = 0.75) -> list[tuple]:
    """Return the pairs `(i, j, score)` that look like the same record."""
    texts = [record_text(record) for record in records]
    if len(texts) < 2:
        return []

    if encoder is None:  # pragma: no cover - loads several hundred megabytes
        from sentence_transformers import SentenceTransformer

        encoder = SentenceTransformer(MODEL_NAME)

    # One batched call. Encoding record by record wastes most of the machine.
    try:
        vectors = [unit(v) for v in encoder.encode(texts)]
    except Exception as error:  # noqa: BLE001 - a model failure is not the caller's fault
        raise EncodingFailed(str(error)) from error
    if len(vectors) != len(texts):
        raise EncodingFailed(f"{len(vectors)} vectors returned for {len(texts)} records")
    if len({len(v) for v in vectors}) != 1 or not all(math.isfinite(x) for v in vectors for x in v):
        raise EncodingFailed("vectors of different sizes, or holding something other than numbers")

    # Every pair at once, as one matrix product. Quadratic all the same — the
    # matrix has one cell per pair — but not at the speed of a Python loop.
    matrix = np.array(vectors, dtype=float)
    scores = matrix @ matrix.T
    above = np.triu_indices(len(vectors), k=1)
    # A hair of tolerance: in floating point, two identical records can score just under 1.0.
    kept = scores[above] >= threshold - 1e-9
    pairs = [
        (int(i), int(j), round(float(scores[i, j]), 3))
        for i, j in zip(above[0][kept], above[1][kept])
    ]
    return sorted(pairs, key=lambda pair: (-pair[2], pair[0], pair[1]))

/**
 * Find duplicate records with a self-hosted sentence encoder.
 *
 * Rung N2. N1 compares spellings; two records that say the same thing in
 * different words share no character n-grams and stay invisible. A sentence
 * encoder maps each record to a dense vector, the mean of the vectors of its
 * tokens, and records are compared on those vectors rather than on shared
 * letters.
 *
 * Two prices, and the second is the one that grows. The fixed one is the
 * deployment: weights to download and load, and a process to keep warm. The
 * other is the comparison, and it is quadratic — every pair of the file is
 * scored, a thousand records make half a million dot products of three hundred
 * and eighty-four numbers, ten thousand make fifty million. Past a few tens of
 * thousands of records, block first with the key of rung N0, or reach for an
 * approximate index; this snippet does neither, on purpose, because it shows
 * the distance and not the search.
 *
 * The encoder is a parameter with a real default, so the reader sees the
 * loading code while the test injects a local double.
 */

// A multilingual model, because a customer file is rarely in one language.
export const MODEL_NAME = 'Xenova/paraphrase-multilingual-MiniLM-L12-v2';

export class EncodingFailed extends Error {}

/** Lower case, strip accents, invisible characters and punctuation, collapse spaces. */
export function normalise(text) {
  // A zero-width space is a format character: dropped, it does not split a word.
  return String(text ?? '').toLowerCase().normalize('NFKD').replace(/\p{Cf}/gu, '')
    .replace(/\p{Diacritic}/gu, '').replace(/[^\p{L}\p{N}]+/gu, ' ').trim();
}

/**
 * One comparable string per record, columns in a stable order.
 *
 * Sorted by column name, so that two exports of the same data give the same
 * text whatever order their columns come in: n-grams taken across a column
 * boundary would otherwise differ. An empty cell adds nothing.
 */
export function recordText(record) {
  const keys = Object.keys(record).sort();
  return normalise(keys.filter((key) => record[key] !== null && record[key] !== undefined)
    .map((key) => String(record[key])).join(' '));
}

/** Normalise once, so that a cosine is a dot product afterwards. */
export function unit(vector) {
  const norm = Math.hypot(...vector);
  return norm ? vector.map((v) => v / norm) : [...vector];
}

/**
 * Return the pairs `[i, j, score]` that look like the same record.
 *
 * @param {object[]} records
 * @param {object} options
 * @param {{encode: Function}} [options.encoder] injected so this can be
 *   tested without downloading a model; defaults to a real local encoder
 * @param {number} [options.threshold]
 */
export async function findDuplicates(records, { encoder, threshold = 0.75 } = {}) {
  const texts = records.map(recordText);
  if (texts.length < 2) return [];

  if (!encoder) {
    // Downloads and loads the weights, so it is never reached in the tests.
    const { pipeline } = await import('@huggingface/transformers');
    const extract = await pipeline('feature-extraction', MODEL_NAME);
    encoder = { encode: async (batch) => (await extract(batch, { pooling: 'mean' })).tolist() };
  }

  // One batched call. Encoding record by record wastes most of the machine.
  let vectors;
  try {
    vectors = (await encoder.encode(texts)).map(unit);
  } catch (error) {
    throw new EncodingFailed(String(error));
  }
  if (vectors.length !== texts.length) {
    throw new EncodingFailed(`${vectors.length} vectors returned for ${texts.length} records`);
  }
  if (new Set(vectors.map((v) => v.length)).size !== 1 || !vectors.every((v) => v.every(Number.isFinite))) {
    throw new EncodingFailed('vectors of different sizes, or holding something other than numbers');
  }

  // Every pair, in one flat array rather than an array of arrays: quadratic
  // all the same — one dot product per pair — but without a closure per cell.
  const width = vectors[0].length;
  const flat = new Float64Array(vectors.length * width);
  vectors.forEach((vector, i) => flat.set(vector, i * width));

  const pairs = [];
  for (let i = 0; i < vectors.length; i += 1) {
    for (let j = i + 1; j < vectors.length; j += 1) {
      let score = 0;
      for (let d = 0; d < width; d += 1) score += flat[i * width + d] * flat[j * width + d];
      // A hair of tolerance: in floating point, two identical records can score just under 1.0.
      if (score >= threshold - 1e-9) pairs.push([i, j, Math.round(score * 1000) / 1000]);
    }
  }
  return pairs.sort((a, b) => b[2] - a[2] || a[0] - b[0] || a[1] - b[1]);
}