Repérer le spam dans un formulaire de contact

"""
Reject spam in a contact form: honeypot, submission delay, link cap, banned phrases.

Rung N0. Four checks, no dependency, no training data, and a rejection that
comes with a reason you can show to whoever asks why a message was lost.

Two of the checks look at the sender rather than the text. A honeypot field,
hidden in the form and left empty by every human being, and the time spent on
the page, catch the scripts that post to the endpoint without ever rendering
it. That is the bulk of the traffic, and no amount of reading the message
would have caught it any better.

The two text checks are the weak half, and the entry says so.
"""

import re
import unicodedata

# The field is present in the form, hidden by the stylesheet, and named after
# something a naive form filler will want to complete.
HONEYPOT_FIELD = "website"

MINIMUM_SECONDS = 3.0
MAXIMUM_LINKS = 2

LINK = re.compile(r"https?://|www\.|\b[\w-]+\.(?:com|net|org|ru|xyz|top)\b")

# Phrases that no customer of this form has ever written, and that the trade
# they come from cannot do without.
BANNED = re.compile(r"backlink|guest post|seo (?:services|ranking)|casino|crypto|viagra")


def fold(text: str) -> str:
    """Lowercase and strip accents, so `Rétrolien` and `RETROLIEN` match alike."""
    stripped = unicodedata.normalize("NFKD", text)
    return "".join(c for c in stripped if not unicodedata.combining(c)).casefold()


def reasons(fields: dict, seconds_on_page: float) -> list[str]:
    """
    Every reason to reject this submission. An empty list means: accept it.

    Returning reasons rather than a boolean is what makes the rule reviewable:
    a rejection you cannot explain is a rejection you cannot tune.
    """
    found = []
    if fields.get(HONEYPOT_FIELD, "").strip():
        found.append("honeypot filled")
    if seconds_on_page < MINIMUM_SECONDS:
        found.append("submitted too fast")

    message = fold(fields.get("message", ""))
    if len(LINK.findall(message)) > MAXIMUM_LINKS:
        found.append("too many links")
    found.extend(f"banned phrase: {phrase}" for phrase in sorted(set(BANNED.findall(message))))
    return found


def is_spam(fields: dict, seconds_on_page: float) -> bool:
    """The same decision, for callers that only want the verdict."""
    return bool(reasons(fields, seconds_on_page))

/**
 * Reject spam in a contact form: honeypot, submission delay, link cap, banned
 * phrases.
 *
 * Rung N0. Four checks, no dependency, no training data, and a rejection that
 * comes with a reason you can show to whoever asks why a message was lost.
 *
 * Two of the checks look at the sender rather than the text. A honeypot field,
 * hidden in the form and left empty by every human being, and the time spent
 * on the page, catch the scripts that post to the endpoint without ever
 * rendering it. That is the bulk of the traffic, and no amount of reading the
 * message would have caught it any better.
 *
 * The two text checks are the weak half, and the entry says so.
 */

// The field is present in the form, hidden by the stylesheet, and named after
// something a naive form filler will want to complete.
export const HONEYPOT_FIELD = 'website';

export const MINIMUM_SECONDS = 3;
export const MAXIMUM_LINKS = 2;

const LINK = /https?:\/\/|www\.|\b[\w-]+\.(?:com|net|org|ru|xyz|top)\b/g;

// Phrases that no customer of this form has ever written, and that the trade
// they come from cannot do without.
const BANNED = /backlink|guest post|seo (?:services|ranking)|casino|crypto|viagra/g;

/** Lowercase and strip accents, so `Rétrolien` and `RETROLIEN` match alike. */
export function fold(text) {
  return text.normalize('NFKD').replace(/\p{Diacritic}/gu, '').toLowerCase();
}

/**
 * Every reason to reject this submission. An empty list means: accept it.
 *
 * Returning reasons rather than a boolean is what makes the rule reviewable:
 * a rejection you cannot explain is a rejection you cannot tune.
 */
export function reasons(fields, secondsOnPage) {
  const found = [];
  if ((fields[HONEYPOT_FIELD] ?? '').trim()) found.push('honeypot filled');
  if (secondsOnPage < MINIMUM_SECONDS) found.push('submitted too fast');

  const message = fold(fields.message ?? '');
  if ((message.match(LINK) ?? []).length > MAXIMUM_LINKS) found.push('too many links');
  for (const phrase of [...new Set(message.match(BANNED) ?? [])].sort()) {
    found.push(`banned phrase: ${phrase}`);
  }
  return found;
}

/** The same decision, for callers that only want the verdict. */
export function isSpam(fields, secondsOnPage) {
  return reasons(fields, secondsOnPage).length > 0;
}

"""
Tell spam from a real enquiry with a linear classifier on character n-grams.

Rung N1. The rules of N0 look for the words a spammer used last year. This
looks at how the message is written: a few hundred labelled submissions, the
kind an inbox already holds, and the model learns the register rather than the
vocabulary list.

Character n-grams rather than words, for two reasons. They survive the
spellings a sender uses to dodge a word list, `b a c k l i n k s`, `backl1nks`,
and they need no tokeniser that would have to be tuned per language.

The decision is a weighted sum, so you can print
the features that pushed a message over the line, which matters the first time
someone asks why their enquiry was rejected.
"""

import unicodedata

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


def fold(text: str) -> str:
    """Lowercase and strip accents, so casing never doubles the feature space."""
    stripped = unicodedata.normalize("NFKD", text)
    return "".join(c for c in stripped if not unicodedata.combining(c)).casefold()


def train(messages: list[str], labels: list[int]):
    """`labels` is 1 when the submission is spam, 0 when it is a real enquiry."""
    model = make_pipeline(
        # `char_wb` keeps n-grams inside word boundaries, so the model learns
        # word shapes rather than the way two neighbours happen to collide.
        TfidfVectorizer(analyzer="char_wb", ngram_range=(3, 5), sublinear_tf=True, min_df=1),
        # Balanced, because a real inbox holds far more spam than enquiries and
        # the rare class is the one worth getting right. `C` above one because
        # a few hundred examples with a heavy regulariser leave every score
        # sitting near a half, which makes the threshold below meaningless.
        LogisticRegression(class_weight="balanced", C=10.0, max_iter=1000),
    )
    model.fit([fold(m) for m in messages], labels)
    return model


def spam_score(model, message: str) -> float:
    """Probability that the submission is spam, between zero and one."""
    return float(model.predict_proba([fold(message)])[0][1])


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

    Move it towards 1 when losing a real enquiry is the expensive mistake.
    Move it towards 0 when a spam message reaching a human is the expensive one.
    """
    return spam_score(model, message) >= threshold

/**
 * Tell spam from a real enquiry with a linear classifier on character n-grams.
 *
 * Rung N1. The rules of N0 look for the words a spammer used last year. This
 * looks at how the message is written: a few hundred labelled submissions, the
 * kind an inbox already holds, and the model learns the register rather than
 * the vocabulary list.
 *
 * Written out in full rather than pulled from a library, because TF-IDF on
 * character n-grams and a logistic regression fit by gradient descent is forty
 * lines. That is the whole argument of this rung: the classical tool is small
 * enough to read.
 */

const BUCKETS = 1024; // hashing trick: no vocabulary to build, or to ship
const NGRAMS = [3, 4, 5]; // long enough to carry a word, short enough to survive a typo

/** Lowercase and strip accents, so casing never doubles the feature space. */
export function fold(text) {
  return text.normalize('NFKD').replace(/\p{Diacritic}/gu, '').toLowerCase();
}

/** How often each hashed character n-gram occurs in the message. */
function counts(text) {
  const padded = ` ${fold(text)} `;
  const seen = new Float64Array(BUCKETS);
  for (const n of NGRAMS) {
    for (let i = 0; i + n <= padded.length; i += 1) {
      let h = 2166136261;
      for (const c of padded.slice(i, i + n)) h = ((h ^ c.codePointAt(0)) * 16777619) >>> 0;
      seen[h % BUCKETS] += 1;
    }
  }
  return seen;
}

/** Sublinear term frequency, weighted by inverse document frequency, L2 normalised. */
function vector(seen, idf) {
  const v = seen.map((count, j) => (count ? (1 + Math.log(count)) * idf[j] : 0));
  const norm = Math.sqrt(v.reduce((sum, x) => sum + x * x, 0));
  return norm ? v.map((x) => x / norm) : v;
}

/** `labels` is 1 when the submission is spam, 0 when it is a real enquiry. */
export function train(messages, labels, { epochs = 300, rate = 0.5 } = {}) {
  const raw = messages.map(counts);
  // Inverse document frequency: an n-gram every message carries says nothing.
  const idf = new Float64Array(BUCKETS).map((_, j) =>
    Math.log((1 + raw.length) / (1 + raw.filter((seen) => seen[j] > 0).length)) + 1);
  const rows = raw.map((seen) => vector(seen, idf));
  const model = { weights: new Float64Array(BUCKETS), bias: 0, idf };

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

/** The logistic of the weighted sum: one number between zero and one. */
function probability(model, row) {
  let z = model.bias;
  for (let j = 0; j < BUCKETS; j += 1) z += model.weights[j] * row[j];
  return 1 / (1 + Math.exp(-z));
}

/** Probability that the submission is spam. */
export function spamScore(model, message) {
  return probability(model, vector(counts(message), model.idf));
}

/**
 * Returns a decision, and the threshold is yours to set.
 *
 * Move it towards 1 when losing a real enquiry is the expensive mistake.
 * Move it towards 0 when a spam message reaching a human is the expensive one.
 */
export function isSpam(model, message, threshold = 0.5) {
  return spamScore(model, message) >= threshold;
}

"""
Sort a contact form submission 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: cap the input size, retry a
provider that failed, parse an answer that is only probably valid JSON, and
refuse to guess 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.

Note too what it cannot do. The message goes into the same prompt as the
instructions, and nothing in the protocol tells the model which of the two to
obey.
"""

from __future__ import annotations

import json

PROMPT = (
    "You moderate the contact form of a small company.\n"
    "Decide whether the submission below is unsolicited commercial spam.\n"
    'Answer with JSON only: {{"spam": true or false, "reason": "one short sentence"}}\n\n'
    "Submission:\n{message}"
)

MAX_CHARACTERS = 4000


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


def classify(message: str, client=None, *, attempts: int = 3) -> dict:
    """
    Return `{"spam": bool, "reason": str}` for one form submission.

    `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, and a form field is a place where anyone
    # can paste a novel. Refusing oversized input is not an optimisation, it is
    # a cost control.
    if len(message) > MAX_CHARACTERS:
        raise ValueError(f"submission longer than {MAX_CHARACTERS} characters")

    verdict = _ask(client, message, attempts)
    return {"spam": verdict["spam"], "reason": str(verdict.get("reason", ""))}


def _ask(client, message: str, attempts: int) -> dict:
    last_error: Exception | None = None
    for _ in range(attempts):
        try:
            # Temperature zero, because a moderation decision that changes
            # between two identical calls cannot be reviewed.
            answer = client.complete(prompt=PROMPT.format(message=message), temperature=0)
            parsed = json.loads(answer)
            if isinstance(parsed, dict) and isinstance(parsed.get("spam"), bool):
                return parsed
            last_error = ValueError("the model answered without a usable verdict")
        except Exception as error:  # noqa: BLE001 - any provider failure is retried
            last_error = error
    raise ClassificationUnavailable(str(last_error))

/**
 * Sort a contact form submission 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: cap the input size, retry a
 * provider that failed, parse an answer that is only probably valid JSON, and
 * refuse to guess 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.
 *
 * Note too what it cannot do. The message goes into the same prompt as the
 * instructions, and nothing in the protocol tells the model which of the two
 * to obey.
 */

const PROMPT = [
  'You moderate the contact form of a small company.',
  'Decide whether the submission below is unsolicited commercial spam.',
  'Answer with JSON only: {"spam": true or false, "reason": "one short sentence"}',
  '',
  'Submission:',
].join('\n');

export const MAX_CHARACTERS = 4000;

export class ClassificationUnavailable extends Error {}

/**
 * Return `{ spam, reason }` for one form submission.
 *
 * @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 classify(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, and a form field is a place where anyone can
  // paste a novel. Refusing oversized input is not an optimisation, it is a
  // cost control.
  if (message.length > MAX_CHARACTERS) {
    throw new RangeError(`submission longer than ${MAX_CHARACTERS} characters`);
  }

  const verdict = await ask(client, message, attempts);
  return { spam: verdict.spam, reason: String(verdict.reason ?? '') };
}

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 moderation decision that changes between
        // two identical calls cannot be reviewed.
        temperature: 0,
      });
      const parsed = JSON.parse(answer);
      if (parsed && typeof parsed.spam === 'boolean') return parsed;
      lastError = new Error('the model answered without a usable verdict');
    } catch (error) {
      lastError = error;
    }
  }
  throw new ClassificationUnavailable(String(lastError));
}