Find duplicate records in a file

"""
Find duplicate records: normalise, block, then compare inside a block only.

Rung N0. Deterministic, standard library only, and the only rung here that
stays fast when the file grows, because it never compares every pair.

Two things make this work.

First, normalisation removes what a human ignores when reading a name: case,
accents, punctuation, double spaces. Two spellings that only differ by those
become the same string, and cost nothing to detect.

Second, blocking. Comparing every pair of ten thousand records is fifty
million comparisons. Grouping them by a cheap key first, and comparing only
inside a group, turns that into a few thousand. The whole cost of the rung is
in the key you pick, and so is its blind spot.
"""

import unicodedata


def normalise(text: str) -> str:
    """Lower case, strip accents and punctuation, collapse spaces."""
    decomposed = unicodedata.normalize("NFKD", text.lower())
    letters = "".join(c for c in decomposed if not unicodedata.combining(c))
    return " ".join("".join(c if c.isalnum() else " " for c in letters).split())


def record_text(record: dict) -> str:
    """One comparable string per record."""
    return normalise(" ".join(str(value) for value in record.values()))


def blocking_key(record: dict) -> str:
    """
    Two records that do not share this key are never compared.

    Three letters of the family name and the postcode: short enough to group
    real duplicates together, specific enough to keep the groups small.
    """
    words = normalise(record["name"]).split()
    family_name = words[-1] if words else ""
    return f"{family_name[:3]}:{normalise(str(record['postcode']))}"


def edit_distance(a: str, b: str) -> int:
    """Levenshtein distance, two rows at a time rather than a full matrix."""
    previous = list(range(len(b) + 1))
    for i, char_a in enumerate(a, start=1):
        current = [i]
        for j, char_b in enumerate(b, start=1):
            cost = 0 if char_a == char_b else 1
            current.append(min(previous[j] + 1, current[j - 1] + 1, previous[j - 1] + cost))
        previous = current
    return previous[-1]


def similarity(a: str, b: str) -> float:
    """1.0 for identical strings, 0.0 for strings sharing nothing."""
    longest = max(len(a), len(b))
    return 1.0 if longest == 0 else 1.0 - edit_distance(a, b) / longest


def find_duplicates(records: list[dict], threshold: float = 0.85) -> list[tuple]:
    """
    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.
    """
    blocks: dict[str, list[int]] = {}
    for index, record in enumerate(records):
        blocks.setdefault(blocking_key(record), []).append(index)

    pairs = []
    for indexes in blocks.values():
        for position, i in enumerate(indexes):
            for j in indexes[position + 1:]:
                score = similarity(record_text(records[i]), record_text(records[j]))
                if score >= threshold:
                    pairs.append((i, j, round(score, 3)))
    return sorted(pairs, key=lambda pair: (-pair[2], pair[0], pair[1]))

/**
 * Find duplicate records: normalise, block, then compare inside a block only.
 *
 * Rung N0. Deterministic, no dependency, and the only rung here that stays
 * fast when the file grows, because it never compares every pair.
 *
 * Two things make this work.
 *
 * First, normalisation removes what a human ignores when reading a name:
 * case, accents, punctuation, double spaces. Two spellings that only differ
 * by those become the same string, and cost nothing to detect.
 *
 * Second, blocking. Comparing every pair of ten thousand records is fifty
 * million comparisons. Grouping them by a cheap key first, and comparing only
 * inside a group, turns that into a few thousand. The whole cost of the rung
 * is in the key you pick, and so is its blind spot.
 */

/** Lower case, strip accents and punctuation, collapse spaces. */
export function normalise(text) {
  return String(text)
    .toLowerCase()
    .normalize('NFKD')
    .replace(/\p{Diacritic}/gu, '')
    .replace(/[^\p{L}\p{N}]+/gu, ' ')
    .trim();
}

/** One comparable string per record. */
export function recordText(record) {
  return normalise(Object.values(record).join(' '));
}

/**
 * Two records that do not share this key are never compared.
 *
 * Three letters of the family name and the postcode: short enough to group
 * real duplicates together, specific enough to keep the groups small.
 */
export function blockingKey(record) {
  const words = normalise(record.name).split(' ').filter(Boolean);
  const familyName = words.at(-1) ?? '';
  return `${familyName.slice(0, 3)}:${normalise(record.postcode)}`;
}

/** Levenshtein distance, two rows at a time rather than a full matrix. */
export function editDistance(a, b) {
  let previous = Array.from({ length: b.length + 1 }, (_, i) => i);
  for (let i = 1; i <= a.length; i += 1) {
    const current = [i];
    for (let j = 1; j <= b.length; j += 1) {
      const cost = a[i - 1] === b[j - 1] ? 0 : 1;
      current.push(Math.min(previous[j] + 1, current[j - 1] + 1, previous[j - 1] + cost));
    }
    previous = current;
  }
  return previous.at(-1);
}

/** 1.0 for identical strings, 0.0 for strings sharing nothing. */
export function similarity(a, b) {
  const longest = Math.max(a.length, b.length);
  return longest === 0 ? 1 : 1 - editDistance(a, b) / longest;
}

/**
 * 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.85) {
  const blocks = new Map();
  records.forEach((record, index) => {
    const key = blockingKey(record);
    if (!blocks.has(key)) blocks.set(key, []);
    blocks.get(key).push(index);
  });

  const pairs = [];
  for (const indexes of blocks.values()) {
    for (let a = 0; a < indexes.length; a += 1) {
      for (let b = a + 1; b < indexes.length; b += 1) {
        const [i, j] = [indexes[a], indexes[b]];
        const score = similarity(recordText(records[i]), recordText(records[j]));
        if (score >= threshold) pairs.push([i, j, Math.round(score * 1000) / 1000]);
      }
    }
  }
  return pairs.sort((x, y) => y[2] - x[2] || x[0] - y[0] || x[1] - y[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
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, in the worst case, every pair.
A neighbour index earns its place well before the file gets large.
"""

import unicodedata

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


def normalise(text: str) -> str:
    """Lower case, strip accents and punctuation, collapse spaces."""
    decomposed = unicodedata.normalize("NFKD", text.lower())
    letters = "".join(c for c in decomposed if not unicodedata.combining(c))
    return " ".join("".join(c if c.isalnum() else " " for c in letters).split())


def record_text(record: dict) -> str:
    """One comparable string per record."""
    return normalise(" ".join(str(value) for value in record.values()))


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 len(texts) < 2:
        return []

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

    # A neighbour index, not a full similarity matrix: the matrix is the thing
    # that stops fitting in memory first.
    index = NearestNeighbors(metric="cosine").fit(vectors)
    distances, neighbours = index.radius_neighbors(vectors, radius=1 - threshold)

    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, which is why it fits in one screen with no dependency.
 *
 * The price is the search itself: no key means, in the worst case, every
 * pair. A neighbour index earns its place well before the file gets large.
 */

const NGRAM_SIZES = [2, 3, 4];

/** Lower case, strip accents and punctuation, collapse spaces. */
export function normalise(text) {
  return String(text).toLowerCase().normalize('NFKD')
    .replace(/\p{Diacritic}/gu, '').replace(/[^\p{L}\p{N}]+/gu, ' ').trim();
}

/** One comparable string per record. */
export function recordText(record) {
  return normalise(Object.values(record).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) {
      let score = 0;
      // Walking the shorter vector: only the grams the two records share
      // contribute anything to a cosine.
      for (const [gram, weight] of vectors[i]) score += weight * (vectors[j].get(gram) ?? 0);
      if (score >= threshold) 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 vector where "Société Nationale des Chemins de
Fer" and "SNCF" can land close together, because the model was trained on text
where they occur in the same places.

What this rung really costs is not the comparison, it is the deployment: a few
hundred megabytes of weights to load, a process to keep warm, and a vector
index once the file no longer fits in a list.

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 unicodedata

# 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) -> str:
    """Lower case, strip accents and punctuation, collapse spaces."""
    decomposed = unicodedata.normalize("NFKD", text.lower())
    letters = "".join(c for c in decomposed if not unicodedata.combining(c))
    return " ".join("".join(c if c.isalnum() else " " for c in letters).split())


def record_text(record: dict) -> str:
    """One comparable string per record."""
    return normalise(" ".join(str(value) for value in record.values()))


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")

    pairs = []
    for i in range(len(vectors)):
        for j in range(i + 1, len(vectors)):
            score = sum(x * y for x, y in zip(vectors[i], vectors[j]))
            if score >= threshold:
                pairs.append((i, j, round(score, 3)))
    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 vector where "Société Nationale des Chemins
 * de Fer" and "SNCF" can land close together, because the model was trained
 * on text where they occur in the same places.
 *
 * What this rung really costs is not the comparison, it is the deployment: a
 * few hundred megabytes of weights to load, a process to keep warm, and a
 * vector index once the file no longer fits in an array.
 *
 * 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 and punctuation, collapse spaces. */
export function normalise(text) {
  return String(text).toLowerCase().normalize('NFKD')
    .replace(/\p{Diacritic}/gu, '').replace(/[^\p{L}\p{N}]+/gu, ' ').trim();
}

/** One comparable string per record. */
export function recordText(record) {
  return normalise(Object.values(record).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('@xenova/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`);
  }

  const pairs = [];
  for (let i = 0; i < vectors.length; i += 1) {
    for (let j = i + 1; j < vectors.length; j += 1) {
      const score = vectors[i].reduce((sum, value, d) => sum + value * vectors[j][d], 0);
      if (score >= threshold) 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]);
}