963a65536f
This commit implements the full Eugen bot based on specifications in CLAUDE.md and eugen_claude.md. Features implemented: - Smart name recognition (@Eugen, Eugen:, etc.) - Persistent conversation memory per user (max 25 messages, 1 hour retention) - Perplexity Sonar API integration for AI responses - Live monitoring dashboard with PySimpleGUI - Setup wizard for first-time configuration - Comprehensive logging (main log + API debug log) Files added: - config.py: Configuration management from .env and config.json - utils.py: MentionDetector and Logger utility classes - memory.py: ConversationMemory for persistent chat history - ai_provider.py: PerplexityProvider for API integration - gui.py: Dashboard and SetupWizard GUI components - chatbot.py: Main EugenBot orchestrator with IRC handling - requirements.txt: Python dependencies - .env.example: Template for environment variables - .gitignore: Renamed from gitignore for proper Git usage Updated: - README.md: Complete usage instructions and documentation The bot is ready to use - users just need to add their API keys and run python chatbot.py
302 lines
9.6 KiB
Python
302 lines
9.6 KiB
Python
"""
|
|
Eugen - Intelligent Twitch Chat Bot
|
|
Main entry point and orchestrator
|
|
"""
|
|
import asyncio
|
|
import irc.bot
|
|
import irc.strings
|
|
import threading
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from config import Config
|
|
from memory import ConversationMemory
|
|
from ai_provider import PerplexityProvider
|
|
from gui import Dashboard, SetupWizard
|
|
from utils import MentionDetector, Logger
|
|
|
|
|
|
class EugenBot(irc.bot.SingleServerIRCBot):
|
|
"""Main bot class - orchestrates IRC, Memory, AI, and GUI"""
|
|
|
|
def __init__(self, config):
|
|
"""
|
|
Initialize the bot
|
|
|
|
Args:
|
|
config (Config): Configuration object
|
|
"""
|
|
self.config = config
|
|
self.bot_name = config.bot_name
|
|
self.channel = config.twitch_channel
|
|
|
|
# Initialize components
|
|
self.memory = ConversationMemory(
|
|
data_dir=config.data_dir,
|
|
retention_hours=config.context_retention_hours
|
|
)
|
|
self.ai = PerplexityProvider(
|
|
api_key=config.perplexity_key,
|
|
model=config.model,
|
|
max_tokens=config.max_tokens
|
|
)
|
|
self.detector = MentionDetector(bot_name=self.bot_name)
|
|
self.logger = Logger(log_dir=config.log_dir, debug_mode=config.debug_mode)
|
|
|
|
# Dashboard (will be initialized in GUI thread)
|
|
self.dashboard = None
|
|
|
|
# IRC connection setup
|
|
server = "irc.chat.twitch.tv"
|
|
port = 6667
|
|
nickname = self.bot_name
|
|
token = config.twitch_token
|
|
|
|
# Initialize IRC bot
|
|
irc.bot.SingleServerIRCBot.__init__(
|
|
self,
|
|
[(server, port, token)],
|
|
nickname,
|
|
nickname
|
|
)
|
|
|
|
self.is_running = False
|
|
self.loop = None
|
|
|
|
self.logger.info(f"Bot initialized: {nickname} → {self.channel}")
|
|
|
|
def on_welcome(self, connection, event):
|
|
"""Called when bot connects to IRC server"""
|
|
self.logger.info("Connected to Twitch IRC")
|
|
if self.dashboard:
|
|
self.dashboard.log_event("info", {"message": "Connected to Twitch IRC"})
|
|
|
|
# Request Twitch-specific capabilities
|
|
connection.cap("REQ", ":twitch.tv/membership")
|
|
connection.cap("REQ", ":twitch.tv/tags")
|
|
connection.cap("REQ", ":twitch.tv/commands")
|
|
|
|
# Join channel
|
|
connection.join(self.channel)
|
|
self.logger.info(f"Joined channel: {self.channel}")
|
|
if self.dashboard:
|
|
self.dashboard.log_event("info", {"message": f"Joined {self.channel}"})
|
|
|
|
def on_pubmsg(self, connection, event):
|
|
"""Called when a message is received in chat"""
|
|
# Extract username and message
|
|
username = event.source.nick
|
|
message = event.arguments[0]
|
|
|
|
self.logger.chat_message(username, message)
|
|
if self.dashboard:
|
|
self.dashboard.log_event("chat_message", {
|
|
"username": username,
|
|
"content": message
|
|
})
|
|
|
|
# Check if bot was mentioned
|
|
if self.detector.is_mentioned(message):
|
|
self.logger.debug(f"Mention detected from {username}")
|
|
if self.dashboard:
|
|
self.dashboard.log_event("mention_detected", {"username": username})
|
|
|
|
# Process in async context
|
|
asyncio.run_coroutine_threadsafe(
|
|
self.handle_mention(username, message),
|
|
self.loop
|
|
)
|
|
|
|
async def handle_mention(self, username, message):
|
|
"""
|
|
Handle a message where bot was mentioned
|
|
|
|
Args:
|
|
username (str): User who sent the message
|
|
message (str): Full message content
|
|
"""
|
|
# Extract actual content without mention
|
|
content = self.detector.extract_content(message)
|
|
|
|
if not content:
|
|
return
|
|
|
|
try:
|
|
# Load user's conversation history
|
|
history = self.memory.get_user_history(username, limit=5)
|
|
self.logger.debug(f"Loaded {len(history)} messages for {username}")
|
|
|
|
if self.dashboard:
|
|
self.dashboard.log_event("context_loaded", {
|
|
"username": username,
|
|
"count": len(history)
|
|
})
|
|
|
|
# Build messages for API
|
|
messages = [{"role": "system", "content": self.config.get_system_prompt()}]
|
|
|
|
# Add conversation history
|
|
if history:
|
|
messages.extend(self.memory.format_for_prompt(history))
|
|
|
|
# Add current message
|
|
messages.append({"role": "user", "content": content})
|
|
|
|
self.logger.api_call("chat.completions", self.config.model, len(messages))
|
|
if self.dashboard:
|
|
self.dashboard.log_event("api_call", {
|
|
"model": self.config.model,
|
|
"messages": len(messages)
|
|
})
|
|
|
|
# Get AI response
|
|
response = await self.ai.get_response(messages)
|
|
|
|
if response:
|
|
# Log and display response
|
|
self.logger.debug(f"Received response: {response[:100]}...")
|
|
if self.dashboard:
|
|
self.dashboard.log_event("api_response", {"content": response})
|
|
|
|
# Save to memory
|
|
self.memory.add_message(username, "user", content)
|
|
self.memory.add_message(username, "assistant", response)
|
|
|
|
# Send to chat
|
|
self.send_chat_message(username, response)
|
|
|
|
else:
|
|
error_msg = "Sorry, I couldn't process that right now."
|
|
self.logger.error("API call failed")
|
|
if self.dashboard:
|
|
self.dashboard.log_event("error", {"error": "API call failed"})
|
|
self.send_chat_message(username, error_msg)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error handling mention: {str(e)}")
|
|
if self.dashboard:
|
|
self.dashboard.log_event("error", {"error": str(e)})
|
|
|
|
def send_chat_message(self, username, response):
|
|
"""
|
|
Send a message to chat
|
|
|
|
Args:
|
|
username (str): User to address
|
|
response (str): Message content
|
|
"""
|
|
# Format message to address user
|
|
message = f"@{username} {response}"
|
|
|
|
# Send via IRC
|
|
self.connection.privmsg(self.channel, message)
|
|
|
|
self.logger.bot_response(username, response)
|
|
if self.dashboard:
|
|
self.dashboard.log_event("bot_response", {
|
|
"username": username,
|
|
"content": response
|
|
})
|
|
|
|
def start(self):
|
|
"""Start the bot"""
|
|
self.is_running = True
|
|
|
|
# Create event loop for async operations
|
|
self.loop = asyncio.new_event_loop()
|
|
|
|
# Start bot in separate thread
|
|
bot_thread = threading.Thread(target=self._run_bot, daemon=True)
|
|
bot_thread.start()
|
|
|
|
# Run dashboard in main thread
|
|
self.dashboard = Dashboard(self)
|
|
self.logger.info("Starting dashboard...")
|
|
self.dashboard.run()
|
|
|
|
def _run_bot(self):
|
|
"""Run the IRC bot (called in thread)"""
|
|
asyncio.set_event_loop(self.loop)
|
|
try:
|
|
self.logger.info("Starting IRC bot...")
|
|
super().start()
|
|
except Exception as e:
|
|
self.logger.error(f"Bot error: {str(e)}")
|
|
if self.dashboard:
|
|
self.dashboard.log_event("error", {"error": f"Bot crashed: {str(e)}"})
|
|
|
|
def stop(self):
|
|
"""Stop the bot"""
|
|
self.is_running = False
|
|
self.logger.info("Stopping bot...")
|
|
if self.dashboard:
|
|
self.dashboard.log_event("info", {"message": "Shutting down..."})
|
|
|
|
try:
|
|
self.connection.quit("Bot shutting down")
|
|
self.die()
|
|
except:
|
|
pass
|
|
|
|
if self.loop:
|
|
self.loop.stop()
|
|
|
|
|
|
def check_env_file():
|
|
"""Check if .env file exists, if not run setup wizard"""
|
|
env_path = Path(".env")
|
|
|
|
if not env_path.exists():
|
|
print("No .env file found. Running setup wizard...")
|
|
wizard = SetupWizard()
|
|
config = wizard.run()
|
|
|
|
if config:
|
|
# Create .env file
|
|
with open(".env", "w") as f:
|
|
for key, value in config.items():
|
|
f.write(f"{key}={value}\n")
|
|
print(".env file created successfully!")
|
|
return True
|
|
else:
|
|
print("Setup cancelled. Please create .env manually or run setup again.")
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def main():
|
|
"""Main entry point"""
|
|
print("""
|
|
╔════════════════════════════════════════════════════════════════╗
|
|
║ EUGEN TWITCH BOT ║
|
|
║ Gaming & 3D-Druck Chat Assistant ║
|
|
╚════════════════════════════════════════════════════════════════╝
|
|
""")
|
|
|
|
# Check for configuration
|
|
if not check_env_file():
|
|
return
|
|
|
|
# Load configuration
|
|
config = Config()
|
|
|
|
if not config.is_configured():
|
|
print("ERROR: Configuration incomplete!")
|
|
print("Please check your .env file or run setup wizard again.")
|
|
print("Delete .env to run setup wizard on next start.")
|
|
return
|
|
|
|
print(f"Bot Name: {config.bot_name}")
|
|
print(f"Channel: {config.twitch_channel}")
|
|
print(f"Model: {config.model}")
|
|
print(f"Debug Mode: {config.debug_mode}")
|
|
print("\nStarting bot...\n")
|
|
|
|
# Create and start bot
|
|
bot = EugenBot(config)
|
|
bot.start()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|