Masquer les coordonnées dans un message

"""
Mask personal data in a chat message: normalisation, then regular expressions.

Rung N0. Deterministic, standard library only, fast enough that you will never
find it in a profile.

Two things make this work.

First, normalisation only touches the spellings of a space. Rewriting the whole
message before matching would destroy the very characters an email address is
made of.

Second, each pattern tolerates the separators people actually type inside a
number, instead of assuming one canonical form.
"""

import re
import unicodedata

# The space characters French typography puts inside numbers.
UNUSUAL_SPACES = re.compile(r"[     ]")

EMAIL = re.compile(r"[\w.+-]+@[\w-]+(?:\.[\w-]+)+")

# French numbers: 0X XX XX XX XX, or +33 X XX XX XX XX. The separator between
# digits may be a space, a dot or a dash, or absent.
SEP = r"[ .-]?"
PHONE = re.compile(rf"(?<![\d+]){SEP}(?:\+{SEP}33{SEP}|0)[1-9](?:{SEP}\d){{8}}(?!\d)")

# IBAN: two letters, two check digits, then up to thirty alphanumerics,
# conventionally grouped in fours.
IBAN = re.compile(r"(?<![A-Z0-9])[A-Z]{2} ?\d{2}(?: ?[A-Z0-9]){10,28}(?![A-Z0-9])")

# Order matters: an email may contain digits that would otherwise be read as
# the start of a phone number.
PATTERNS = ((EMAIL, "[email]"), (IBAN, "[iban]"), (PHONE, "[phone]"))


def normalise(text: str) -> str:
    """Reduce the many spellings of a space to a plain one."""
    return UNUSUAL_SPACES.sub(" ", unicodedata.normalize("NFKC", text))


def mask(text: str) -> str:
    """
    Replace contact details with a label naming what was removed.

    A label beats a row of asterisks: whoever reads the thread later can see
    that a phone number was removed, not merely that something was.
    """
    text = normalise(text)
    for pattern, label in PATTERNS:
        text = pattern.sub(label, text)
    return text

/**
 * Mask personal data in a chat message: normalisation, then regular expressions.
 *
 * Rung N0. Deterministic, no dependency, fast enough that you will never find
 * it in a profile.
 *
 * Two things make this work.
 *
 * First, normalisation only touches the spellings of a space. Rewriting the
 * whole message before matching would destroy the very characters an email
 * address is made of.
 *
 * Second, each pattern tolerates the separators people actually type inside a
 * number, instead of assuming one canonical form.
 */

// The space characters French typography puts inside numbers.
const UNUSUAL_SPACES = /[    ⁠]/g;

const EMAIL = /[\w.+-]+@[\w-]+(?:\.[\w-]+)+/g;

// French numbers: 0X XX XX XX XX, or +33 X XX XX XX XX. The separator between
// digits may be a space, a dot or a dash, or absent.
const SEP = '[ .-]?';
const PHONE = new RegExp(`(?<![\\d+])(?:\\+${SEP}33${SEP}|0)[1-9](?:${SEP}\\d){8}(?!\\d)`, 'g');

// IBAN: two letters, two check digits, then up to thirty alphanumerics,
// conventionally grouped in fours.
const IBAN = /(?<![A-Z0-9])[A-Z]{2} ?\d{2}(?: ?[A-Z0-9]){10,28}(?![A-Z0-9])/g;

// Order matters: an email may contain digits that would otherwise be read as
// the start of a phone number.
const PATTERNS = [
  [EMAIL, '[email]'],
  [IBAN, '[iban]'],
  [PHONE, '[phone]'],
];

/** Reduce the many spellings of a space to a plain one. */
export function normalise(text) {
  return text.normalize('NFKC').replace(UNUSUAL_SPACES, ' ');
}

/**
 * Replace contact details with a label naming what was removed.
 *
 * A label beats a row of asterisks: whoever reads the thread later can see
 * that a phone number was removed, not merely that something was.
 */
export function mask(text) {
  let out = normalise(text);
  for (const [pattern, label] of PATTERNS) {
    out = out.replace(pattern, label);
  }
  return out;
}

"""
Catch obfuscated contact details with a light classifier.

Rung N1. The regular expressions of N0 see one spelling of a phone number.
This sees the shape of one: a run of tokens that is mostly digits, mostly
short, and sitting next to words like "call" or "reach".

Training data is a few hundred labelled messages, not a few million. The
weights are small enough to keep in the repository next to this file, and
there is no service to run: the model loads with the process.
"""

import re

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline

# Digits written as words, and the characters used to stand in for digits.
DIGIT_WORDS = ("zero|one|two|three|four|five|six|seven|eight|nine|ten|"
               "zéro|un|deux|trois|quatre|cinq|six|sept|huit|neuf|dix")
LOOKALIKES = str.maketrans({"O": "0", "o": "0", "I": "1", "i": "1", "l": "1", "S": "5", "s": "5"})


def _is_digits_in_disguise(token: str) -> bool:
    """
    True when folding lookalike characters turns the whole token into digits.

    The token must already contain one real digit. Folding unconditionally
    would be a mistake: « loll » would fold to 1011 and read as a fragment of
    a phone number.
    """
    if not any(c.isdigit() for c in token):
        return False
    return token.translate(LOOKALIKES).isdigit()


def shape(text: str) -> str:
    """
    Turn a message into the features that matter, and drop the rest.

    A classifier trained on raw text memorises the training phone numbers.
    Trained on shapes, it learns what a hidden number looks like.
    """
    out = []
    for token in re.findall(r"[^\W_]+", text):
        lowered = token.lower()
        # Case matters for lookalikes, so fold before lowering.
        if re.fullmatch(DIGIT_WORDS, lowered):
            out.append("D")
        elif _is_digits_in_disguise(token):
            out.append("D" * len(token))
        elif len(lowered) >= 2:
            out.append(lowered)
    return " ".join(out)


def train(messages: list[str], labels: list[int]):
    """`labels` is 1 when the message hides contact details, 0 when it does not."""
    model = make_pipeline(
        TfidfVectorizer(analyzer="char_wb", ngram_range=(2, 4), min_df=1),
        LogisticRegression(class_weight="balanced", max_iter=1000),
    )
    model.fit([shape(m) for m in messages], labels)
    return model


def is_hiding_contact_details(model, message: str, threshold: float = 0.5) -> bool:
    """
    Returns a decision, and the threshold is yours to set.

    Move it towards 1 if a false positive means blocking a legitimate message.
    Move it towards 0 if letting one through is the worse outcome.
    """
    score = model.predict_proba([shape(message)])[0][1]
    return bool(score >= threshold)

/**
 * Catch obfuscated contact details with a light classifier.
 *
 * Rung N1. The regular expressions of N0 see one spelling of a phone number.
 * This sees the shape of one: a run of tokens that is mostly digits, mostly
 * short, sitting next to words like "call" or "reach".
 *
 * Written out in full rather than pulled from a library, because logistic
 * regression on hashed character n-grams is forty lines. That is the whole
 * argument of this rung: the classical tool is small enough to read.
 */

// Digits written as words, in English and in French.
const DIGIT_WORDS = new Set(
  ('zero one two three four five six seven eight nine ten ' +
    'zéro un deux trois quatre cinq six sept huit neuf dix').split(' '),
);

// Characters used to stand in for digits.
const LOOKALIKES = { O: '0', o: '0', I: '1', i: '1', l: '1', S: '5', s: '5' };

const BUCKETS = 512; // hashing trick: no vocabulary to build or ship

/**
 * True when folding lookalike characters turns the whole token into digits.
 * The token must already carry one real digit; folding unconditionally would
 * turn "loll" into 1011.
 */
function isDigitsInDisguise(token) {
  if (!/\d/.test(token)) return false;
  const folded = [...token].map((c) => LOOKALIKES[c] ?? c).join('');
  return /^\d+$/.test(folded);
}

/** Turn a message into the features that matter, and drop the rest. */
export function shape(text) {
  const out = [];
  for (const token of text.match(/[\p{L}\p{N}]+/gu) ?? []) {
    const lowered = token.toLowerCase();
    if (DIGIT_WORDS.has(lowered)) out.push('D');
    else if (isDigitsInDisguise(token)) out.push('D'.repeat(token.length));
    else if (lowered.length >= 2) out.push(lowered);
  }
  return out.join(' ');
}

/** Character n-grams of the shaped message, hashed into a fixed vector. */
function features(text) {
  const shaped = ` ${shape(text)} `;
  const vector = new Float64Array(BUCKETS);
  for (let n = 2; n <= 4; n += 1) {
    for (let i = 0; i + n <= shaped.length; i += 1) {
      let h = 2166136261;
      for (const c of shaped.slice(i, i + n)) h = ((h ^ c.codePointAt(0)) * 16777619) >>> 0;
      vector[h % BUCKETS] += 1;
    }
  }
  const norm = Math.hypot(...vector);
  return norm ? vector.map((v) => v / norm) : vector;
}

/** `labels` is 1 when the message hides contact details, 0 when it does not. */
export function train(messages, labels, { epochs = 400, rate = 0.5 } = {}) {
  const rows = messages.map(features);
  const weights = new Float64Array(BUCKETS);
  let bias = 0;

  for (let epoch = 0; epoch < epochs; epoch += 1) {
    for (let i = 0; i < rows.length; i += 1) {
      let z = bias;
      for (let j = 0; j < BUCKETS; j += 1) z += weights[j] * rows[i][j];
      const error = 1 / (1 + Math.exp(-z)) - labels[i];
      for (let j = 0; j < BUCKETS; j += 1) weights[j] -= rate * error * rows[i][j];
      bias -= rate * error;
    }
  }
  return { weights, bias };
}

/**
 * Returns a decision, and the threshold is yours to set.
 *
 * Move it towards 1 if a false positive means blocking a legitimate message.
 * Move it towards 0 if letting one through is the worse outcome.
 */
export function isHidingContactDetails(model, message, threshold = 0.5) {
  const vector = features(message);
  let z = model.bias;
  for (let j = 0; j < BUCKETS; j += 1) z += model.weights[j] * vector[j];
  return 1 / (1 + Math.exp(-z)) >= threshold;
}

"""
Mask personal data by asking a general-purpose model.

Rung N3. This is the option people reach for first. It is here so you can see
what it costs, not because this entry recommends it.

Note what the code has to do that N0 did not: retry on failure, cap the input
size, parse an answer that is only probably valid JSON, and fall back when the
answer is unusable. That plumbing is the real cost of this rung, and it is the
part your tests have to cover, because the model itself is not testable.
"""

from __future__ import annotations

import json

PROMPT = (
    "Find every piece of personal contact information in the message below.\n"
    "Answer with JSON only: a list of objects with keys `text` and `kind`,\n"
    "where `kind` is one of email, phone, iban, address.\n"
    "If there is none, answer with an empty list.\n\n"
    "Message:\n{message}"
)

MAX_CHARACTERS = 8000


class MaskingUnavailable(Exception):
    """The provider could not be reached, or answered something unusable."""


def mask(message: str, client=None, *, attempts: int = 3) -> str:
    """
    Replace contact details with a label naming what was removed.

    `client` is injected so this function can be tested without a network call.
    In production it defaults to a real provider client.
    """
    if client is None:  # pragma: no cover - needs a key and a network
        from openai import OpenAI

        client = OpenAI()

    # A model charges by the token. Refusing oversized input is not an
    # optimisation, it is a cost control.
    if len(message) > MAX_CHARACTERS:
        raise ValueError(f"message longer than {MAX_CHARACTERS} characters")

    found = _ask(client, message, attempts)

    # Replace the longest matches first, so a substring never eats its parent.
    for item in sorted(found, key=lambda i: len(i.get("text", "")), reverse=True):
        text, kind = item.get("text"), item.get("kind")
        if text and kind:
            message = message.replace(text, f"[{kind}]")
    return message


def _ask(client, message: str, attempts: int) -> list[dict]:
    last_error: Exception | None = None
    for _ in range(attempts):
        try:
            answer = client.complete(prompt=PROMPT.format(message=message), temperature=0)
            parsed = json.loads(answer)
            if isinstance(parsed, list):
                return parsed
            last_error = ValueError("the model answered something that is not a list")
        except Exception as error:  # noqa: BLE001 - any provider failure is retried
            last_error = error
    raise MaskingUnavailable(str(last_error))

/**
 * Mask personal data by asking a general-purpose model.
 *
 * Rung N3. This is the option people reach for first. It is here so you can
 * see what it costs, not because this entry recommends it.
 *
 * Note what the code has to do that N0 did not: retry on failure, cap the
 * input size, parse an answer that is only probably valid JSON, and refuse to
 * pass the message through unmasked when the answer is unusable. That plumbing
 * is the real cost of this rung, and it is the part your tests have to cover,
 * because the model itself is not testable.
 */

const PROMPT = [
  'Find every piece of personal contact information in the message below.',
  'Answer with JSON only: a list of objects with keys `text` and `kind`,',
  'where `kind` is one of email, phone, iban, address.',
  'If there is none, answer with an empty list.',
  '',
  'Message:',
].join('\n');

export const MAX_CHARACTERS = 8000;

export class MaskingUnavailable extends Error {}

/**
 * Replace contact details with a label naming what was removed.
 *
 * @param {string} message
 * @param {object} options
 * @param {{complete: Function}} [options.client] injected so this can be
 *   tested without a network call; defaults to a real provider client
 * @param {number} [options.attempts]
 */
export async function mask(message, { client, attempts = 3 } = {}) {
  if (!client) {
    // Needs a key and a network, so it is never reached in the tests.
    const { OpenAI } = await import('openai');
    client = new OpenAI();
  }

  // A model charges by the token. Refusing oversized input is not an
  // optimisation, it is a cost control.
  if (message.length > MAX_CHARACTERS) {
    throw new RangeError(`message longer than ${MAX_CHARACTERS} characters`);
  }

  const found = await ask(client, message, attempts);

  // Replace the longest matches first, so a substring never eats its parent.
  let out = message;
  for (const { text, kind } of [...found].sort((a, b) => (b.text?.length ?? 0) - (a.text?.length ?? 0))) {
    if (text && kind) out = out.split(text).join(`[${kind}]`);
  }
  return out;
}

async function ask(client, message, attempts) {
  let lastError;
  for (let i = 0; i < attempts; i += 1) {
    try {
      const answer = await client.complete({
        prompt: `${PROMPT}\n${message}`,
        // Temperature zero, because a masking decision that changes between
        // two identical calls cannot be reviewed.
        temperature: 0,
      });
      const parsed = JSON.parse(answer);
      if (Array.isArray(parsed)) return parsed;
      lastError = new Error('the model answered something that is not a list');
    } catch (error) {
      lastError = error;
    }
  }
  throw new MaskingUnavailable(String(lastError));
}