Analyse de sentiment multilingue sur Twitter avec Claude

Ce que vous saurez faire à la fin

1. Ingérer un flux Twitter (X API v2) filtré par mots-clés et hashtags pertinents pour votre marque
2. Analyser le sentiment en français, anglais, wolof transcrit et arabe avec un seul prompt Claude
3. Détecter automatiquement les topics émergents et les regrouper dans un dashboard temps réel
4. Configurer des alertes crise (Slack, WhatsApp Business) déclenchées par des seuils précis
5. Construire un mini-data warehouse SQLite ou PostgreSQL pour requêtes historiques et reporting

Durée : 5h. Pré-requis : compte Anthropic API, accès X API v2 niveau Basic à 100 USD/mois (62 000 FCFA), Python 3.10+ avec tweepy, anthropic, fastapi et uvicorn, base PostgreSQL ou SQLite locale, budget API mensuel 18 000 à 40 000 FCFA pour 5 000 tweets/jour analysés.

Étape 1 — Définir le périmètre de veille et les mots-clés

Listez 3 catégories de mots-clés : votre marque (variantes orthographiques, fautes courantes), vos produits phares (max 10 noms), vos concurrents directs (3 à 5 marques). Ajoutez des hashtags secteur (#Senegal, #Dakar, #Sunugaal, #BusinessAfrica) si vous voulez surveiller un marché. Pour une PME télécom au Sénégal, le périmètre type est : nom de marque + 5 offres principales + 3 concurrents + #Senegal + #DakarBusiness, soit environ 15 mots-clés actifs.

Étape 2 — Configurer l’accès Twitter API v2 et la règle de stream

import tweepy

BEARER_TOKEN = "votre_bearer_token_x"

class FluxTwitter(tweepy.StreamingClient):
    def on_tweet(self, tweet):
        traiter_tweet(tweet)
    def on_error(self, status):
        print(f"Erreur stream : {status}")

flux = FluxTwitter(BEARER_TOKEN)

regles_existantes = flux.get_rules().data or []
if regles_existantes:
    flux.delete_rules([r.id for r in regles_existantes])

regle = "(MarqueXYZ OR #Sunugaal OR Orange Senegal OR Free Senegal) lang:fr -is:retweet"
flux.add_rules(tweepy.StreamRule(regle))

flux.filter(tweet_fields=["created_at", "lang", "author_id", "public_metrics"])

Étape 3 — Concevoir le prompt d’analyse multilingue

PROMPT_SENTIMENT = """Tu es analyste social media multilingue spécialisé sur l'Afrique de l'Ouest.
Le tweet peut être en français, anglais, wolof transcrit en alphabet latin, ou arabe.

Tweet à analyser :
'{tweet_text}'

Analyse et retourne UNIQUEMENT du JSON valide :
{{
    "langue_detectee": "fr" ou "en" ou "wo" ou "ar" ou "mixte",
    "sentiment": "positif" ou "neutre" ou "negatif" ou "tres_negatif",
    "score_intensite": 0 à 10,
    "topics": ["topic1", "topic2"],
    "mention_marque": true ou false,
    "mention_concurrent": "nom du concurrent ou null",
    "appel_action": true ou false,
    "potentiel_viral": "faible" ou "moyen" ou "fort",
    "alerte_crise": true ou false,
    "raison_alerte": "explication courte ou null"
}}

CRITÈRES ALERTE CRISE :
- Tweet d'un compte avec > 5 000 followers ET sentiment tres_negatif
- Mention de mots : 'arnaque', 'scandale', 'porter plainte', 'CDP', 'ARTP', 'avocat'
- Plus de 3 réponses négatives en cascade sur le même fil
- Mention en wolof de termes péjoratifs forts ('xaalis bi', 'sax sax', etc.)

Topics : utilise des mots-clés courts en français même si le tweet est en wolof ou arabe."""

Étape 4 — Implémenter la fonction de traitement temps réel

import anthropic
import json
from datetime import datetime

client = anthropic.Anthropic()

def analyser_tweet(tweet_text):
    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=400,
        messages=[{
            "role": "user",
            "content": PROMPT_SENTIMENT.format(tweet_text=tweet_text)
        }]
    )
    try:
        return json.loads(response.content[0].text)
    except json.JSONDecodeError:
        return {"erreur": "JSON invalide", "raw": response.content[0].text}

def traiter_tweet(tweet):
    analyse = analyser_tweet(tweet.text)
    enregistrer_db(tweet, analyse)
    if analyse.get("alerte_crise"):
        envoyer_alerte(tweet, analyse)

Étape 5 — Créer la base de données pour stockage historique

import sqlite3

def init_db():
    conn = sqlite3.connect("veille_twitter.db")
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS tweets (
        tweet_id TEXT PRIMARY KEY,
        author_id TEXT,
        text TEXT,
        created_at TEXT,
        followers INTEGER,
        retweets INTEGER,
        likes INTEGER,
        langue TEXT,
        sentiment TEXT,
        score_intensite INTEGER,
        topics TEXT,
        mention_marque INTEGER,
        mention_concurrent TEXT,
        potentiel_viral TEXT,
        alerte_crise INTEGER,
        raison_alerte TEXT,
        analyse_at TEXT
    )''')
    c.execute('CREATE INDEX IF NOT EXISTS idx_sentiment ON tweets(sentiment)')
    c.execute('CREATE INDEX IF NOT EXISTS idx_alerte ON tweets(alerte_crise)')
    c.execute('CREATE INDEX IF NOT EXISTS idx_date ON tweets(created_at)')
    conn.commit()
    return conn

def enregistrer_db(tweet, analyse):
    conn = sqlite3.connect("veille_twitter.db")
    conn.execute('''INSERT OR REPLACE INTO tweets VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)''', (
        str(tweet.id),
        str(tweet.author_id),
        tweet.text,
        tweet.created_at.isoformat(),
        tweet.public_metrics.get("followers_count", 0),
        tweet.public_metrics.get("retweet_count", 0),
        tweet.public_metrics.get("like_count", 0),
        analyse.get("langue_detectee"),
        analyse.get("sentiment"),
        analyse.get("score_intensite"),
        json.dumps(analyse.get("topics", []), ensure_ascii=False),
        int(analyse.get("mention_marque", False)),
        analyse.get("mention_concurrent"),
        analyse.get("potentiel_viral"),
        int(analyse.get("alerte_crise", False)),
        analyse.get("raison_alerte"),
        datetime.now().isoformat()
    ))
    conn.commit()
    conn.close()

Étape 6 — Implémenter les alertes crise multi-canal

import requests

SLACK_WEBHOOK = "https://hooks.slack.com/services/XXX"
WHATSAPP_API = "https://graph.facebook.com/v18.0/PHONE_ID/messages"
WHATSAPP_TOKEN = "votre_token_whatsapp_business"

def envoyer_alerte(tweet, analyse):
    message = (
        f"ALERTE VEILLE TWITTER\n"
        f"Auteur : {tweet.author_id}\n"
        f"Texte : {tweet.text[:200]}\n"
        f"Sentiment : {analyse['sentiment']} (intensité {analyse['score_intensite']}/10)\n"
        f"Raison : {analyse['raison_alerte']}\n"
        f"URL : https://twitter.com/i/web/status/{tweet.id}"
    )
    requests.post(SLACK_WEBHOOK, json={"text": message})
    requests.post(
        WHATSAPP_API,
        headers={"Authorization": f"Bearer {WHATSAPP_TOKEN}"},
        json={
            "messaging_product": "whatsapp",
            "to": "221771234567",
            "type": "text",
            "text": {"body": message}
        }
    )

Étape 7 — Gérer la spécificité du wolof transcrit

Le wolof transcrit (« Orange dafa nax niit yi », « loolu deugue la ») n’est dans aucun corpus standard. Claude le comprend grâce à son entraînement multilingue. Quelques expressions clés à connaître pour le sentiment : baax na = c’est bien (positif), baaxul = ce n’est pas bien (négatif), dafa nax = c’est trompeur (très négatif), jërëjëf = merci (positif), weccet bi rey na = le service est rapide (positif). Ajoutez ces expressions dans le prompt si la précision wolof descend sous 80%.

Étape 8 — Construire le dashboard temps réel avec FastAPI

from fastapi import FastAPI
from fastapi.responses import HTMLResponse
import sqlite3

app = FastAPI()

@app.get("/api/kpi")
def kpi():
    conn = sqlite3.connect("veille_twitter.db")
    c = conn.cursor()
    c.execute("SELECT COUNT(*) FROM tweets WHERE date(created_at) = date('now')")
    total_jour = c.fetchone()[0]
    c.execute("SELECT sentiment, COUNT(*) FROM tweets WHERE date(created_at) = date('now') GROUP BY sentiment")
    repartition = dict(c.fetchall())
    c.execute("SELECT COUNT(*) FROM tweets WHERE alerte_crise = 1 AND date(created_at) = date('now')")
    alertes = c.fetchone()[0]
    c.execute("SELECT topics FROM tweets WHERE date(created_at) = date('now')")
    topics_brut = c.fetchall()
    return {
        "total_jour": total_jour,
        "repartition_sentiment": repartition,
        "alertes_jour": alertes,
        "topics_top": agreger_topics(topics_brut)
    }

def agreger_topics(rows):
    from collections import Counter
    tous = []
    for r in rows:
        tous.extend(json.loads(r[0] or "[]"))
    return Counter(tous).most_common(10)

Étape 9 — Calculer le coût mensuel et choisir Haiku vs Sonnet

Pour la veille standard, Haiku est largement suffisant. Réservez Sonnet aux comptes influenceurs (> 50 000 followers) et aux tweets candidats à la viralité (re-analysez en Sonnet seulement les tweets avec score_intensite ≥ 8).

Étape 10 — Détecter les topics émergents avec clustering

def detecter_topics_emergents(fenetre_heures=2):
    conn = sqlite3.connect("veille_twitter.db")
    c = conn.cursor()
    c.execute("""
        SELECT topics FROM tweets
        WHERE created_at > datetime('now', '-' || ? || ' hours')
    """, (fenetre_heures,))
    from collections import Counter
    tous = []
    for row in c.fetchall():
        tous.extend(json.loads(row[0] or "[]"))
    courant = Counter(tous)
    c.execute("""
        SELECT topics FROM tweets
        WHERE created_at < datetime('now', '-' || ? || ' hours')
        AND created_at > datetime('now', '-' || ? || ' hours')
    """, (fenetre_heures, fenetre_heures * 24))
    historique = []
    for row in c.fetchall():
        historique.extend(json.loads(row[0] or "[]"))
    base = Counter(historique)
    emergents = []
    for topic, n in courant.items():
        baseline = max(base.get(topic, 1), 1) / 24
        if n > baseline * 5:
            emergents.append({"topic": topic, "volume_courant": n, "ratio": n / baseline})
    return sorted(emergents, key=lambda x: x["ratio"], reverse=True)[:10]

Étape 11 — Mettre en place un résumé quotidien généré par Claude

Chaque matin à 07h00, générez un résumé de 200 mots de la veille de la nuit (00h-07h). Ce résumé est envoyé par email au comité de direction. Format type : volume total, répartition sentiment, top 3 topics, alertes traitées, recommandations action.

PROMPT_RESUME = """Voici les statistiques de veille Twitter des dernières 24h pour la marque XYZ au Sénégal.

Volume total : {total} tweets
Répartition sentiment : {repartition}
Top topics : {topics}
Alertes crise : {alertes}
Tweets les plus engageants : {top_tweets}

Rédige un résumé exécutif de 200 mots maximum pour le comité de direction.
Structure : (1) constat principal, (2) faits saillants, (3) recommandation d'action concrète.
Ton : factuel, direct, sans jargon. Inclus 1 chiffre clé en première phrase."""

def generer_resume_quotidien(stats):
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=500,
        messages=[{"role": "user", "content": PROMPT_RESUME.format(**stats)}]
    )
    return response.content[0].text

Étape 12 — Gérer les pics de volume et les rate limits

L’API Twitter Basic limite à 10 000 tweets/mois en lecture stream. Au-delà, passez au plan Pro à 5 000 USD/mois (rarement justifié pour une PME). En cas de pic (lancement produit, polémique virale), implémentez un buffer Redis : les tweets entrants s’empilent dans une file, et un worker les traite à 50 tweets/minute en respectant le rate limit Anthropic (50 RPM en tier 1, 1000 RPM en tier 4). Sans buffer, vous perdrez 30 à 60% des tweets pendant un pic.

Étape 13 — Conformité RGPD et loi sénégalaise CDP

La collecte de tweets publics est légale, mais le stockage de données personnelles (author_id, contenu) est encadré. Au Sénégal, la loi 2008-12 sur la protection des données personnelles (CDP) impose : déclaration du traitement à la CDP, durée de conservation maximale de 12 mois, droit d’accès et de suppression sur demande. Implémentez une route DELETE /api/tweets/by-author/{author_id} pour répondre aux demandes individuelles. Documentez le traitement dans un registre.

Étape 14 — Itérer sur le prompt avec feedback humain mensuel

Chaque fin de mois, échantillonnez 100 tweets aléatoires de la base et faites valider par un analyste : sentiment correct ? topics pertinents ? alerte justifiée ou faux positif ? Si la précision sentiment passe sous 88% ou si plus de 3 alertes par semaine sont des faux positifs, ajustez le prompt. Versionnez chaque changement dans Git avec un commentaire indiquant le delta de précision mesuré.

Erreurs classiques à éviter

• Stream sans filtre de langue ni filtre retweet : conséquence, 80% du flux est du bruit, votre quota Twitter explose en 3 jours.
• Alertes crise déclenchées sur le simple mot ‘nul’ : conséquence, 200 alertes par jour, votre équipe désactive les notifications, vous ratez la vraie crise.
• Pas de stockage en base, tout en mémoire : conséquence, redémarrage du serveur = perte de toute la veille, impossible de produire un rapport mensuel.
• Sonnet utilisé sur 100% des tweets : conséquence, facture mensuelle multipliée par 3 sans gain de précision significatif sur les tweets courts.
• Oublier la déclaration CDP : conséquence, sanction administrative jusqu’à 100 millions de FCFA en cas de plainte.

Checklist Sentiment Twitter multilingue

✓ Périmètre veille défini (15 mots-clés max)
✓ Accès X API v2 Basic activé et règle de stream testée
✓ Prompt Claude validé sur français, anglais, wolof, arabe
✓ Base SQLite ou PostgreSQL initialisée avec index
✓ Alertes Slack et WhatsApp Business configurées
✓ Coût mensuel API calculé selon volume cible
✓ Détection topics émergents avec ratio courant/historique
✓ Dashboard FastAPI accessible en interne
✓ Résumé quotidien automatique à 07h00 par email
✓ Buffer Redis pour gérer les pics de volume
✓ Conformité RGPD et déclaration CDP Sénégal documentées
✓ Route de suppression sur demande implémentée
✓ Validation humaine mensuelle de 100 tweets échantillonnés
✓ Versioning Git du prompt avec changelog
✓ Procédure escalade crise réelle vers direction documentée