Repérer les doublons dans un fichier

"""
Repérer les doublons d'un fichier en comparant l'orthographe, pas les
chaînes.

Niveau N1. La clé de blocage de N0 décide à l'avance quelles paires méritent
un regard, et tout ce qu'elle place dans deux groupes différents reste
invisible. Ce niveau abandonne la clé : chaque fiche devient un vecteur de
ses n-grammes de caractères, et les voisins se cherchent par cosinus.

Pourquoi des n-grammes de caractères plutôt que des mots : ils survivent à
une faute de frappe, à un ordre des mots inversé et à un champ tronqué,
parce qu'un mot mal orthographié partage encore l'essentiel de ses tranches
de trois lettres avec le mot correct. Rien ici n'est appris d'un corpus, et
il n'y a aucun modèle à entraîner. C'est une pondération et une distance.

Le prix, c'est la recherche elle-même : sans clé, toutes les paires du
fichier reçoivent un score, et le coût croît donc avec le carré du fichier.
La classe de latence affichée pour ce niveau est mesurée sur le fichier
nominal de ses tests ; vingt mille fiches sont une affaire de secondes, pas
de millisecondes.
"""

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

/**
 * Repérer les doublons d'un fichier en comparant l'orthographe, pas les
 * chaînes.
 *
 * Niveau N1. La clé de blocage de N0 décide à l'avance quelles paires méritent
 * un regard, et tout ce qu'elle place dans deux groupes différents reste
 * invisible. Ce niveau abandonne la clé : chaque fiche devient un vecteur de
 * ses n-grammes de caractères, et les paires sont classées par cosinus.
 *
 * Pourquoi des n-grammes de caractères plutôt que des mots : ils survivent à
 * une faute de frappe, à un ordre des mots inversé et à un champ tronqué,
 * parce qu'un mot mal orthographié partage encore l'essentiel de ses tranches
 * de trois lettres avec le mot correct. Rien ici n'est appris d'un corpus, et
 * il n'y a aucun modèle à entraîner. C'est une pondération et une distance,
 * écrites ici sans dépendance.
 *
 * Le prix, c'est la recherche elle-même : sans clé, toutes les paires du
 * fichier reçoivent un score, et le coût croît donc avec le carré du fichier.
 * La classe de latence affichée pour ce niveau est mesurée sur le fichier
 * nominal de ses tests ; vingt mille fiches sont une affaire de secondes, pas
 * de millisecondes.
 */

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]);
}

"""
Repérer les doublons d'un fichier avec un encodeur de phrases auto-hébergé.

Niveau N2. N1 compare des graphies ; deux fiches qui disent la même chose
avec d'autres mots ne partagent aucun n-gramme de caractères et restent
invisibles. Un encodeur de phrases envoie chaque fiche vers un vecteur dense,
la moyenne des vecteurs de ses jetons, et les fiches sont comparées sur ces
vecteurs plutôt que sur les lettres qu'elles partagent.

Deux prix, et c'est le second qui grossit. Le prix fixe est le déploiement :
des poids à télécharger et à charger, et un processus à garder chaud.
L'autre est la comparaison, et il est quadratique — toutes les paires du
fichier reçoivent un score, mille fiches font un demi-million de produits
scalaires de trois cent quatre-vingt-quatre nombres, dix mille en font
cinquante millions. Au-delà de quelques dizaines de milliers de fiches, il
faut bloquer d'abord avec la clé du niveau N0, ou passer à un index
approché ; cet extrait ne fait ni l'un ni l'autre, à dessein, parce qu'il
montre la distance et non la recherche.

L'encodeur est un paramètre doté d'une vraie valeur par défaut : le lecteur
voit le code de chargement, et le test injecte un double local.
"""

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

/**
 * Repérer les doublons d'un fichier avec un encodeur de phrases auto-hébergé.
 *
 * Niveau N2. N1 compare des graphies ; deux fiches qui disent la même chose
 * avec d'autres mots ne partagent aucun n-gramme de caractères et restent
 * invisibles. Un encodeur de phrases envoie chaque fiche vers un vecteur dense,
 * la moyenne des vecteurs de ses jetons, et les fiches sont comparées sur ces
 * vecteurs plutôt que sur les lettres qu'elles partagent.
 *
 * Deux prix, et c'est le second qui grossit. Le prix fixe est le déploiement :
 * des poids à télécharger et à charger, et un processus à garder chaud.
 * L'autre est la comparaison, et il est quadratique — toutes les paires du
 * fichier reçoivent un score, mille fiches font un demi-million de produits
 * scalaires de trois cent quatre-vingt-quatre nombres, dix mille en font
 * cinquante millions. Au-delà de quelques dizaines de milliers de fiches, il
 * faut bloquer d'abord avec la clé du niveau N0, ou passer à un index
 * approché ; cet extrait ne fait ni l'un ni l'autre, à dessein, parce qu'il
 * montre la distance et non la recherche.
 *
 * L'encodeur est un paramètre doté d'une vraie valeur par défaut : le lecteur
 * voit le code de chargement, et le test injecte un double local.
 */

// 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]);
}