Полный пример бота#

import os
import time
import logging
from whatsapp_chatgpt_python import (
    WhatsappGptBot,
    ImageMessageHandler,
    TextMessageHandler
)
from dotenv import load_dotenv

# Загрузка переменных окружения из файла .env
load_dotenv()
# Настройка логирования
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger('whatsapp_chatgpt_python')

# Получение переменных окружения
ID_INSTANCE = os.environ.get("INSTANCE_ID")
API_TOKEN = os.environ.get("INSTANCE_TOKEN")
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")

# Инициализация бота
bot = WhatsappGptBot(
    id_instance=ID_INSTANCE,
    api_token_instance=API_TOKEN,
    openai_api_key=OPENAI_API_KEY,
    model="gpt-4o",  # Использует модель GPT-4o (которая поддерживает изображения)
    system_message="Вы - полезный ассистент. Будьте краткими и дружелюбными в своих ответах.",
    max_history_length=15,  # Хранить последние 15 сообщений в истории разговора
    temperature=0.7,  # Немного более творческие ответы
    session_timeout=1800,  # Сессии истекают после 30 минут неактивности
    error_message="Извините, ваше сообщение не удалось обработать.",  # Пользовательское сообщение об ошибке
)


# Пользовательский обработчик изображений с расширенными инструкциями
class EnhancedImageHandler(ImageMessageHandler):
    async def process_message(self, notification, openai_client=None, model=None):
        # Вызов метода родительского класса для получения базового результата
        result = await super().process_message(notification, openai_client, model)

        # Для текстовых ответов (не визуальные модели)
        if isinstance(result, str):
            return result.replace(
                "[The user sent an image",
                "[Пользователь отправил изображение. Проанализируйте, что может быть на нем, основываясь на подписи"
            )

        # Для моделей с поддержкой изображений, улучшаем текстовую инструкцию
        if isinstance(result, list) and len(result) > 0 and isinstance(result[0], dict):
            if result[0].get('type') == 'text':
                text = result[0].get('text', '')
                if text == "Analyzing this image":
                    result[0]['text'] = "Подробно опишите это изображение и то, что вы на нем видите."

        return result


# Пример пользовательского текстового обработчика
class EnhancedTextHandler(TextMessageHandler):
    async def process_message(self, notification, *args, **kwargs):
        # Получение текстового сообщения с помощью метода родительского класса
        text = await super().process_message(notification, *args, **kwargs)

        if not text:
            return text

        lower_text = text.lower()

        if any(term in lower_text for term in ['код', 'функция', 'скрипт', 'программа']):
            return f"🧑‍💻 ЗАПРОС НА КОД: {text}\n\n[Я отформатирую мой ответ с кодом с правильной подсветкой синтаксиса]"

        elif text.endswith('?') or text.lower().startswith(
                ('что', 'почему', 'как', 'когда', 'где', 'кто', 'можно', 'мог')):
            return f"❓ ВОПРОС: {text}\n\n[Я предоставлю четкий и исчерпывающий ответ]"

        return text


# Замена обработчиков по умолчанию на наши расширенные версии
bot.replace_handler(ImageMessageHandler, EnhancedImageHandler())
bot.replace_handler(TextMessageHandler, EnhancedTextHandler())


# Промежуточное ПО для логирования всех сообщений и добавления информации о трекинге
def logging_middleware(notification, message_content, messages, session_data):
    user_id = notification.sender
    if isinstance(message_content, str):
        content_display = message_content[:100] + "..." if len(message_content) > 100 else message_content
    else:
        content_display = "сложный контент (вероятно, содержит медиа)"

    logger.info(f"Сообщение от {user_id}: {content_display}")

    # Добавление информации о трекинге в контекст сессии
    if not session_data.context.get("variables"):
        session_data.context["variables"] = {}

    session_data.context["variables"].update({
        "last_interaction": int(time.time()),
        "message_count": session_data.context.get("variables", {}).get("message_count", 0) + 1
    })

    # Возвращаем неизмененный контент и сообщения
    return {"message_content": message_content, "messages": messages}


# Промежуточное ПО для форматирования ответов перед отправкой пользователю
def formatting_middleware(response, messages, session_data):
    # Форматируем ответ, добавляя подпись в конце длинных сообщений
    formatted_response = response.strip()

    # Не добавляем подпись к коротким ответам
    if len(formatted_response) > 100 and not formatted_response.endswith("_"):
        message_count = session_data.context.get("variables", {}).get("message_count", 0)
        formatted_response += f"\n\n_Сообщение #{message_count} • Работает на GPT_"

    return {"response": formatted_response, "messages": messages}


# Добавление промежуточного ПО
bot.add_message_middleware(logging_middleware)
bot.add_response_middleware(formatting_middleware)


# Обработчик команды /clear для сброса истории разговора
@bot.router.message(command="clear")
def clear_history_handler(notification):
    chat_id = notification.chat

    # Получение данных сессии
    session_data = bot.get_session_data(chat_id)

    # Поиск системного сообщения, если оно существует
    system_message = None
    for msg in session_data.messages:
        if msg.get("role") == "system":
            system_message = msg
            break

    # Сброс сообщений, но сохранение системного сообщения
    if system_message:
        session_data.messages = [system_message]
    else:
        session_data.messages = []

    # Обновление сессии
    bot.update_session_data(chat_id, session_data)

    notification.answer("🗑️ История разговора очищена! Начнем заново.")


# Обработчик команды /help для отображения доступных команд
@bot.router.message(command="help")
def help_handler(notification):
    help_text = (
        "🤖 *WhatsApp GPT Бот* 🤖\n\n"
        "Доступные команды:\n"
        "• */help* - Показать это сообщение помощи\n"
        "• */clear* - Очистить историю разговора\n"
        "• */info* - Показать информацию о боте\n"
        "• */weather* - Пример обработчика, пропускающего GPT\n\n"
        "Вы можете отправлять текст, изображения, аудио и многое другое. Я буду интеллектуально отвечать на ваши сообщения."
    )
    notification.answer(help_text)


# Добавление команды info
@bot.router.message(command="info")
def info_handler(notification):
    chat_id = notification.chat
    session_data = bot.get_session_data(chat_id)

    # Получение статистики сессии
    message_count = len(session_data.messages) - 1  # Вычитаем системное сообщение
    if message_count < 0:
        message_count = 0

    vision_capable = "Да" if bot.supports_images() else "Нет"

    info_text = (
        "📊 *Информация о боте* 📊\n\n"
        f"Модель: {bot.get_model()}\n"
        f"Поддержка изображений: {vision_capable}\n"
        f"Сообщений в текущей сессии: {message_count}\n"
        f"Максимальная длина истории: {bot.max_history_length}\n"
        f"Таймаут сессии: {bot.session_timeout} секунд\n\n"
        "Чтобы очистить текущий разговор, используйте */clear*"
    )
    notification.answer(info_text)


# Пример обработчика погоды, пропускающего обработку GPT
@bot.router.message(command="weather")
def weather_handler(notification):
    notification.answer(
        "🌤️ Это заглушка ответа о погоде от пользовательского обработчика.\n\n"
        "В реальном боте здесь было бы обращение к API погоды.\n\n"
        "Этот обработчик демонстрирует пропуск обработки GPT."
    )


# Запуск бота
if __name__ == "__main__":
    logger.info("Запуск WhatsApp GPT Бота...")
    logger.info(f"Используемая модель: {bot.get_model()}")
    bot.run_forever()