chatbot.py

import json
import os
import re
import uuid
from datetime import datetime
from typing import List, Tuple, Dict

try:
    import faiss
except ImportError:
    raise ImportError(
        "faiss module not found. Please install it using: pip install faiss-cpu\n"
        "Or for GPU support: pip install faiss-gpu"
    )

try:
    import numpy as np
except ImportError:
    raise ImportError(
        "numpy module not found. Please install it using: pip install numpy"
    )

try:
    import requests
except ImportError:
    raise ImportError(
        "requests module not found. Please install it using: pip install requests"
    )

try:
    from openai import OpenAI
except ImportError:
    raise ImportError(
        "openai module not found. Please install it using: pip install openai"
    )

try:
    from sentence_transformers import SentenceTransformer
except ImportError:
    raise ImportError(
        "sentence_transformers module not found. Please install it using: pip install sentence-transformers"
    )
import tkinter as tk
from tkinter import scrolledtext, messagebox


class HarryPotterChatbot:
    def __init__(self):
        # Initialize components
        self.dialog_id = str(uuid.uuid4())
        self.conversation_history = []
        self.texts = []
        self.faiss_index = None
        self.embedder = None
        self.api_key = None
        self.api_url = None
        self.openai_client = None
        
        # Load configuration
        self.load_api_config()
        
        # Initialize OpenAI client
        self.init_openai_client()
        
        # Load and index data
        self.load_data()
        self.create_faiss_index()
        
        # Initialize UI
        self.init_ui()
    
    def load_api_config(self):
        """Load API configuration from api_config.json"""
        # Try to find api_config.json in current directory or script directory
        config_paths = [
            'api_config.json',  # Current working directory
            os.path.join(os.path.dirname(__file__), 'api_config.json'),  # Script directory
            os.path.join(os.path.dirname(os.path.abspath(__file__)), 'api_config.json')  # Absolute path
        ]
        
        config_file = None
        for path in config_paths:
            if os.path.exists(path):
                config_file = path
                break
        
        if not config_file:
            messagebox.showerror(
                "Config Error",
                f"api_config.json not found. Searched in:\n" + "\n".join(config_paths)
            )
            self.api_key = None
            self.api_url = None
            return
        
        try:
            with open(config_file, 'r', encoding='utf-8') as f:
                config = json.load(f)
                self.api_key = config.get('api_key', '').strip()
                self.api_url = config.get('api_url', '').strip()
                
                # Debug output
                print(f"Loaded config from: {config_file}")
                print(f"API Key loaded: {self.api_key[:15]}... (length: {len(self.api_key)})")
                print(f"API URL loaded: {self.api_url}")
                
                # Check if API key is missing or is a placeholder
                if not self.api_key or self.api_key == 'YOUR_API_KEY_HERE' or len(self.api_key) < 10:
                    messagebox.showwarning(
                        "API Key Missing",
                        f"API key is missing or invalid in {config_file}\nCurrent value: '{self.api_key[:20]}...'"
                    )
                    self.api_key = None
                else:
                    print("✓ API key loaded successfully")
                
                if not self.api_url:
                    messagebox.showwarning(
                        "API URL Missing",
                        f"API URL is missing in {config_file}"
                    )
                else:
                    print("✓ API URL loaded successfully")
                    
        except FileNotFoundError:
            messagebox.showerror(
                "Config Error",
                f"api_config.json not found at: {config_file}"
            )
            self.api_key = None
            self.api_url = None
        except json.JSONDecodeError as e:
            messagebox.showerror(
                "Config Error",
                f"Invalid JSON in api_config.json: {str(e)}"
            )
            self.api_key = None
            self.api_url = None
        except Exception as e:
            messagebox.showerror(
                "Config Error",
                f"Error loading api_config.json: {str(e)}"
            )
            self.api_key = None
            self.api_url = None
    
    def init_openai_client(self):
        """Initialize OpenAI client for DashScope compatible API"""
        if self.api_key and self.api_url:
            try:
                self.openai_client = OpenAI(
                    api_key=self.api_key,
                    base_url=self.api_url,
                )
                print("✓ OpenAI client initialized successfully")
            except Exception as e:
                print(f"Warning: Could not initialize OpenAI client: {str(e)}")
                self.openai_client = None
        else:
            print("Warning: API key or URL not set, OpenAI client not initialized")
            self.openai_client = None
    
    def load_data(self):
        """Load Harry Potter data from harry_potter_info.txt"""
        try:
            with open('harry_potter_info.txt', 'r', encoding='utf-8') as f:
                lines = f.readlines()
                self.texts = [line.strip() for line in lines if line.strip()]
            
            if not self.texts:
                messagebox.showwarning(
                    "Data Warning",
                    "harry_potter_info.txt is empty. Please add your Harry Potter data."
                )
        except FileNotFoundError:
            messagebox.showerror(
                "Data Error",
                "harry_potter_info.txt not found."
            )
            self.texts = []
    
    def create_faiss_index(self):
        """Create FAISS index from text embeddings"""
        if not self.texts:
            return
        
        try:
            self.embedder = SentenceTransformer('all-MiniLM-L6-v2')
            
            embeddings = self.embedder.encode(self.texts, show_progress_bar=True)
            embeddings = np.array(embeddings).astype('float32')
            
            dimension = embeddings.shape[1]
            self.faiss_index = faiss.IndexFlatL2(dimension)
            self.faiss_index.add(embeddings)
            
            print(f"Created FAISS index with {len(self.texts)} texts")
        except Exception as e:
            messagebox.showerror("Index Error", f"Error creating FAISS index: {str(e)}")
    
    def sanitize_input(self, text: str) -> str:
        """Sanitize user input to prevent injection attacks"""
        text = re.sub(r'(?i)(system|assistant|user):', '', text)
        text = re.sub(r'(?i)(ignore|forget|override)', '', text)
        
        if len(text) > 1000:
            text = text[:1000]
        
        text = ' '.join(text.split())
        
        return text.strip()
    
    def is_question_relevant(self, question: str, similarity_score: float, threshold: float = 0.3) -> bool:
        """Check if question is relevant to Harry Potter"""
        question_lower = question.lower()
        
        hp_keywords = [
            'harry', 'potter', 'harry potter', 'hogwarts', 'voldemort', 'dumbledore', 'hermione',
            'ron', 'weasley', 'slytherin', 'gryffindor', 'ravenclaw', 'hufflepuff',
            'quidditch', 'wand', 'spell', 'magic', 'wizard', 'witch', 'muggle',
            'deathly hallows', 'horcrux', 'philosopher', 'sorcerer', 'stone',
            'chamber', 'secrets', 'prisoner', 'azkaban', 'goblet', 'fire',
            'order', 'phoenix', 'half-blood', 'prince', 'hallows'
        ]
        
        has_keyword = any(keyword in question_lower for keyword in hp_keywords)
        
        if has_keyword:
            return True
        
        return similarity_score >= threshold
    
    def search_similar_chunks(self, question: str, top_k: int = 5) -> Tuple[List[str], List[float]]:
        """Search for similar text chunks using FAISS"""
        if not self.faiss_index or not self.embedder or not self.texts:
            return [], []
        
        try:
            question_embedding = self.embedder.encode([question])
            question_embedding = np.array(question_embedding).astype('float32')
            
            distances, indices = self.faiss_index.search(question_embedding, min(top_k, len(self.texts)))
            
            if len(distances[0]) > 0:
                max_dist = distances[0].max()
                if max_dist > 0:
                    # Normalize and invert: similarity increases as distance decreases
                    similarities = 1.0 / (1.0 + distances[0] / max_dist)
                else:
                    similarities = np.ones(len(distances[0]))
            else:
                similarities = np.array([0.0])
            
            retrieved_texts = [self.texts[idx] for idx in indices[0] if 0 <= idx < len(self.texts)]
            similarity_scores = similarities[:len(retrieved_texts)]
            
            return retrieved_texts, similarity_scores.tolist()
        except Exception as e:
            print(f"Error in similarity search: {str(e)}")
            return [], []
    
    def build_messages(self, question: str, context_chunks: List[str], conversation_history: List[Dict]) -> List[Dict]:
        """Build messages list for OpenAI API format"""
        system_prompt = """You are a helpful assistant that answers questions about Harry Potter based on the provided context. 
You MUST ONLY answer questions related to Harry Potter. 
If asked about unrelated topics, you MUST respond with exactly: "This query is outside my wizarding world domain. Please ask a Harry Potter-related question."
Do not answer questions about other topics, even if you know the answer.
Stay strictly within the Harry Potter universe."""
        
        context = "\n\n".join([f"Context {i+1}: {chunk}" for i, chunk in enumerate(context_chunks)])
        
        messages = [
            {
                'role': 'system',
                'content': f"""{system_prompt}

Context from Harry Potter information:
{context}"""
            }
        ]
        
        if conversation_history:
            for entry in conversation_history[-5:]:
                messages.append({
                    'role': 'user',
                    'content': entry['question']
                })
                messages.append({
                    'role': 'assistant',
                    'content': entry['answer']
                })
        
        messages.append({
            'role': 'user',
            'content': question
        })
        
        return messages
    
    def log_api_request(self, url: str, method: str = "POST"):
        """Log the API URL being used for the request"""
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        log_message = f"[{timestamp}] API Request: {method} {url}\n"
        
        print(log_message.strip())
        
        try:
            with open('api_requests.log', 'a', encoding='utf-8') as f:
                f.write(log_message)
        except Exception as e:
            print(f"Warning: Could not write to log file: {str(e)}")
    
    def call_qwen_api(self, messages: List[Dict]) -> str:
        """Call Qwen API using OpenAI-compatible format"""
        if not self.openai_client:
            error_msg = f"Error: OpenAI client not initialized.\nAPI Key: {'Set' if self.api_key else 'NOT SET'}\nAPI URL: {'Set' if self.api_url else 'NOT SET'}"
            print(error_msg)
            return error_msg
        
        self.log_api_request(self.api_url)
        
        print(f"Making API request with key: {self.api_key[:15]}...")
        
        try:
            completion = self.openai_client.chat.completions.create(
                model="qwen-plus",  # You can change this to qwen-turbo, qwen-max, etc.
                messages=messages,
                temperature=0.7,
                max_tokens=500
            )
            
            if completion.choices and len(completion.choices) > 0:
                return completion.choices[0].message.content.strip()
            else:
                return "Error: No response from API"
            
        except Exception as e:
            error_msg = str(e)
            print(f"API Error: {error_msg}")
            
            if "401" in error_msg or "unauthorized" in error_msg.lower():
                return "Error: Authentication failed (401). Please check your API key in api_config.json is correct and valid."
            elif "403" in error_msg or "forbidden" in error_msg.lower():
                return "Error: Access forbidden (403). Your API key may not have permission for this operation."
            elif "429" in error_msg or "rate limit" in error_msg.lower():
                return "Error: Rate limit exceeded (429). Please wait a moment and try again."
            else:
                return f"Error: API request failed - {error_msg}"
    
    def store_conversation(self, question: str, answer: str):
        """Store conversation to JSON file"""
        conversation_entry = {
            'dialog_id': self.dialog_id,
            'timestamp': datetime.now().isoformat(),
            'question': question,
            'answer': answer
        }
        
        conversations = []
        if os.path.exists('conversations.json'):
            try:
                with open('conversations.json', 'r', encoding='utf-8') as f:
                    conversations = json.load(f)
            except (json.JSONDecodeError, FileNotFoundError):
                conversations = []
        
        conversations.append(conversation_entry)
        
        try:
            with open('conversations.json', 'w', encoding='utf-8') as f:
                json.dump(conversations, f, indent=2, ensure_ascii=False)
        except Exception as e:
            print(f"Error saving conversation: {str(e)}")
    
    def process_question(self, question: str) -> str:
        """Process user question and return answer"""
        question = self.sanitize_input(question)
        
        if not question:
            return "Please enter a valid question."
        
        retrieved_chunks, similarity_scores = self.search_similar_chunks(question, top_k=5)
        
        max_similarity = max(similarity_scores) if similarity_scores else 0
        
        print(f"Question: '{question}' | Max similarity: {max_similarity:.3f} | Chunks found: {len(retrieved_chunks)}")
        
        if not self.is_question_relevant(question, max_similarity):
            response = "This query is outside my wizarding world domain. Please ask a Harry Potter-related question."
            self.store_conversation(question, response)
            return response
        
        messages = self.build_messages(question, retrieved_chunks, self.conversation_history)
        
        answer = self.call_qwen_api(messages)
        
        answer_lower = answer.lower().strip()
        if 'This query is outside my wizarding world domain. Please ask a Harry Potter-related question.' in answer_lower:
            if not self.is_question_relevant(question, max_similarity, threshold=0.1):
                answer = "This query is outside my wizarding world domain. Please ask a Harry Potter-related question."
        
        self.store_conversation(question, answer)
        
        self.conversation_history.append({
            'question': question,
            'answer': answer
        })
        
        return answer
    
    def init_ui(self):
        """Initialize Tkinter UI"""
        self.root = tk.Tk()
        self.root.title("Harry Potter Chatbot")
        self.root.geometry("800x600")
        
        self.chat_display = scrolledtext.ScrolledText(
            self.root,
            wrap=tk.WORD,
            width=80,
            height=30,
            state=tk.DISABLED,
            font=('Arial', 10)
        )
        self.chat_display.pack(padx=10, pady=10, fill=tk.BOTH, expand=True)
        
        input_frame = tk.Frame(self.root)
        input_frame.pack(padx=10, pady=5, fill=tk.X)
        
        self.question_input = tk.Entry(input_frame, font=('Arial', 11))
        self.question_input.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(0, 5))
        self.question_input.bind('<Return>', lambda e: self.send_question())
        
        send_button = tk.Button(
            input_frame,
            text="Send",
            command=self.send_question,
            font=('Arial', 11),
            bg='#4CAF50',
            fg='white',
            padx=20
        )
        send_button.pack(side=tk.RIGHT)
        
        self.add_message("Bot", f"Hello! I'm a Harry Potter chatbot. Ask me anything about Harry Potter! (Dialog ID: {self.dialog_id})")
    
    def add_message(self, sender: str, message: str):
        """Add message to chat display"""
        self.chat_display.config(state=tk.NORMAL)
        self.chat_display.insert(tk.END, f"{sender}: {message}\n\n")
        self.chat_display.config(state=tk.DISABLED)
        self.chat_display.see(tk.END)
    
    def send_question(self):
        """Handle send button click"""
        question = self.question_input.get()
        if not question.strip():
            return
        
        self.question_input.delete(0, tk.END)
        
        self.add_message("You", question)
        
        self.chat_display.config(state=tk.NORMAL)
        self.chat_display.insert(tk.END, "Bot: Thinking...\n\n")
        self.chat_display.config(state=tk.DISABLED)
        self.chat_display.see(tk.END)
        self.root.update()
        
        answer = self.process_question(question)
        
        self.chat_display.config(state=tk.NORMAL)
        self.chat_display.delete('end-2l', 'end-1l')  
        self.chat_display.insert(tk.END, f"Bot: {answer}\n\n")
        self.chat_display.config(state=tk.DISABLED)
        self.chat_display.see(tk.END)
    
    def run(self):
        """Run the chatbot application"""
        self.root.mainloop()


def main():
    """Main entry point"""
    try:
        chatbot = HarryPotterChatbot()
        chatbot.run()
    except Exception as e:
        messagebox.showerror("Error", f"Failed to start application: {str(e)}")


if __name__ == "__main__":
    main()