#!/usr/bin/env python3 """ Multi-Personality Chat Bot Flask Application A hackathon project featuring 10 distinct AI personality types with Google AI integration. """ import os import logging import builtins from datetime import datetime from flask import Flask, render_template, request, jsonify, session from flask_socketio import SocketIO, emit, join_room, leave_room import google.generativeai as genai import sqlite3 import json import secrets from modules.simple_personality_engine import PersonalityEngine from modules.database import ChatDatabase # Configure logging (reduced verbosity) logging.basicConfig(level=logging.WARNING) logger = logging.getLogger(__name__) # Optionally silence print output when QUIET=1 (default) if os.getenv("QUIET", "1") == "1": builtins.print = lambda *args, **kwargs: None # Initialize Flask app app = Flask(__name__) # Use env var if provided; otherwise generate a secure random key at startup app.config['SECRET_KEY'] = os.getenv('SECRET_KEY', secrets.token_hex(32)) # Initialize SocketIO (quiet) socketio = SocketIO(app, cors_allowed_origins="*", logger=False, engineio_logger=False) # Configure Google AI strictly via environment variable (no bundled default key) GOOGLE_AI_API_KEY = os.getenv("GOOGLE_AI_API_KEY") if GOOGLE_AI_API_KEY: genai.configure(api_key=GOOGLE_AI_API_KEY) logger.info("Google AI configured using environment variable") else: logger.warning("GOOGLE_AI_API_KEY not set. AI features will be unavailable.") # Initialize components personality_engine = PersonalityEngine() chat_db = ChatDatabase() @app.route('/') def index(): """Main personality selection page""" personalities = personality_engine.get_all_personalities() return render_template('index.html', personalities=personalities) @app.route('/chat/') def chat_personality(personality_type): """Individual personality chat interface""" try: personality_config = personality_engine.get_personality_config(personality_type) if not personality_config: return render_template('error.html', error_message=f"Personality type '{personality_type}' not found"), 404 return render_template('chat_personality.html', personality_type=personality_type, personality_config=personality_config) except Exception as e: logger.error(f"Error loading personality {personality_type}: {str(e)}") return render_template('error.html', error_message="Failed to load personality configuration"), 500 @app.route('/test-ai') def test_ai(): """Test endpoint for Google AI connectivity""" try: model = genai.GenerativeModel('gemini-1.5-flash') response = model.generate_content("Say hello in a sarcastic way") return jsonify({ 'success': True, 'response': response.text, 'message': 'Google AI is working correctly!' }) except Exception as e: logger.error(f"AI test failed: {str(e)}") return jsonify({ 'success': False, 'error': str(e), 'message': 'Google AI connection failed' }), 500 # Socket.IO Event Handlers @socketio.on('connect') def handle_connect(): """Handle client connection""" logger.info(f'Client connected: {request.sid}') emit('status', {'message': 'Connected to Multi-Personality Bot!'}) @socketio.on('disconnect') def handle_disconnect(): """Handle client disconnection""" logger.info(f'Client disconnected: {request.sid}') @socketio.on('join_personality_room') def handle_join_personality_room(data): """Handle joining a personality room""" personality_type = data.get('personality') username = data.get('username', 'Anonymous') room = f"personality_{personality_type}" join_room(room) personality_config = personality_engine.get_personality_config(personality_type) if personality_config: emit('personality_ready', { 'personality': personality_type, 'config': personality_config, 'welcome_message': personality_config.get('welcome', 'Hello!') }) @socketio.on('join_personality') def handle_join_personality(data): """Handle joining a personality room (backup handler)""" # Redirect to join_personality_room handler handle_join_personality_room(data) @socketio.on('personality_message') def handle_personality_message(data): """Handle incoming personality chat messages""" try: user_message = data.get('message', '').strip() personality_type = data.get('personality', 'sarcastic') username = data.get('username', 'Anonymous') if not user_message: emit('error', {'message': 'Please enter a message'}) return # Log and persist user message logger.info(f"User message ({personality_type}): {user_message}") try: chat_db.save_message(username, user_message, personality_type, 'user') except Exception as dberr: logger.warning(f"DB save user message failed: {dberr}") # Send typing indicator emit('bot_typing', {'personality': personality_type}) # Generate AI response try: bot_response = personality_engine.generate_response( message=user_message, personality_type=personality_type, context={} ) logger.info(f"AI response ({personality_type}): {bot_response[:120]}...") # Send response to client emit('personality_response', { 'message': bot_response, 'personality': personality_type, 'timestamp': datetime.now().isoformat() }) # persist bot message try: chat_db.save_message(username, bot_response, personality_type, 'bot') except Exception as dberr2: logger.warning(f"DB save bot message failed: {dberr2}") except Exception as ai_error: logger.error(f"AI generation error: {str(ai_error)}") error_response = "I'm having difficulty connecting to my AI brain right now. Please try again in a moment! 🤖" emit('personality_response', { 'message': error_response, 'personality': personality_type, 'timestamp': datetime.now().isoformat(), 'error': True }) except Exception as e: logger.error(f"Message handling error: {str(e)}") emit('error', {'message': 'Failed to process message. Please try again.'}) @socketio.on('send_message') def handle_send_message(data): """Handle send_message events (backup handler)""" # Redirect to personality_message handler handle_personality_message(data) @socketio.on('get_chat_history') def handle_get_chat_history(data): """Retrieve chat history for a personality""" try: personality_type = data.get('personality', 'sarcastic') limit = data.get('limit', 50) history = chat_db.get_recent_messages(personality_type, limit) emit('chat_history', {'messages': history}) except Exception as e: logger.error(f"Chat history error: {str(e)}") emit('error', {'message': 'Failed to load chat history'}) @socketio.on('clear_chat') def handle_clear_chat(data): """Clear chat history for a personality""" try: personality_type = data.get('personality', 'sarcastic') username = data.get('username', 'Anonymous') # Note: In a production app, you might want to soft-delete or archive success = chat_db.clear_personality_chat(personality_type, username) if success: emit('chat_cleared', {'personality': personality_type}) else: emit('error', {'message': 'Failed to clear chat history'}) except Exception as e: logger.error(f"Clear chat error: {str(e)}") emit('error', {'message': 'Failed to clear chat history'}) # Error Handlers @app.errorhandler(404) def not_found_error(error): """Handle 404 errors""" return render_template('error.html', error_message="Page not found"), 404 @app.errorhandler(500) def internal_error(error): """Handle 500 errors""" return render_template('error.html', error_message="Internal server error"), 500 if __name__ == '__main__': try: # Initialize database chat_db.initialize_database() # Test Google AI connection logger.info("Testing Google AI connection...") api_test_success = personality_engine.test_api_connection() if api_test_success: logger.info("Google AI connection test successful!") else: logger.warning("Google AI connection test failed - app will still start but may have issues") # Start the application logger.info("Starting Multi-Personality Chat Bot...") logger.info("Available personalities: " + ", ".join(personality_engine.get_personality_list())) # Respect PORT env for platforms like Hugging Face Spaces port = int(os.getenv("PORT", os.getenv("HF_PORT", 7860))) socketio.run(app, host='0.0.0.0', port=port, debug=False, allow_unsafe_werkzeug=True) except Exception as e: logger.error(f"Failed to start application: {str(e)}") print(f"❌ Startup Error: {str(e)}") exit(1)