From 963a65536f06ee9ea071d814adebfe80d60df9ff Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Jan 2026 11:18:40 +0000 Subject: [PATCH] Implement complete Eugen Twitch chatbot 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 --- .env.example | 19 +++ gitignore => .gitignore | 0 README.md | 155 ++++++++++++++++++++- ai_provider.py | 161 +++++++++++++++++++++ chatbot.py | 302 ++++++++++++++++++++++++++++++++++++++++ config.py | 96 +++++++++++++ gui.py | 271 +++++++++++++++++++++++++++++++++++ memory.py | 174 +++++++++++++++++++++++ requirements.txt | 5 + utils.py | 169 ++++++++++++++++++++++ 10 files changed, 1350 insertions(+), 2 deletions(-) create mode 100644 .env.example rename gitignore => .gitignore (100%) create mode 100644 ai_provider.py create mode 100644 chatbot.py create mode 100644 config.py create mode 100644 gui.py create mode 100644 memory.py create mode 100644 requirements.txt create mode 100644 utils.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..6fc60f3 --- /dev/null +++ b/.env.example @@ -0,0 +1,19 @@ +# Twitch Configuration +TWITCH_OAUTH_TOKEN=oauth:your_token_here +TWITCH_BOT_NICKNAME=Eugen +TWITCH_CHANNEL=#your_channel_here + +# Perplexity Configuration +PERPLEXITY_API_KEY=pplx-your_key_here +PERPLEXITY_MODEL=sonar-pro +MAX_TOKENS=450 + +# Bot Configuration +DEBUG_MODE=true +AUTO_RECONNECT=true +RECONNECT_DELAY=10 + +# Data Configuration +DATA_DIR=data/conversations +LOG_DIR=logs +CONTEXT_RETENTION_HOURS=1 diff --git a/gitignore b/.gitignore similarity index 100% rename from gitignore rename to .gitignore diff --git a/README.md b/README.md index c88e0a8..a9299cc 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,153 @@ -# KI-Chat-Bot-Eugen -Twitch KI Chat Bot +# Eugen - Intelligent Twitch Chat Bot + +An AI-powered Twitch chat bot specializing in Gaming and 3D Printing topics, powered by Perplexity Sonar API. + +## Features + +- **Smart Name Recognition**: Automatically responds when mentioned (@Eugen, Eugen:, etc.) +- **Persistent Memory**: Maintains conversation history per user (max 25 messages) +- **Context-Aware**: Remembers previous conversations for up to 1 hour +- **Perplexity Integration**: Uses Perplexity Sonar API for intelligent, real-time responses +- **Live Dashboard**: GUI monitoring interface showing all activity in real-time +- **Topics**: Gaming (WoW, Elden Ring), 3D Printing (Prusa, Bambu, Creality), Tech (Python, Linux) + +## Quick Start + +### Prerequisites + +- Python 3.9+ (3.11+ recommended) +- Twitch account for the bot +- Perplexity API key + +### Installation + +```bash +# Clone the repository +git clone https://github.com/Kenearos/KI-Chat-Bot-Eugen.git +cd KI-Chat-Bot-Eugen + +# Create virtual environment +python -m venv venv + +# Activate virtual environment +# On Windows: +.\venv\Scripts\Activate.ps1 +# On Linux/Mac: +source venv/bin/activate + +# Install dependencies +pip install -r requirements.txt +``` + +### Configuration + +#### Option 1: Setup Wizard (Recommended) +Simply run the bot and the setup wizard will guide you: +```bash +python chatbot.py +``` + +#### Option 2: Manual Configuration +Copy `.env.example` to `.env` and fill in your credentials: +```bash +cp .env.example .env +``` + +Edit `.env` with your values: +``` +TWITCH_OAUTH_TOKEN=oauth:your_token_here +TWITCH_BOT_NICKNAME=Eugen +TWITCH_CHANNEL=#your_channel_here +PERPLEXITY_API_KEY=pplx-your_key_here +``` + +#### Getting API Keys + +**Twitch OAuth Token:** +1. Visit [twitchtokengenerator.com](https://twitchtokengenerator.com/) +2. Generate token with `chat:read` and `chat:edit` scopes +3. Copy the token (should start with `oauth:`) + +**Perplexity API Key:** +1. Visit [perplexity.ai/api](https://www.perplexity.ai/api) +2. Sign up/Login +3. Generate new API key +4. Copy the key (should start with `pplx-`) + +### Running the Bot + +```bash +python chatbot.py +``` + +The dashboard will open automatically showing live activity. + +## Usage + +Once running, the bot responds when mentioned in Twitch chat: + +``` +User: @Eugen what's the best class in WoW? +Eugen: @User The best class depends on your playstyle... + +User: Eugen, how do I level my 3D printer? +Eugen: @User For bed leveling, start by... +``` + +## Project Structure + +``` +eugen/ +├── chatbot.py # Main entry point +├── config.py # Configuration management +├── gui.py # Dashboard GUI +├── ai_provider.py # Perplexity API integration +├── memory.py # Conversation memory +├── utils.py # Helper functions +├── requirements.txt # Python dependencies +├── .env # Your secrets (DO NOT COMMIT) +├── data/ +│ └── conversations/ # User chat histories +└── logs/ + ├── eugen.log # Main log + └── api_debug.log # API debug logs +``` + +## Documentation + +- **CLAUDE.md**: Quick reference guide for Claude Code +- **eugen_claude.md**: Detailed German documentation with architecture and implementation details + +## Development + +The bot is built with: +- Python 3.9+ +- Perplexity Sonar API for AI responses +- Twitch IRC for chat integration +- PySimpleGUI for dashboard +- Async/await for concurrent operations + +## Troubleshooting + +**Bot doesn't respond:** +- Check that .env has correct OAuth token and API key +- Verify bot is in the correct channel +- Check logs/eugen.log for errors + +**API errors:** +- Verify Perplexity API key is valid +- Check your API credits at perplexity.ai +- Enable DEBUG_MODE=true in .env for detailed logs + +**IRC connection failed:** +- Verify internet connection +- Check firewall settings for port 6667 +- Regenerate Twitch OAuth token if expired + +## License + +See LICENSE file for details. + +## Contributing + +Contributions welcome! Please open an issue or pull request. diff --git a/ai_provider.py b/ai_provider.py new file mode 100644 index 0000000..60b7475 --- /dev/null +++ b/ai_provider.py @@ -0,0 +1,161 @@ +""" +Perplexity API Provider for Eugen Bot +Handles communication with Perplexity Sonar API +""" +import httpx +import json +import time +from typing import List, Dict, Optional + + +class PerplexityProvider: + """Communicates with Perplexity API for AI responses""" + + def __init__(self, api_key, model="sonar-pro", max_tokens=450, temperature=0.7): + """ + Initialize Perplexity API provider + + Args: + api_key (str): Perplexity API key + model (str): Model to use (default: sonar-pro) + max_tokens (int): Maximum tokens in response + temperature (float): Sampling temperature + """ + self.api_key = api_key + self.model = model + self.max_tokens = max_tokens + self.temperature = temperature + self.base_url = "https://api.perplexity.ai" + + # Statistics + self.total_requests = 0 + self.total_tokens = 0 + self.total_errors = 0 + self.last_response_time = 0 + + async def get_response(self, messages): + """ + Send messages to Perplexity API and get response + + Args: + messages (list): List of message dicts with 'role' and 'content' + Example: [ + {"role": "system", "content": "You are..."}, + {"role": "user", "content": "Hello"} + ] + + Returns: + str: AI response content + None: If error occurred + """ + headers = { + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json" + } + + payload = { + "model": self.model, + "messages": messages, + "max_tokens": self.max_tokens, + "temperature": self.temperature + } + + start_time = time.time() + + try: + async with httpx.AsyncClient(timeout=30.0) as client: + response = await client.post( + f"{self.base_url}/chat/completions", + json=payload, + headers=headers + ) + + self.last_response_time = time.time() - start_time + + if response.status_code == 200: + data = response.json() + content = data['choices'][0]['message']['content'] + tokens_used = data.get('usage', {}).get('total_tokens', 0) + + # Update statistics + self.total_requests += 1 + self.total_tokens += tokens_used + + return content + else: + self.total_errors += 1 + error_msg = f"API Error {response.status_code}: {response.text}" + print(error_msg) + return None + + except httpx.TimeoutException: + self.total_errors += 1 + print("API Timeout: Request took too long") + return None + except Exception as e: + self.total_errors += 1 + print(f"API Error: {str(e)}") + return None + + async def validate_api_key(self): + """ + Validate API key with a simple test request + + Returns: + bool: True if API key is valid + """ + test_messages = [ + {"role": "user", "content": "test"} + ] + + try: + async with httpx.AsyncClient(timeout=5.0) as client: + response = await client.post( + f"{self.base_url}/chat/completions", + json={ + "model": self.model, + "messages": test_messages, + "max_tokens": 10 + }, + headers={ + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json" + } + ) + return response.status_code == 200 + except Exception: + return False + + def get_statistics(self): + """ + Get API usage statistics + + Returns: + dict: Statistics including requests, tokens, errors, avg response time + """ + avg_response_time = self.last_response_time if self.total_requests > 0 else 0 + success_rate = ( + ((self.total_requests - self.total_errors) / self.total_requests * 100) + if self.total_requests > 0 + else 0 + ) + + # Rough cost calculation (example pricing, adjust as needed) + # Perplexity pricing varies, this is an estimate + estimated_cost = self.total_tokens * 0.0003 / 1000 # $0.0003 per 1K tokens + + return { + "total_requests": self.total_requests, + "total_tokens": self.total_tokens, + "total_errors": self.total_errors, + "avg_response_time": avg_response_time, + "success_rate": success_rate, + "estimated_cost": estimated_cost + } + + def reset_statistics(self): + """Reset all statistics counters""" + self.total_requests = 0 + self.total_tokens = 0 + self.total_errors = 0 + self.last_response_time = 0 diff --git a/chatbot.py b/chatbot.py new file mode 100644 index 0000000..167e508 --- /dev/null +++ b/chatbot.py @@ -0,0 +1,302 @@ +""" +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() diff --git a/config.py b/config.py new file mode 100644 index 0000000..54adda8 --- /dev/null +++ b/config.py @@ -0,0 +1,96 @@ +""" +Configuration Management for Eugen Bot +Loads settings from .env and config.json +""" +import os +import json +from pathlib import Path +from dotenv import load_dotenv + + +class Config: + """Manages bot configuration from environment variables and config files""" + + def __init__(self, env_file=".env", config_file="data/config.json"): + # Load .env file + load_dotenv(env_file) + + # Twitch Configuration + self.twitch_token = os.getenv("TWITCH_OAUTH_TOKEN", "") + self.twitch_channel = os.getenv("TWITCH_CHANNEL", "") + self.bot_name = os.getenv("TWITCH_BOT_NICKNAME", "Eugen") + + # Perplexity Configuration + self.perplexity_key = os.getenv("PERPLEXITY_API_KEY", "") + self.model = os.getenv("PERPLEXITY_MODEL", "sonar-pro") + self.max_tokens = int(os.getenv("MAX_TOKENS", "450")) + + # Bot Behavior + self.debug_mode = os.getenv("DEBUG_MODE", "false").lower() == "true" + self.auto_reconnect = os.getenv("AUTO_RECONNECT", "true").lower() == "true" + self.reconnect_delay = int(os.getenv("RECONNECT_DELAY", "10")) + + # Data Configuration + self.data_dir = os.getenv("DATA_DIR", "data/conversations") + self.log_dir = os.getenv("LOG_DIR", "logs") + self.context_retention_hours = int(os.getenv("CONTEXT_RETENTION_HOURS", "1")) + + # Additional settings from config.json + self.config_file = config_file + self._load_json_config() + + def _load_json_config(self): + """Load additional settings from config.json if it exists""" + if os.path.exists(self.config_file): + try: + with open(self.config_file, 'r', encoding='utf-8') as f: + config_data = json.load(f) + # Override or extend settings from JSON + for key, value in config_data.items(): + if not hasattr(self, key): + setattr(self, key, value) + except Exception as e: + print(f"Warning: Could not load config.json: {e}") + + def save_to_json(self): + """Save current configuration to config.json""" + Path(self.config_file).parent.mkdir(parents=True, exist_ok=True) + config_data = { + "bot_name": self.bot_name, + "model": self.model, + "max_tokens": self.max_tokens, + "debug_mode": self.debug_mode, + "auto_reconnect": self.auto_reconnect, + "reconnect_delay": self.reconnect_delay, + "context_retention_hours": self.context_retention_hours + } + with open(self.config_file, 'w', encoding='utf-8') as f: + json.dump(config_data, f, indent=2, ensure_ascii=False) + + def is_configured(self): + """Check if all required settings are present""" + required = [ + self.twitch_token, + self.twitch_channel, + self.perplexity_key, + self.bot_name + ] + return all(required) and self.twitch_token.startswith("oauth:") + + def get_system_prompt(self): + """Returns the system prompt for the AI""" + return """Du bist Eugen, ein hilfreicher und freundlicher Twitch-Chat-Bot. + +Du bist Experte für folgende Themen: +- Gaming: World of Warcraft, Elden Ring, Gamedev +- 3D-Druck: Prusa i3, Bambu Labs, Creality +- Tech: Python, Linux, Home Automation + +Verhalte dich wie ein echter Chat-Teilnehmer: +- Sei freundlich und hilfsbereit +- Antworte kurz und prägnant (max 2-3 Sätze für Twitch-Chat) +- Verwende gelegentlich Gaming- oder Tech-Slang +- Beziehe dich auf vorherige Gespräche wenn möglich +- Sei humorvoll aber respektvoll + +Wenn du etwas nicht weißt, sage es ehrlich. Wenn die Frage nicht zu deinen Themen passt, biete trotzdem Hilfe an oder verweise auf passende Ressourcen.""" diff --git a/gui.py b/gui.py new file mode 100644 index 0000000..498fa2a --- /dev/null +++ b/gui.py @@ -0,0 +1,271 @@ +""" +Dashboard GUI for Eugen Bot +Live monitoring interface using PySimpleGUI +""" +import PySimpleGUI as sg +import threading +from datetime import datetime +from queue import Queue + + +class Dashboard: + """Live monitoring dashboard for bot activity""" + + def __init__(self, bot=None): + """ + Initialize dashboard + + Args: + bot: Reference to the main bot instance + """ + self.bot = bot + sg.theme('DarkBlue3') + + # Event queue for thread-safe updates + self.event_queue = Queue() + self.window = None + self.is_running = False + + # Statistics + self.stats = { + "messages": 0, + "api_calls": 0, + "errors": 0, + "start_time": datetime.now() + } + + def log_event(self, event_type, data): + """ + Add event to queue for display + + Args: + event_type (str): Type of event (chat_message, api_call, etc.) + data (dict): Event data + """ + timestamp = datetime.now().strftime("%H:%M:%S") + + if event_type == "chat_message": + msg = f"{timestamp} | {data['username']}: {data['content']}" + self.stats["messages"] += 1 + elif event_type == "mention_detected": + msg = f"{timestamp} | [MENTION] Bot addressed by {data['username']}" + elif event_type == "api_call": + msg = f"{timestamp} | [API] → Perplexity ({data.get('model', 'sonar-pro')})" + self.stats["api_calls"] += 1 + elif event_type == "api_response": + preview = data['content'][:60] + "..." if len(data['content']) > 60 else data['content'] + msg = f"{timestamp} | [RESPONSE] {preview}" + elif event_type == "bot_response": + preview = data['content'][:60] + "..." if len(data['content']) > 60 else data['content'] + msg = f"{timestamp} | Eugen: @{data['username']} {preview}" + elif event_type == "error": + msg = f"{timestamp} | ❌ ERROR: {data['error']}" + self.stats["errors"] += 1 + elif event_type == "info": + msg = f"{timestamp} | ℹ {data['message']}" + elif event_type == "warning": + msg = f"{timestamp} | ⚠ {data['message']}" + elif event_type == "context_loaded": + msg = f"{timestamp} | [CONTEXT] Loaded {data['count']} messages for {data['username']}" + else: + msg = f"{timestamp} | {event_type}: {data}" + + self.event_queue.put(msg) + + def create_layout(self): + """Create the GUI layout""" + # Header + header = [ + [sg.Text("EUGEN BOT - LIVE DASHBOARD", font=("Arial", 16, "bold"))], + [sg.Text("Status: 🟢 RUNNING", key="-STATUS-", font=("Arial", 10))], + [sg.Text("Uptime: 00:00:00", key="-UPTIME-", size=(20, 1)), + sg.Text("Messages: 0", key="-MSG-COUNT-", size=(20, 1)), + sg.Text("API Calls: 0", key="-API-COUNT-", size=(20, 1)), + sg.Text("Errors: 0", key="-ERROR-COUNT-", size=(20, 1))], + [sg.HorizontalSeparator()] + ] + + # Live feed + feed = [ + [sg.Text("LIVE ACTIVITY FEED", font=("Arial", 12, "bold"))], + [sg.Multiline( + size=(140, 25), + key="-LOG-", + disabled=True, + autoscroll=True, + font=("Courier New", 9) + )] + ] + + # Control buttons + controls = [ + [sg.HorizontalSeparator()], + [ + sg.Button("Clear Log", key="-CLEAR-"), + sg.Button("Reset Stats", key="-RESET-"), + sg.Button("Stop Bot", key="-STOP-", button_color=("white", "red")) + ] + ] + + # Combine all sections + layout = header + feed + controls + + return layout + + def run(self): + """Run the dashboard GUI in main thread""" + layout = self.create_layout() + self.window = sg.Window( + "Eugen Bot Dashboard", + layout, + finalize=True, + size=(1000, 650) + ) + self.is_running = True + + # Main event loop + while self.is_running: + event, values = self.window.read(timeout=100) + + # Handle window close + if event == sg.WINDOW_CLOSED or event == "-STOP-": + self.is_running = False + if self.bot: + self.bot.stop() + break + + # Handle clear log + if event == "-CLEAR-": + self.window["-LOG-"].update("") + + # Handle reset stats + if event == "-RESET-": + self.stats = { + "messages": 0, + "api_calls": 0, + "errors": 0, + "start_time": datetime.now() + } + + # Update log with queued events + while not self.event_queue.empty(): + try: + msg = self.event_queue.get_nowait() + self.window["-LOG-"].print(msg) + except: + break + + # Update statistics + self._update_stats() + + self.window.close() + + def _update_stats(self): + """Update statistics display""" + if not self.window: + return + + # Calculate uptime + uptime = datetime.now() - self.stats["start_time"] + hours, remainder = divmod(int(uptime.total_seconds()), 3600) + minutes, seconds = divmod(remainder, 60) + uptime_str = f"{hours:02d}:{minutes:02d}:{seconds:02d}" + + # Update displays + self.window["-UPTIME-"].update(f"Uptime: {uptime_str}") + self.window["-MSG-COUNT-"].update(f"Messages: {self.stats['messages']}") + self.window["-API-COUNT-"].update(f"API Calls: {self.stats['api_calls']}") + self.window["-ERROR-COUNT-"].update(f"Errors: {self.stats['errors']}") + + def show_error(self, title, message): + """Show error popup""" + sg.popup_error(message, title=title) + + def show_info(self, title, message): + """Show info popup""" + sg.popup(message, title=title) + + def stop(self): + """Stop the dashboard""" + self.is_running = False + if self.window: + self.window.close() + + +class SetupWizard: + """Configuration wizard for first-time setup""" + + def __init__(self): + sg.theme('DarkBlue3') + + def run(self): + """ + Run the setup wizard + + Returns: + dict: Configuration values or None if cancelled + """ + layout = [ + [sg.Text("EUGEN CONFIGURATION WIZARD", font=("Arial", 16, "bold"))], + [sg.HorizontalSeparator()], + + [sg.Text("TWITCH CONFIGURATION", font=("Arial", 12, "bold"))], + [sg.Text("Bot Nickname:", size=(20, 1)), sg.Input("Eugen", key="-BOT-NAME-")], + [sg.Text("OAuth Token:", size=(20, 1)), sg.Input("oauth:", key="-OAUTH-", password_char="*")], + [sg.Text("Channel:", size=(20, 1)), sg.Input("#", key="-CHANNEL-")], + + [sg.HorizontalSeparator()], + + [sg.Text("PERPLEXITY CONFIGURATION", font=("Arial", 12, "bold"))], + [sg.Text("API Key:", size=(20, 1)), sg.Input("pplx-", key="-API-KEY-", password_char="*")], + [sg.Text("Model:", size=(20, 1)), sg.Combo(["sonar-pro", "sonar"], default_value="sonar-pro", key="-MODEL-")], + [sg.Text("Max Tokens:", size=(20, 1)), sg.Input("450", key="-TOKENS-")], + + [sg.HorizontalSeparator()], + + [sg.Checkbox("Enable Debug Mode", default=True, key="-DEBUG-")], + [sg.Checkbox("Auto Reconnect", default=True, key="-RECONNECT-")], + + [sg.HorizontalSeparator()], + + [sg.Button("Save & Start", key="-SAVE-"), sg.Button("Cancel", key="-CANCEL-")] + ] + + window = sg.Window("Eugen Setup", layout) + + config = None + while True: + event, values = window.read() + + if event == sg.WINDOW_CLOSED or event == "-CANCEL-": + break + + if event == "-SAVE-": + # Validate inputs + if not values["-OAUTH-"].startswith("oauth:"): + sg.popup_error("Twitch OAuth token must start with 'oauth:'") + continue + + if not values["-CHANNEL-"].startswith("#"): + sg.popup_error("Channel must start with '#'") + continue + + if not values["-API-KEY-"].startswith("pplx-"): + sg.popup_error("Perplexity API key must start with 'pplx-'") + continue + + # Build config + config = { + "TWITCH_BOT_NICKNAME": values["-BOT-NAME-"], + "TWITCH_OAUTH_TOKEN": values["-OAUTH-"], + "TWITCH_CHANNEL": values["-CHANNEL-"], + "PERPLEXITY_API_KEY": values["-API-KEY-"], + "PERPLEXITY_MODEL": values["-MODEL-"], + "MAX_TOKENS": values["-TOKENS-"], + "DEBUG_MODE": "true" if values["-DEBUG-"] else "false", + "AUTO_RECONNECT": "true" if values["-RECONNECT-"] else "false" + } + break + + window.close() + return config diff --git a/memory.py b/memory.py new file mode 100644 index 0000000..8678913 --- /dev/null +++ b/memory.py @@ -0,0 +1,174 @@ +""" +Conversation Memory for Eugen Bot +Stores and retrieves chat history per user with time-based filtering +""" +import json +from pathlib import Path +from datetime import datetime, timedelta +from typing import List, Dict, Optional + + +class ConversationMemory: + """Manages persistent conversation history for each user""" + + def __init__(self, data_dir="data/conversations", max_messages=25, retention_hours=1): + """ + Initialize conversation memory + + Args: + data_dir (str): Directory to store conversation JSON files + max_messages (int): Maximum messages to store per user + retention_hours (int): How long to keep messages in context + """ + self.data_dir = Path(data_dir) + self.data_dir.mkdir(parents=True, exist_ok=True) + self.max_messages = max_messages + self.retention_hours = retention_hours + + def _get_user_file(self, username): + """Get the file path for a user's conversation history""" + # Sanitize username for filesystem + safe_username = "".join(c for c in username.lower() if c.isalnum() or c in "._-") + return self.data_dir / f"{safe_username}.json" + + def get_user_history(self, username, limit=5): + """ + Load recent chat history for a user + + Args: + username (str): Twitch username + limit (int): Maximum number of messages to return + + Returns: + list: List of message dicts with role, content, timestamp + """ + file_path = self._get_user_file(username) + + if not file_path.exists(): + return [] + + try: + with open(file_path, 'r', encoding='utf-8') as f: + history = json.load(f) + except Exception as e: + print(f"Error loading history for {username}: {e}") + return [] + + # Filter by retention time + cutoff_time = datetime.now() - timedelta(hours=self.retention_hours) + recent = [] + + for msg in history: + try: + msg_time = datetime.fromisoformat(msg['timestamp']) + if msg_time > cutoff_time: + recent.append(msg) + except (KeyError, ValueError): + continue + + # Return only the most recent messages up to limit + return recent[-limit:] if recent else [] + + def add_message(self, username, role, content): + """ + Add a message to user's conversation history + + Args: + username (str): Twitch username + role (str): 'user' or 'assistant' + content (str): Message content + """ + file_path = self._get_user_file(username) + + # Load existing history + history = [] + if file_path.exists(): + try: + with open(file_path, 'r', encoding='utf-8') as f: + history = json.load(f) + except Exception as e: + print(f"Error loading history for {username}: {e}") + history = [] + + # Add new message + history.append({ + "role": role, + "content": content, + "timestamp": datetime.now().isoformat() + }) + + # Enforce max message limit + if len(history) > self.max_messages: + history = history[-self.max_messages:] + + # Save back to file + try: + with open(file_path, 'w', encoding='utf-8') as f: + json.dump(history, f, ensure_ascii=False, indent=2) + except Exception as e: + print(f"Error saving history for {username}: {e}") + + def format_for_prompt(self, history): + """ + Convert history to format suitable for AI API + + Args: + history (list): List of message dicts from get_user_history + + Returns: + list: List of dicts with 'role' and 'content' keys + """ + return [ + { + "role": msg['role'], + "content": msg['content'] + } + for msg in history + ] + + def clear_user_history(self, username): + """ + Clear all history for a specific user + + Args: + username (str): Twitch username + """ + file_path = self._get_user_file(username) + if file_path.exists(): + try: + file_path.unlink() + except Exception as e: + print(f"Error clearing history for {username}: {e}") + + def get_all_users(self): + """ + Get list of all users with conversation history + + Returns: + list: List of usernames + """ + users = [] + for file_path in self.data_dir.glob("*.json"): + users.append(file_path.stem) + return users + + def get_user_message_count(self, username): + """ + Get total message count for a user + + Args: + username (str): Twitch username + + Returns: + int: Number of messages in history + """ + file_path = self._get_user_file(username) + if not file_path.exists(): + return 0 + + try: + with open(file_path, 'r', encoding='utf-8') as f: + history = json.load(f) + return len(history) + except Exception: + return 0 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..7e5c8ec --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +irc==20.1.0 +python-dotenv==1.0.0 +PySimpleGUI==4.60.0 +requests==2.31.0 +httpx==0.25.0 diff --git a/utils.py b/utils.py new file mode 100644 index 0000000..b20ce32 --- /dev/null +++ b/utils.py @@ -0,0 +1,169 @@ +""" +Utility classes for Eugen Bot +Includes MentionDetector for name recognition and Logger for file logging +""" +import re +import logging +from pathlib import Path +from datetime import datetime + + +class MentionDetector: + """Detects if the bot was mentioned in a chat message""" + + def __init__(self, bot_name="Eugen"): + self.bot_name = bot_name + # Create patterns for various mention formats + self.patterns = [ + rf"@{bot_name}", # @Eugen + rf"{bot_name}:", # Eugen: + rf"{bot_name},", # Eugen, + rf"^{bot_name}\s", # Eugen at start + rf"\s{bot_name}\s", # Eugen in middle + rf"\s{bot_name}$", # Eugen at end + ] + # Case-insensitive compilation + self.compiled_patterns = [ + re.compile(pattern, re.IGNORECASE) for pattern in self.patterns + ] + + def is_mentioned(self, message): + """ + Check if bot was mentioned in message + + Args: + message (str): Chat message to check + + Returns: + bool: True if bot was mentioned + """ + if not message: + return False + + for pattern in self.compiled_patterns: + if pattern.search(message): + return True + return False + + def extract_content(self, message): + """ + Extract message content without the mention + + Args: + message (str): Original message with mention + + Returns: + str: Message content without mention prefix + """ + if not message: + return "" + + # Remove common mention patterns + content = message + patterns_to_remove = [ + rf"@{self.bot_name}[,:]?\s*", + rf"{self.bot_name}[,:]?\s*", + ] + + for pattern in patterns_to_remove: + content = re.sub(pattern, "", content, flags=re.IGNORECASE) + + return content.strip() + + +class Logger: + """File-based logger for bot events""" + + def __init__(self, log_dir="logs", debug_mode=False): + self.log_dir = Path(log_dir) + self.log_dir.mkdir(parents=True, exist_ok=True) + self.debug_mode = debug_mode + + # Setup main logger + self.main_logger = self._setup_logger( + "eugen_main", + self.log_dir / "eugen.log", + logging.INFO if not debug_mode else logging.DEBUG + ) + + # Setup API debug logger + self.api_logger = self._setup_logger( + "eugen_api", + self.log_dir / "api_debug.log", + logging.DEBUG + ) + + def _setup_logger(self, name, log_file, level): + """Setup a logger with file handler""" + logger = logging.getLogger(name) + logger.setLevel(level) + + # Avoid duplicate handlers + if logger.handlers: + return logger + + # File handler + fh = logging.FileHandler(log_file, encoding='utf-8') + fh.setLevel(level) + + # Format + formatter = logging.Formatter( + '%(asctime)s | %(levelname)s | %(message)s', + datefmt='%Y-%m-%d %H:%M:%S' + ) + fh.setFormatter(formatter) + logger.addHandler(fh) + + # Console handler for debug mode + if self.debug_mode: + ch = logging.StreamHandler() + ch.setLevel(level) + ch.setFormatter(formatter) + logger.addHandler(ch) + + return logger + + def info(self, message): + """Log info message""" + self.main_logger.info(message) + + def debug(self, message): + """Log debug message""" + self.main_logger.debug(message) + + def error(self, message): + """Log error message""" + self.main_logger.error(message) + + def warning(self, message): + """Log warning message""" + self.main_logger.warning(message) + + def api_call(self, endpoint, model, messages_count): + """Log API call details""" + self.api_logger.debug( + f"API CALL | Endpoint: {endpoint} | Model: {model} | Messages: {messages_count}" + ) + + def api_response(self, status_code, tokens_used, response_time, content_preview): + """Log API response details""" + self.api_logger.debug( + f"API RESPONSE | Status: {status_code} | Tokens: {tokens_used} | " + f"Time: {response_time:.2f}s | Content: {content_preview[:100]}..." + ) + + def api_error(self, status_code, error_message): + """Log API error""" + self.api_logger.error( + f"API ERROR | Status: {status_code} | Error: {error_message}" + ) + + def chat_message(self, username, message): + """Log chat message""" + if self.debug_mode: + self.main_logger.debug(f"CHAT | {username}: {message}") + + def bot_response(self, username, response): + """Log bot response""" + if self.debug_mode: + self.main_logger.debug(f"BOT RESPONSE | To {username}: {response}")