Full bot example#

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

# Load environment variables from .env file
load_dotenv()
# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger('whatsapp_chatgpt_python')

# Get environment variables
ID_INSTANCE = os.environ.get("INSTANCE_ID")
API_TOKEN = os.environ.get("INSTANCE_TOKEN")
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")

# Initialize the bot
bot = WhatsappGptBot(
id_instance=ID_INSTANCE,
api_token_instance=API_TOKEN,
openai_api_key=OPENAI_API_KEY,
model="gpt-4o", # Uses the GPT-4o model (which supports images)
system_message="You are a helpful assistant. Be brief and friendly in your responses.",
max_history_length=15, # Store the last 15 messages in the conversation history
temperature=0.7, # Slightly more creative responses
session_timeout=1800, # Sessions expire after 30 minutes of inactivity
error_message="Sorry, your message could not be processed.", # Custom error message
)

# Custom image handler with enhanced instructions
class EnhancedImageHandler(ImageMessageHandler):
async def process_message(self, notification, openai_client=None, model=None):
# Call parent class method to get base result
result = await super().process_message(notification, openai_client, model)

# For text responses (not visual models)
if isinstance(result, str):
return result.replace(
"[The user sent an image",
"[The user sent an image. Analyze what might be in it based on the caption"
)

# For models with image support, enhance the text instruction
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'] = "Describe this image and what you see in it in detail."

return result

# Custom text handler example
class EnhancedTextHandler(TextMessageHandler):
async def process_message(self, notification, *args, **kwargs):
# Get the text message using the parent class method
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 ['code', 'function', 'script', 'program']):
return f"🧑‍💻 CODE REQUEST: {text}\n\n[I will format my code response with proper syntax highlighting]"

elif text.endswith('?') or text.lower().startswith(
('what', 'why', 'how', 'when', 'where', 'who', 'can', 'could')):
return f"❓ QUESTION: {text}\n\n[I will provide a clear and comprehensive answer]"

return text

# Replace default handlers with our enhanced versions
bot.replace_handler(ImageMessageHandler, EnhancedImageHandler())
bot.replace_handler(TextMessageHandler, EnhancedTextHandler())

# Middleware for logging all messages and adding tracking information
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 = "complex content (likely contains media)"

logger.info(f"Message from {user_id}: {content_display}")

# Add tracking information to session context
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 unchanged content and messages
return {"message_content": message_content, "messages": messages}

# Middleware for formatting responses before sending to the user
def formatting_middleware(response, messages, session_data):
# Format the response by adding a signature to the end of long messages
formatted_response = response.strip()

# Do not add a signature to short responses
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 #{message_count} • Powered by GPT_"

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

# Adding middleware
bot.add_message_middleware(logging_middleware)
bot.add_response_middleware(formatting_middleware)


# Handler for the /clear command to clear the chat history
@bot.router.message(command="clear")
def clear_history_handler(notification):
chat_id = notification.chat

# Get session data
session_data = bot.get_session_data(chat_id)

# Find the system message, if it exists
system_message = None
for msg in session_data.messages:
if msg.get("role") == "system":
system_message = msg
break

# Clear messages, but keep the system message
if system_message:
session_data.messages = [system_message]
else:
session_data.messages = []

# Update the session
bot.update_session_data(chat_id, session_data)

notification.answer("🗑️ Chat history cleared! Let's get started again.")

# Handler for the /help command to display available commands
@bot.router.message(command="help")
def help_handler(notification):
help_text = (
"🤖 *WhatsApp GPT Bot* 🤖\n\n"
"Available commands:\n"
"• */help* - Show this help message\n"
"• */clear* - Clear the conversation history\n"
"• */info* - Show information about the bot\n"
"• */weather* - Example of a handler that skips GPT\n\n"
"You can send text, images, audio, and more. I will intelligently respond to your messages."
)
notification.answer(help_text)

# Adding the info command
@bot.router.message(command="info")
def info_handler(notification):
chat_id = notification.chat
session_data = bot.get_session_data(chat_id)

# Getting session statistics
message_count = len(session_data.messages) - 1 # Subtract the system message
if message_count < 0:
message_count = 0

vision_capable = "Yes" if bot.supports_images() else "No"

info_text = (
"📊 *Bot info* 📊\n\n"
f"Model: {bot.get_model()}\n"
f"Image support: {vision_capable}\n"
f"Messages in the current session: {message_count}\n"
f"Maximum history length: {bot.max_history_length}\n"
f"Session timeout: {bot.session_timeout} seconds\n\n"
"To clear the current conversation, use */clear*"
)
notification.answer(info_text)

# Example of weather handler skipping GPT processing
@bot.router.message(command="weather")
def weather_handler(notification):
notification.answer(
"🌤️ This is a stub for a weather response from a custom handler.\n\n"
"In a real bot, this would be a call to the weather API.\n\n"
"This handler demonstrates skipping GPT processing."
)

# Starting the bot
if __name__ == "__main__":
logger.info("Starting WhatsApp GPT Bot...")
logger.info(f"Model used: {bot.get_model()}")
bot.run_forever()