chatbot-maisonnette/chatbot.py

import os
import openai
import discord
import aiohttp
import asyncio
import base64
import logging

from dotenv import load_dotenv

# Charger les variables d'environnement depuis le fichier .env
load_dotenv()
DISCORD_TOKEN = os.getenv('DISCORD_TOKEN')
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')

# Vérifier que les tokens sont récupérés
if DISCORD_TOKEN is None or OPENAI_API_KEY is None:
    raise ValueError("Les tokens ne sont pas définis dans les variables d'environnement.")

# Log configuration
log_format='%(asctime)-13s : %(name)-20s : %(levelname)-8s : %(message)s'
logging.basicConfig(handlers=[logging.FileHandler("./chatbot.log", 'a', 'utf-8')], format=log_format, level="DEBUG")

console = logging.StreamHandler()
console.setLevel(logging.INFO)
console.setFormatter(logging.Formatter(log_format))

logger = logging.getLogger("chatbot")
logger.setLevel("DEBUG")

logging.getLogger('').addHandler(console)

# Initialiser les intents
intents = discord.Intents.default()
intents.message_content = True  # Activer l'intent pour les contenus de message

# Initialiser le client Discord avec les intents modifiés
client_discord = discord.Client(intents=intents)

# Initialiser l'API OpenAI avec un client
client_openai = openai.OpenAI(api_key=OPENAI_API_KEY)

# Dictionnaire pour stocker l'historique des conversations pour chaque utilisateur
conversation_history = {}

# L'ID du salon spécifique où le bot est autorisé à répondre
chatgpt_channel_id = 1284699709188997150  # Remplace par l'ID réel de ton salon

def calculate_cost(usage):
    input_tokens = usage.get('prompt_tokens', 0)
    output_tokens = usage.get('completion_tokens', 0)

    # Coûts estimés
    input_cost = input_tokens / 1_000_000 * 5.00  # 5$ pour 1M tokens d'entrée
    output_cost = output_tokens / 1_000_000 * 15.00  # 15$ pour 1M tokens de sortie
    total_cost = input_cost + output_cost

    return input_tokens, output_tokens, total_cost

async def read_text_file(attachment):
    # Télécharger et lire le contenu du fichier texte
    async with aiohttp.ClientSession() as session:
        async with session.get(attachment.url) as resp:
            return await resp.text()

async def encode_image_from_attachment(attachment):
    async with aiohttp.ClientSession() as session:
        async with session.get(attachment.url) as resp:
            image_data = await resp.read()
            return base64.b64encode(image_data).decode('utf-8')

async def call_openai_api(user_id, user_text, image_data=None):
    # Récupérer l'historique de la conversation pour l'utilisateur
    user_history = conversation_history.get(user_id, [])

    # Préparer le contenu de l'utilisateur
    user_content = [{"type": "text", "text": user_text}]
    if image_data:
        user_content.append({
            "type": "image_url",
            "image_url": {"url": f"data:image/jpeg;base64,{image_data}"}
        })

    # Ajouter le contenu à l'historique
    user_history.append({
        "role": "user",
        "content": user_content
    })

    payload = {
        "model": "gpt-4o",
        "messages": user_history,
        "max_tokens": 500,
        "stop": ["\n"]  # Arrête la réponse à la fin d'une phrase
    }

    headers = {
       "Content-Type": "application/json",
        "Authorization": f"Bearer {OPENAI_API_KEY}"
    }

    try:
        async with aiohttp.ClientSession() as session:
            async with session.post("https://api.openai.com/v1/chat/completions", headers=headers, json=payload) as resp:
                result = await resp.json()
                if resp.status != 200:
                    raise ValueError(f"API Error: {result.get('error', {}).get('message', 'Unknown error')}")

                # Calculer les coûts
                usage = result.get('usage', {})
                input_tokens, output_tokens, total_cost = calculate_cost(usage)

                # Afficher dans la console
                logger.info(f"Estimated Cost: ${total_cost:.4f} / Input Tokens: {input_tokens} / Output Tokens: {output_tokens} / Total Tokens: {input_tokens + output_tokens}")

                return result
    except Exception as e:
        logger.error(f"Error calling OpenAI API: {e}")
        return None

@client_discord.event
async def on_ready():
    logger.info(f'Bot connecté en tant que {client_discord.user}')

@client_discord.event
async def on_message(message):
    # Vérifier si le message provient du canal autorisé
    if message.channel.id != chatgpt_channel_id:
        return

    # Vérifier si l'auteur du message est le bot lui-même
    if message.author == client_discord.user:
        return

    user_text = message.content.strip()
    image_data = None
    file_content = None

    # Extensions de fichiers autorisées
    allowed_extensions = ['.txt', '.py', '.html', '.css', '.js']

    # Vérifier s'il y a une pièce jointe
    if message.attachments:
        for attachment in message.attachments:
            # Vérifier si c'est un fichier avec une extension autorisée
            if any(attachment.filename.endswith(ext) for ext in allowed_extensions):
                file_content = await read_text_file(attachment)
                break
            # Vérifier si c'est une image
            elif attachment.content_type.startswith('image/'):
                image_data = await encode_image_from_attachment(attachment)
                break

    # Ajouter le contenu du fichier à la requête si présent
    if file_content:
        user_text += f"\nContenu du fichier {attachment.filename}:\n{file_content}"

    # Appeler l'API OpenAI
    result = await call_openai_api(message.author.id, user_text, image_data)
    if result:
        reply = result['choices'][0]['message']['content']
        await message.channel.send(reply)

        # Ajouter la réponse du modèle à l'historique
        add_to_conversation_history(message.author.id, {
            "role": "assistant",
            "content": reply
        })

MAX_HISTORY_LENGTH = 50 # Nombre maximum de messages à conserver

def add_to_conversation_history(user_id, new_message):
    if user_id not in conversation_history:
        # Ajouter la personnalité de l'IA en tant que premier message
        conversation_history[user_id] = [
            {
                "role": "system",
                "content": "Tu es une utilisatrice Discord qui parle français."
                           "Tu ne dis jamais que tu es un IA."
                           "Tu ne peux pas poster de message de plusieurs lignes."
                           "Tu peux analyser des images et des pièces jointes aux formats txt, py, html, css et js."
                           "Tu es cynique et sarcastique."
                           "Tu apprécies les compliments."
                           "Réponds sans poser de questions finales comme 'Avez-vous besoin de quelque chose d'autre ?'"
                           "Privilégie des réponses courtes et percutantes."
                           "Tu évites de générer des réponses de plus de 500 tokens."
            }
        ]

    conversation_history[user_id].append(new_message)

    # Limiter la taille de l'historique
    if len(conversation_history[user_id]) > MAX_HISTORY_LENGTH:
        # Garder le premier message de personnalité et les messages les plus récents
        conversation_history[user_id] = conversation_history[user_id][:1] + conversation_history[user_id][-MAX_HISTORY_LENGTH:]

# Démarrer le bot Discord
client_discord.run(DISCORD_TOKEN)