Avatar IA : répondre à sa place sur les réseaux sociaux

┌─────────────────────────────────────────────┐
│              AVATAR IA SOCIAL                │
├─────────────────────────────────────────────┤
│                                             │
│  ┌─────────┐  ┌──────────┐  ┌───────────┐  │
│  │  Veille  │→│ Génération│→│  Review    │  │
│  │  & Feed  │  │ Contenu  │  │ & Filtres │  │
│  └─────────┘  └──────────┘  └─────┬─────┘  │
│                                    │        │
│  ┌─────────┐  ┌──────────┐  ┌─────▼─────┐  │
│  │ Mémoire │  │   Ton    │  │Publication│  │
│  │Persistante│ │Personnel │  │  Multi-   │  │
│  │ (Vector) │  │ (Style) │  │ Plateforme│  │
│  └─────────┘  └──────────┘  └───────────┘  │
│                                             │
│  ┌─────────────────────────────────────┐    │
│  │     APIs : X / LinkedIn / IG / TG   │    │
│  └─────────────────────────────────────┘    │
└─────────────────────────────────────────────┘

import anthropic

client = anthropic.Anthropic(api_key="sk-...")

def generate_post(platform: str, topic: str, tone_guide: str, context: str) -> str:
    """Génère un post adapté à la plateforme."""

    system_prompt = f"""Tu es l'avatar IA de [Nom]. Tu écris EXACTEMENT comme lui/elle.

GUIDE DE TON :
{tone_guide}

PLATEFORME : {platform}
RÈGLES SPÉCIFIQUES :
- LinkedIn : professionnel, insights, 1200-1500 caractères max
- Twitter/X : punchy, opinion tranchée, 280 caractères max
- Instagram : visuel, émotionnel, hashtags pertinents
- Telegram : informel, direct, liens utiles

CONTEXTE RÉCENT (mémoire) :
{context}

INTERDIT : politique clivante, attaques personnelles, promesses commerciales non validées.
"""

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        system=system_prompt,
        messages=[{"role": "user", "content": f"Écris un post {platform} sur : {topic}"}]
    )

    return response.content[0].text

import chromadb
from datetime import datetime

class AvatarMemory:
    def __init__(self, db_path: str = "./avatar_memory"):
        self.client = chromadb.PersistentClient(path=db_path)
        self.posts = self.client.get_or_create_collection("posts_history")
        self.positions = self.client.get_or_create_collection("positions")
        self.conversations = self.client.get_or_create_collection("conversations")

    def remember_post(self, platform: str, content: str, engagement: dict):
        """Stocke un post et son engagement pour apprentissage."""
        self.posts.add(
            documents=[content],
            metadatas=[{
                "platform": platform,
                "date": datetime.now().isoformat(),
                "likes": engagement.get("likes", 0),
                "comments": engagement.get("comments", 0),
                "shares": engagement.get("shares", 0),
            }],
            ids=[f"{platform}_{datetime.now().timestamp()}"]
        )

    def get_context(self, topic: str, n_results: int = 5) -> str:
        """Récupère le contexte pertinent pour un sujet."""
        results = self.posts.query(query_texts=[topic], n_results=n_results)
        positions = self.positions.query(query_texts=[topic], n_results=3)

        context_parts = []
        for doc, meta in zip(results["documents"][0], results["metadatas"][0]):
            context_parts.append(
                f"[{meta['platform']} - {meta['date'][:10]}] {doc[:200]}..."
            )

        for doc in positions["documents"][0]:
            context_parts.append(f"[POSITION] {doc}")

        return "\n".join(context_parts)

    def add_position(self, topic: str, position: str):
        """Enregistre une position sur un sujet."""
        self.positions.add(
            documents=[f"{topic}: {position}"],
            metadatas=[{"topic": topic, "date": datetime.now().isoformat()}],
            ids=[f"pos_{topic.replace(' ', '_')}"]
        )

TONE_PROFILES = {
    "linkedin": {
        "style": "professionnel, réfléchi, orienté valeur",
        "longueur": "800-1500 caractères",
        "structure": "hook → insight → exemple → call-to-action",
        "vocabulaire": "business, innovation, croissance, stratégie",
        "interdits": "argot, emojis excessifs, polémique",
        "exemple": "J'ai découvert quelque chose de contre-intuitif sur l'IA...",
    },
    "twitter": {
        "style": "punchy, opinion tranchée, conversationnel",
        "longueur": "140-280 caractères",
        "structure": "affirmation forte OU question provocante",
        "vocabulaire": "direct, moderne, parfois argotique",
        "interdits": "jargon corporate, texte mur",
        "exemple": "Les gens qui disent que l'IA va tout remplacer n'ont jamais essayé de lui faire écrire un email correct.",
    },
    "instagram": {
        "style": "inspirant, visuel, storytelling",
        "longueur": "300-600 caractères + hashtags",
        "structure": "émotion → histoire → leçon → hashtags",
        "vocabulaire": "créatif, émotionnel, accessible",
        "interdits": "trop technique, liens (pas cliquables)",
        "exemple": "Ce matin, j'ai laissé mon IA gérer mes emails pendant 2h. Résultat ? 👇",
    },
    "telegram": {
        "style": "direct, informel, utile",
        "longueur": "200-1000 caractères",
        "structure": "info brute → contexte → lien/ressource",
        "vocabulaire": "tech-friendly, décontracté",
        "interdits": "formalisme excessif",
        "exemple": "Nouveau truc cool : Claude peut maintenant...",
    },
}

def build_tone_prompt(platform: str, custom_traits: dict = None) -> str:
    """Construit le prompt de ton pour une plateforme."""
    profile = TONE_PROFILES[platform]

    prompt = f"""TON POUR {platform.upper()} :
- Style : {profile['style']}
- Longueur cible : {profile['longueur']}
- Structure : {profile['structure']}
- Vocabulaire : {profile['vocabulaire']}
- À éviter : {profile['interdits']}
- Exemple de ton : "{profile['exemple']}"
"""

    if custom_traits:
        prompt += f"\nTRAITS PERSONNELS :\n"
        for trait, value in custom_traits.items():
            prompt += f"- {trait} : {value}\n"

    return prompt

class CommentResponder:
    def __init__(self, llm_client, memory: AvatarMemory, tone_profiles: dict):
        self.llm = llm_client
        self.memory = memory
        self.tones = tone_profiles

    def classify_comment(self, comment: str) -> dict:
        """Classifie un commentaire pour décider de la réponse."""
        response = self.llm.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=256,
            messages=[{
                "role": "user",
                "content": f"""Classifie ce commentaire :
"{comment}"

Réponds en JSON :
{{
  "intent": "question|compliment|critique|spam|troll|demande_contact",
  "sentiment": "positif|neutre|négatif",
  "urgence": "haute|moyenne|basse",
  "necessite_humain": true/false,
  "raison_humain": "..."
}}"""
            }]
        )
        import json
        return json.loads(response.content[0].text)

    def generate_reply(self, platform: str, original_post: str, 
                       comment: str, classification: dict) -> str | None:
        """Génère une réponse ou renvoie None si humain requis."""

        # Garde-fou : certains cas nécessitent un humain
        if classification["necessite_humain"]:
            self.notify_human(comment, classification["raison_humain"])
            return None

        # Pas de réponse aux trolls/spam
        if classification["intent"] in ("spam", "troll"):
            return None

        tone = build_tone_prompt(platform)
        context = self.memory.get_context(original_post)

        response = self.llm.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=512,
            system=f"""Tu réponds à un commentaire en tant que [Nom].
{tone}
Contexte de tes posts récents : {context}
Sois authentique, bref, et utile.""",
            messages=[{
                "role": "user", 
                "content": f"Post original : {original_post}\nCommentaire : {comment}\nIntent : {classification['intent']}"
            }]
        )

        return response.content[0].text

    def notify_human(self, comment: str, reason: str):
        """Notifie l'humain quand l'avatar ne doit pas répondre seul."""
        # Envoi via Telegram, email, ou webhook
        print(f"🚨 HUMAIN REQUIS : {reason}\nCommentaire : {comment}")

class SocialWatcher:
    """Surveille les tendances et mentions pertinentes."""

    def __init__(self, topics: list[str], competitors: list[str]):
        self.topics = topics
        self.competitors = competitors

    async def scan_trends(self) -> list[dict]:
        """Scanne les tendances sur chaque plateforme."""
        trends = []

        # Twitter/X : trending topics + mentions
        x_trends = await self.scan_x_trends()
        trends.extend(x_trends)

        # LinkedIn : posts viraux dans le secteur
        li_trends = await self.scan_linkedin_feed()
        trends.extend(li_trends)

        # RSS + newsletters : veille sectorielle
        rss_items = await self.scan_rss_feeds()
        trends.extend(rss_items)

        return self.rank_by_relevance(trends)

    def rank_by_relevance(self, trends: list[dict]) -> list[dict]:
        """Classe les tendances par pertinence pour l'avatar."""
        # Utilise le LLM pour scorer la pertinence
        # Score basé sur : alignement thématique, potentiel engagement, timing
        return sorted(trends, key=lambda t: t.get("relevance_score", 0), reverse=True)

class ContentFilter:
    """Filtre de sécurité avant publication."""

    BLACKLIST_TOPICS = [
        "politique partisane",
        "religion",
        "santé (conseils médicaux)",
        "finance (conseils investissement)",
        "attaques personnelles",
        "données confidentielles",
    ]

    MAX_POSTS_PER_DAY = {
        "twitter": 5,
        "linkedin": 2,
        "instagram": 1,
        "telegram": 3,
    }

    def check(self, post: str, platform: str, daily_count: int) -> dict:
        """Vérifie qu'un post est safe à publier."""
        issues = []

        # Limite quotidienne
        if daily_count >= self.MAX_POSTS_PER_DAY[platform]:
            issues.append(f"Limite quotidienne atteinte ({daily_count}/{self.MAX_POSTS_PER_DAY[platform]})")

        # Vérification LLM des sujets interdits
        safety_check = self.llm_safety_check(post)
        if not safety_check["safe"]:
            issues.append(f"Sujet sensible détecté : {safety_check['reason']}")

        # Détection de contenu trop "IA"
        ai_score = self.detect_ai_patterns(post)
        if ai_score > 0.7:
            issues.append(f"Contenu trop 'IA' (score: {ai_score:.0%}) — reformuler")

        return {
            "approved": len(issues) == 0,
            "issues": issues,
            "post": post,
        }

    def detect_ai_patterns(self, text: str) -> float:
        """Détecte les patterns typiques du contenu IA."""
        red_flags = [
            "il est important de noter",
            "dans le paysage actuel",
            "en fin de compte",
            "il convient de souligner",
            "force est de constater",
            "plongeons dans",
        ]
        score = sum(1 for flag in red_flags if flag.lower() in text.lower())
        return min(score / 3, 1.0)

class MultiPlatformPublisher:
    """Publie sur toutes les plateformes configurées."""

    def __init__(self, credentials: dict):
        self.platforms = {}
        if "twitter" in credentials:
            self.platforms["twitter"] = TwitterAPI(credentials["twitter"])
        if "linkedin" in credentials:
            self.platforms["linkedin"] = LinkedInAPI(credentials["linkedin"])
        # ... etc

    async def publish(self, post: str, platform: str, 
                      schedule_time: datetime = None) -> dict:
        """Publie ou programme un post."""
        api = self.platforms[platform]

        if schedule_time:
            return await api.schedule(post, schedule_time)

        result = await api.post(post)

        return {
            "platform": platform,
            "post_id": result["id"],
            "url": result["url"],
            "published_at": datetime.now().isoformat(),
        }

# guardrails.yaml
safety:
  max_posts_per_day:
    twitter: 5
    linkedin: 2
    instagram: 1
    telegram: 3

  forbidden_topics:
    - politique
    - religion
    - santé/médical
    - finance/investissement
    - vie privée de tiers

  require_human_approval:
    - réponse à une critique
    - DM avec un inconnu
    - contenu mentionnant un concurrent
    - tout post avec un montant en euros/dollars

  auto_approved:
    - remerciement à un compliment
    - partage d'article avec commentaire
    - réponse FAQ (base de connaissances)

  escalation:
    channel: telegram  # où notifier l'humain
    timeout_minutes: 60  # si pas de réponse, ne pas publier

def humanize_post(post: str, personal_tics: list[str]) -> str:
    """Ajoute des touches humaines au contenu généré."""
    import random

    # Parfois ajouter un tic de langage personnel
    if random.random() > 0.6 and personal_tics:
        tic = random.choice(personal_tics)
        post = f"{tic} {post}" if random.random() > 0.5 else f"{post} {tic}"

    # Parfois raccourcir un mot (le style SMS modéré)
    replacements = {"quelque chose": "un truc", "beaucoup": "pas mal"}
    if random.random() > 0.7:
        for formal, casual in replacements.items():
            if formal in post.lower():
                post = post.replace(formal, casual, 1)
                break

    return post