Spaces:

aradhyapavan
/

code-hypothesis-multi-personality-bot

Sleeping

code-hypothesis-multi-personality-bot / app.py

Update app.py

7e0ac7c verified about 1 year ago

9.84 kB

#!/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/<personality_type>')
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)

	#!/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/<personality_type>')
	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)