Créer un moniteur de citations IA en Python (prototype local, code pédagogique)

Ce prototype enregistre des observations déjà collectées dans un fichier CSV. Il n’interroge aucun moteur, ne contourne aucune interface, n’utilise aucune clé API et ne prétend pas mesurer un classement universel. Son rôle est plus modeste : rendre une collecte manuelle contrôlable, dédupliquée et exportable.

Réponse en bref

Stockez chaque réponse avec son moteur, son interface, son prompt exact, sa date, ses liens et son codage. Importez ces lignes dans SQLite avec une empreinte SHA-256 pour éviter les doublons. Calculez les taux à partir de numérateurs et dénominateurs visibles. Le code ci-dessous est un prototype pédagogique à relire et à tester dans votre environnement avant tout usage réel.

Ce que le prototype fait et ne fait pas

Il fait

importer un CSV encodé en UTF-8 ;
valider les colonnes et les dates ;
normaliser trois indicateurs booléens ;
conserver une réponse et les URL citées ;
créer une empreinte déterministe ;
ignorer une observation strictement dupliquée ;
calculer les taux par moteur ;
exporter les lignes en JSON.

Il ne fait pas

lancer des prompts automatiquement ;
ouvrir un navigateur ou simuler un utilisateur ;
appeler une API ;
vérifier qu’un user-agent correspond réellement à un robot ;
décider seul si une description est exacte ;
chiffrer la base ;
démontrer un lien causal entre une action SEO et une citation.

Cette frontière est volontaire. L’automatisation de la collecte dépend des conditions, API, coûts et politiques de chaque fournisseur. Elle doit être conçue séparément.

Préparer le fichier d’observations

Créez un fichier observations.csv avec ces colonnes :

Colonne	Format	Règle
observed_at	ISO 8601 avec fuseau	exemple : 2026-07-12T09:30:00+02:00
engine	texte	nom stable choisi par l’équipe
interface	texte	web, mobile ou API
model	texte facultatif	uniquement si le modèle ou mode est affiché
locale	texte	exemple : fr-FR
prompt_id	texte	identifiant stable du panel
prompt	texte	formulation exacte
response	texte	réponse brute autorisée à être stockée
cited_urls	URL séparées par une barre verticale	laisser vide si aucune
mentioned	yes ou no	l’entité est-elle explicitement nommée ?
recommended	yes ou no	est-elle explicitement proposée comme option ?
accurate	yes, no ou unknown	seulement si la mention est évaluable
notes	texte facultatif	contexte ou désaccord de codage

Exemple entièrement fictif :

observed_at,engine,interface,model,locale,prompt_id,prompt,response,cited_urls,mentioned,recommended,accurate,notes
2026-07-12T09:30:00+02:00,Moteur-A,web,,fr-FR,formation-01,"Quelles ressources comparer ?","Réponse d'exemple sans entité suivie.",,no,no,unknown,pilote fictif
2026-07-12T09:45:00+02:00,Moteur-B,web,,fr-FR,formation-01,"Quelles ressources comparer ?","Réponse fictive mentionnant Exemple Conseil.",https://example.org/ressource,yes,yes,yes,pilote fictif

Ne mettez pas de réponse réelle dans un dépôt public si elle contient des données personnelles, des informations de compte ou des contenus dont la republication n’est pas autorisée.

Le prototype Python

Enregistrez ce code sous citation_monitor.py dans un espace local de travail. Il utilise uniquement la bibliothèque standard de Python.

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import csv
import hashlib
import json
import sqlite3
from datetime import datetime
from pathlib import Path
from typing import Any


REQUIRED_COLUMNS = {
    "observed_at",
    "engine",
    "interface",
    "model",
    "locale",
    "prompt_id",
    "prompt",
    "response",
    "cited_urls",
    "mentioned",
    "recommended",
    "accurate",
    "notes",
}

SCHEMA = """
CREATE TABLE IF NOT EXISTS observations (
    id INTEGER PRIMARY KEY,
    fingerprint TEXT NOT NULL UNIQUE,
    observed_at TEXT NOT NULL,
    engine TEXT NOT NULL,
    interface TEXT NOT NULL,
    model TEXT NOT NULL,
    locale TEXT NOT NULL,
    prompt_id TEXT NOT NULL,
    prompt TEXT NOT NULL,
    response TEXT NOT NULL,
    cited_urls TEXT NOT NULL,
    mentioned INTEGER NOT NULL CHECK (mentioned IN (0, 1)),
    recommended INTEGER NOT NULL CHECK (recommended IN (0, 1)),
    accurate INTEGER CHECK (accurate IN (0, 1) OR accurate IS NULL),
    notes TEXT NOT NULL,
    imported_at TEXT NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_observations_engine
ON observations(engine);

CREATE INDEX IF NOT EXISTS idx_observations_prompt
ON observations(prompt_id);
"""


def parse_flag(value: str, *, allow_unknown: bool = False) -> int | None:
    normalized = (value or "").strip().lower()
    if normalized in {"yes", "oui", "true", "1"}:
        return 1
    if normalized in {"no", "non", "false", "0"}:
        return 0
    if allow_unknown and normalized in {"", "unknown", "inconnu", "na"}:
        return None
    expected = "yes/no/unknown" if allow_unknown else "yes/no"
    raise ValueError(f"Valeur booléenne invalide {value!r}; attendu : {expected}")


def require_text(row: dict[str, str], field: str) -> str:
    value = (row.get(field) or "").strip()
    if not value:
        raise ValueError(f"Champ obligatoire vide : {field}")
    return value


def normalize_observation(row: dict[str, str]) -> dict[str, Any]:
    observed_at = require_text(row, "observed_at")
    parsed_date = datetime.fromisoformat(observed_at.replace("Z", "+00:00"))
    if parsed_date.tzinfo is None:
        raise ValueError("observed_at doit contenir un fuseau horaire")

    mentioned = parse_flag(row.get("mentioned", ""))
    recommended = parse_flag(row.get("recommended", ""))
    accurate = parse_flag(row.get("accurate", ""), allow_unknown=True)

    if not mentioned and accurate is not None:
        raise ValueError("accurate doit valoir unknown si l'entité n'est pas mentionnée")
    if recommended and not mentioned:
        raise ValueError("recommended=yes exige mentioned=yes")

    urls = sorted(
        {
            url.strip()
            for url in (row.get("cited_urls") or "").split("|")
            if url.strip()
        }
    )

    payload: dict[str, Any] = {
        "observed_at": parsed_date.isoformat(),
        "engine": require_text(row, "engine"),
        "interface": require_text(row, "interface"),
        "model": (row.get("model") or "").strip(),
        "locale": require_text(row, "locale"),
        "prompt_id": require_text(row, "prompt_id"),
        "prompt": require_text(row, "prompt"),
        "response": require_text(row, "response"),
        "cited_urls": "|".join(urls),
        "mentioned": mentioned,
        "recommended": recommended,
        "accurate": accurate,
        "notes": (row.get("notes") or "").strip(),
    }

    canonical = json.dumps(
        payload,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    )
    payload["fingerprint"] = hashlib.sha256(
        canonical.encode("utf-8")
    ).hexdigest()
    return payload


def connect(database: Path) -> sqlite3.Connection:
    connection = sqlite3.connect(database)
    connection.row_factory = sqlite3.Row
    connection.executescript(SCHEMA)
    return connection


def import_csv(connection: sqlite3.Connection, source: Path) -> tuple[int, int]:
    inserted = 0
    duplicates = 0

    with source.open("r", encoding="utf-8", newline="") as stream:
        reader = csv.DictReader(stream)
        columns = set(reader.fieldnames or [])
        missing = REQUIRED_COLUMNS - columns
        if missing:
            raise ValueError(
                "Colonnes manquantes : " + ", ".join(sorted(missing))
            )

        with connection:
            for line_number, row in enumerate(reader, start=2):
                try:
                    item = normalize_observation(row)
                except (ValueError, TypeError) as error:
                    raise ValueError(
                        f"Ligne {line_number} invalide : {error}"
                    ) from error

                cursor = connection.execute(
                    """
                    INSERT OR IGNORE INTO observations (
                        fingerprint, observed_at, engine, interface, model,
                        locale, prompt_id, prompt, response, cited_urls,
                        mentioned, recommended, accurate, notes, imported_at
                    ) VALUES (
                        :fingerprint, :observed_at, :engine, :interface, :model,
                        :locale, :prompt_id, :prompt, :response, :cited_urls,
                        :mentioned, :recommended, :accurate, :notes, :imported_at
                    )
                    """,
                    {
                        **item,
                        "imported_at": datetime.now().astimezone().isoformat(),
                    },
                )
                if cursor.rowcount == 1:
                    inserted += 1
                else:
                    duplicates += 1

    return inserted, duplicates


def percentage(numerator: int, denominator: int) -> str:
    if denominator == 0:
        return "n/a"
    return f"{100 * numerator / denominator:.1f}%"


def print_report(connection: sqlite3.Connection) -> None:
    rows = connection.execute(
        """
        SELECT
            engine,
            COUNT(*) AS total,
            SUM(mentioned) AS mentions,
            SUM(recommended) AS recommendations,
            SUM(CASE WHEN cited_urls <> '' THEN 1 ELSE 0 END) AS citations,
            SUM(CASE WHEN accurate IS NOT NULL THEN 1 ELSE 0 END) AS accuracy_known,
            COALESCE(SUM(accurate), 0) AS accurate_count
        FROM observations
        GROUP BY engine
        ORDER BY engine
        """
    ).fetchall()

    if not rows:
        print("Aucune observation.")
        return

    header = (
        "engine",
        "n",
        "mention",
        "citation",
        "recommandation",
        "exactitude*",
    )
    print("\t".join(header))
    for row in rows:
        print(
            "\t".join(
                (
                    row["engine"],
                    str(row["total"]),
                    percentage(row["mentions"], row["total"]),
                    percentage(row["citations"], row["total"]),
                    percentage(row["recommendations"], row["total"]),
                    percentage(row["accurate_count"], row["accuracy_known"]),
                )
            )
        )
    print("* exactitude calculée uniquement sur les mentions évaluables")


def export_json(connection: sqlite3.Connection, destination: Path) -> int:
    rows = [
        dict(row)
        for row in connection.execute(
            """
            SELECT observed_at, engine, interface, model, locale,
                   prompt_id, prompt, response, cited_urls,
                   mentioned, recommended, accurate, notes
            FROM observations
            ORDER BY observed_at, engine, prompt_id
            """
        )
    ]
    destination.write_text(
        json.dumps(rows, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    return len(rows)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Moniteur local d'observations de citations IA"
    )
    parser.add_argument(
        "--db",
        type=Path,
        default=Path("citations.sqlite3"),
        help="chemin de la base SQLite locale",
    )
    commands = parser.add_subparsers(dest="command", required=True)
    commands.add_parser("init", help="initialiser la base")

    import_command = commands.add_parser("import", help="importer un CSV")
    import_command.add_argument("source", type=Path)

    commands.add_parser("report", help="afficher les taux par moteur")

    export_command = commands.add_parser("export", help="exporter en JSON")
    export_command.add_argument("destination", type=Path)
    return parser


def main() -> None:
    args = build_parser().parse_args()
    with connect(args.db) as connection:
        if args.command == "init":
            print(f"Base initialisée : {args.db}")
        elif args.command == "import":
            inserted, duplicates = import_csv(connection, args.source)
            print(f"Importées : {inserted}; doublons ignorés : {duplicates}")
        elif args.command == "report":
            print_report(connection)
        elif args.command == "export":
            count = export_json(connection, args.destination)
            print(f"Exportées : {count} observations vers {args.destination}")


if __name__ == "__main__":
    main()

Utilisation locale

Le prototype suppose une version maintenue de Python 3. Il faut confirmer la version réellement disponible avant usage.

python3 citation_monitor.py --db citations.sqlite3 init
python3 citation_monitor.py --db citations.sqlite3 import observations.csv
python3 citation_monitor.py --db citations.sqlite3 report
python3 citation_monitor.py --db citations.sqlite3 export observations.json

Le code utilise des paramètres nommés pour l’insertion SQL. L’empreinte dépend de tous les champs normalisés ; corriger une réponse ou une note crée donc une nouvelle observation. Si vous voulez gérer des révisions, ajoutez un identifiant d’observation et une table d’historique au lieu de modifier silencieusement les lignes.

Lire les indicateurs

Le rapport calcule :

mentions / observations totales ;
observations avec au moins une URL / observations totales ;
recommandations / observations totales ;
mentions exactes / mentions dont l’exactitude a été codée.

Il affiche le nombre d’observations pour empêcher un pourcentage de masquer un petit échantillon. Un taux de 100 % sur deux lignes n’a pas la même portée qu’un taux observé sur plusieurs fenêtres comparables.

Le champ cited_urls décrit la présence d’un lien dans la réponse enregistrée. Il ne prouve pas que le moteur a lu la page, que l’utilisateur a cliqué ni que la citation est positive.

Améliorations avant un usage réel

Qualité des données

ajouter une table des prompts et versions du panel ;
conserver l’identité de l’observateur dans un identifiant interne ;
faire relire un échantillon de codage ;
séparer les erreurs critiques des imprécisions mineures ;
documenter les changements d’interface.

Sécurité

chiffrer le volume ou utiliser un stockage approuvé ;
retirer les réponses contenant des données non nécessaires ;
limiter les droits du fichier et les sauvegardes ;
définir une durée de conservation ;
ne jamais mettre de clé API dans le CSV, la base ou le dépôt.

Analyse

produire les indicateurs par famille d’intention et période ;
afficher les numérateurs et dénominateurs ;
calculer des intervalles adaptés lorsque le volume le justifie ;
conserver les ruptures de série au lieu de les lisser ;
relier le panel, les analytics et les journaux sans les confondre.

Automatisation

Si une API officielle est utilisée plus tard, ajoutez un adaptateur distinct par fournisseur, une estimation de coût, une limite de débit, une journalisation des erreurs et un mécanisme d’arrêt. Ne remplacez pas la collecte manuelle par un navigateur automatisé sans vérifier les conditions applicables.

Limites

Ce code n’a pas valeur de produit prêt pour la production. Il doit être relu, testé et adapté à l’environnement cible. Il ne gère ni chiffrement, ni migrations, ni concurrence, ni sauvegarde, ni consentement. SHA-256 sert ici à reconnaître une ligne identique, pas à protéger le contenu de la base. Les indicateurs dépendent entièrement de la qualité du panel et du codage humain.

Prochaine étape

Avant d’automatiser la collecte, définir le panel, les règles de codage et la politique de conservation permet de savoir précisément quelles données le moniteur doit, et ne doit pas, stocker.

Sources vérifiées

Source externe, docs.python.org , consultée le 12 septembre 2026
Source externe, docs.python.org , consultée le 12 septembre 2026
Source externe, docs.python.org , consultée le 12 septembre 2026
Source externe, developers.google.com , consultée le 12 septembre 2026
Source externe, help.openai.com , consultée le 12 septembre 2026
Source externe, docs.perplexity.ai , consultée le 12 septembre 2026