""" Analytics Engine Module Author: RSK World (https://rskworld.in) Founder: Molla Samser Designer & Tester: Rima Khatun Contact: help@rskworld.in, +91 93305 39277 Year: 2026 Description: Advanced analytics engine for comprehensive chatbot insights and metrics """ import sqlite3 import json import logging from datetime import datetime, timedelta from typing import Dict, List, Optional, Any, Tuple import pandas as pd import numpy as np from collections import defaultdict, Counter import matplotlib.pyplot as plt import seaborn as sns import base64 from io import BytesIO import os logger = logging.getLogger(__name__) class AnalyticsEngine: def __init__(self, db_path: str = "analytics.db"): self.db_path = db_path self._init_database() # Analytics settings self.retention_days = 90 self.cache_timeout = 300 # 5 minutes self.cache = {} def _init_database(self): """Initialize analytics database""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # Create analytics tables cursor.execute(''' CREATE TABLE IF NOT EXISTS user_analytics ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, session_id TEXT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, event_type TEXT NOT NULL, event_data TEXT, language TEXT, sentiment_score REAL, emotion_data TEXT, response_time REAL, message_length INTEGER, metadata TEXT ) ''') cursor.execute(''' CREATE TABLE IF NOT EXISTS conversation_analytics ( id INTEGER PRIMARY KEY AUTOINCREMENT, conversation_id TEXT NOT NULL, user_id TEXT NOT NULL, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, message_count INTEGER, avg_response_time REAL, language_distribution TEXT, sentiment_summary TEXT, emotion_summary TEXT, topic_analysis TEXT, satisfaction_score REAL, metadata TEXT ) ''') cursor.execute(''' CREATE TABLE IF NOT EXISTS system_analytics ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, metric_name TEXT NOT NULL, metric_value REAL, metric_type TEXT, tags TEXT, metadata TEXT ) ''') cursor.execute(''' CREATE TABLE IF NOT EXISTS performance_analytics ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, endpoint TEXT, response_time REAL, status_code INTEGER, error_message TEXT, user_agent TEXT, ip_address TEXT, metadata TEXT ) ''') # Create indexes cursor.execute('CREATE INDEX IF NOT EXISTS idx_user_analytics_user_id ON user_analytics(user_id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_user_analytics_timestamp ON user_analytics(timestamp)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_conversation_analytics_user_id ON conversation_analytics(user_id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_system_analytics_timestamp ON system_analytics(timestamp)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_performance_analytics_timestamp ON performance_analytics(timestamp)') conn.commit() conn.close() logger.info("Analytics database initialized successfully") except Exception as e: logger.error(f"Analytics database initialization error: {str(e)}") def track_user_event(self, user_id: str, event_type: str, event_data: Dict = None, language: str = None, sentiment_score: float = None, emotion_data: Dict = None, response_time: float = None, message_length: int = None, metadata: Dict = None, session_id: str = None): """Track user interaction events""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' INSERT INTO user_analytics (user_id, session_id, event_type, event_data, language, sentiment_score, emotion_data, response_time, message_length, metadata) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( user_id, session_id, event_type, json.dumps(event_data) if event_data else None, language, sentiment_score, json.dumps(emotion_data) if emotion_data else None, response_time, message_length, json.dumps(metadata) if metadata else None )) conn.commit() conn.close() except Exception as e: logger.error(f"Error tracking user event: {str(e)}") def track_conversation(self, conversation_id: str, user_id: str, message_count: int, avg_response_time: float, language_distribution: Dict, sentiment_summary: Dict, emotion_summary: Dict, topic_analysis: Dict = None, satisfaction_score: float = None, metadata: Dict = None): """Track conversation analytics""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' INSERT INTO conversation_analytics (conversation_id, user_id, message_count, avg_response_time, language_distribution, sentiment_summary, emotion_summary, topic_analysis, satisfaction_score, metadata) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( conversation_id, user_id, message_count, avg_response_time, json.dumps(language_distribution), json.dumps(sentiment_summary), json.dumps(emotion_summary), json.dumps(topic_analysis) if topic_analysis else None, satisfaction_score, json.dumps(metadata) if metadata else None )) conn.commit() conn.close() except Exception as e: logger.error(f"Error tracking conversation: {str(e)}") def track_system_metric(self, metric_name: str, metric_value: float, metric_type: str = 'gauge', tags: Dict = None, metadata: Dict = None): """Track system performance metrics""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' INSERT INTO system_analytics (metric_name, metric_value, metric_type, tags, metadata) VALUES (?, ?, ?, ?, ?) ''', ( metric_name, metric_value, metric_type, json.dumps(tags) if tags else None, json.dumps(metadata) if metadata else None )) conn.commit() conn.close() except Exception as e: logger.error(f"Error tracking system metric: {str(e)}") def track_performance(self, endpoint: str, response_time: float, status_code: int, error_message: str = None, user_agent: str = None, ip_address: str = None, metadata: Dict = None): """Track API performance metrics""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' INSERT INTO performance_analytics (endpoint, response_time, status_code, error_message, user_agent, ip_address, metadata) VALUES (?, ?, ?, ?, ?, ?, ?) ''', ( endpoint, response_time, status_code, error_message, user_agent, ip_address, json.dumps(metadata) if metadata else None )) conn.commit() conn.close() except Exception as e: logger.error(f"Error tracking performance: {str(e)}") def get_user_analytics(self, user_id: str = None, days: int = 30) -> Dict: """Get comprehensive user analytics""" try: cache_key = f"user_analytics_{user_id}_{days}" if cache_key in self.cache: return self.cache[cache_key] conn = sqlite3.connect(self.db_path) # Date filter start_date = datetime.now() - timedelta(days=days) query = ''' SELECT event_type, language, sentiment_score, response_time, message_length, timestamp, emotion_data FROM user_analytics WHERE timestamp >= ? ''' params = [start_date] if user_id: query += ' AND user_id = ?' params.append(user_id) df = pd.read_sql_query(query, conn, params=params) conn.close() if df.empty: return {'error': 'No data available'} # Process analytics analytics = { 'overview': self._get_user_overview(df), 'engagement': self._get_engagement_metrics(df), 'language_usage': self._get_language_usage(df), 'sentiment_analysis': self._get_sentiment_analysis(df), 'emotion_analysis': self._get_emotion_analysis(df), 'response_times': self._get_response_time_analysis(df), 'activity_patterns': self._get_activity_patterns(df), 'message_analysis': self._get_message_analysis(df) } # Cache result self.cache[cache_key] = analytics return analytics except Exception as e: logger.error(f"Error getting user analytics: {str(e)}") return {'error': 'Failed to get analytics'} def get_conversation_analytics(self, days: int = 30) -> Dict: """Get conversation-level analytics""" try: cache_key = f"conversation_analytics_{days}" if cache_key in self.cache: return self.cache[cache_key] conn = sqlite3.connect(self.db_path) start_date = datetime.now() - timedelta(days=days) query = ''' SELECT message_count, avg_response_time, language_distribution, sentiment_summary, emotion_summary, topic_analysis, satisfaction_score, timestamp FROM conversation_analytics WHERE timestamp >= ? ''' df = pd.read_sql_query(query, conn, params=[start_date]) conn.close() if df.empty: return {'error': 'No data available'} analytics = { 'overview': self._get_conversation_overview(df), 'performance': self._get_conversation_performance(df), 'satisfaction': self._get_satisfaction_analysis(df), 'topic_trends': self._get_topic_trends(df), 'language_trends': self._get_language_trends(df) } self.cache[cache_key] = analytics return analytics except Exception as e: logger.error(f"Error getting conversation analytics: {str(e)}") return {'error': 'Failed to get analytics'} def get_system_analytics(self, days: int = 7) -> Dict: """Get system performance analytics""" try: cache_key = f"system_analytics_{days}" if cache_key in self.cache: return self.cache[cache_key] conn = sqlite3.connect(self.db_path) start_date = datetime.now() - timedelta(days=days) # System metrics system_query = ''' SELECT metric_name, metric_value, metric_type, tags, timestamp FROM system_analytics WHERE timestamp >= ? ''' system_df = pd.read_sql_query(system_query, conn, params=[start_date]) # Performance metrics perf_query = ''' SELECT endpoint, response_time, status_code, timestamp FROM performance_analytics WHERE timestamp >= ? ''' perf_df = pd.read_sql_query(perf_query, conn, params=[start_date]) conn.close() analytics = { 'system_metrics': self._process_system_metrics(system_df), 'performance_metrics': self._process_performance_metrics(perf_df), 'health_status': self._get_health_status(system_df, perf_df), 'alerts': self._generate_alerts(system_df, perf_df) } self.cache[cache_key] = analytics return analytics except Exception as e: logger.error(f"Error getting system analytics: {str(e)}") return {'error': 'Failed to get analytics'} def get_real_time_metrics(self) -> Dict: """Get real-time system metrics""" try: conn = sqlite3.connect(self.db_path) # Last 5 minutes start_time = datetime.now() - timedelta(minutes=5) # Active users active_users_query = ''' SELECT COUNT(DISTINCT user_id) as count FROM user_analytics WHERE timestamp >= ? ''' # Recent conversations recent_conversations_query = ''' SELECT COUNT(DISTINCT conversation_id) as count FROM conversation_analytics WHERE timestamp >= ? ''' # Average response time avg_response_query = ''' SELECT AVG(response_time) as avg_time FROM performance_analytics WHERE timestamp >= ? AND status_code = 200 ''' # Error rate error_rate_query = ''' SELECT COUNT(CASE WHEN status_code >= 400 THEN 1 END) * 100.0 / COUNT(*) as error_rate FROM performance_analytics WHERE timestamp >= ? ''' params = [start_time] active_users = pd.read_sql_query(active_users_query, conn, params=params)['count'].iloc[0] recent_conversations = pd.read_sql_query(recent_conversations_query, conn, params=params)['count'].iloc[0] avg_response_time = pd.read_sql_query(avg_response_query, conn, params=params)['avg_time'].iloc[0] error_rate = pd.read_sql_query(error_rate_query, conn, params=params)['error_rate'].iloc[0] conn.close() return { 'active_users': int(active_users) if active_users else 0, 'recent_conversations': int(recent_conversations) if recent_conversations else 0, 'avg_response_time': float(avg_response_time) if avg_response_time else 0, 'error_rate': float(error_rate) if error_rate else 0, 'timestamp': datetime.now().isoformat() } except Exception as e: logger.error(f"Error getting real-time metrics: {str(e)}") return {'error': 'Failed to get metrics'} def generate_report(self, report_type: str, days: int = 30, format: str = 'json') -> Dict: """Generate comprehensive analytics report""" try: report_data = { 'report_type': report_type, 'period': f'{days} days', 'generated_at': datetime.now().isoformat(), 'data': {} } if report_type == 'user': report_data['data'] = self.get_user_analytics(days=days) elif report_type == 'conversation': report_data['data'] = self.get_conversation_analytics(days=days) elif report_type == 'system': report_data['data'] = self.get_system_analytics(days=days) elif report_type == 'comprehensive': report_data['data'] = { 'user': self.get_user_analytics(days=days), 'conversation': self.get_conversation_analytics(days=days), 'system': self.get_system_analytics(days=days), 'real_time': self.get_real_time_metrics() } if format == 'json': return report_data elif format == 'csv': return self._convert_to_csv(report_data) else: return report_data except Exception as e: logger.error(f"Error generating report: {str(e)}") return {'error': 'Failed to generate report'} def create_visualization(self, data_type: str, days: int = 30) -> str: """Create data visualization and return as base64 image""" try: plt.style.use('seaborn-v0_8') if data_type == 'user_activity': return self._create_user_activity_chart(days) elif data_type == 'language_distribution': return self._create_language_distribution_chart(days) elif data_type == 'sentiment_trends': return self._create_sentiment_trends_chart(days) elif data_type == 'response_times': return self._create_response_times_chart(days) elif data_type == 'conversation_volume': return self._create_conversation_volume_chart(days) else: return None except Exception as e: logger.error(f"Error creating visualization: {str(e)}") return None def _get_user_overview(self, df: pd.DataFrame) -> Dict: """Get user overview statistics""" return { 'total_events': len(df), 'unique_users': df['user_id'].nunique() if 'user_id' in df.columns else 0, 'avg_session_length': df.groupby('user_id').size().mean() if 'user_id' in df.columns else 0, 'most_active_hour': df['timestamp'].dt.hour.mode().iloc[0] if not df.empty else 0 } def _get_engagement_metrics(self, df: pd.DataFrame) -> Dict: """Get user engagement metrics""" if 'event_type' not in df.columns: return {} event_counts = df['event_type'].value_counts() return { 'event_distribution': event_counts.to_dict(), 'engagement_rate': len(df[df['event_type'] == 'message']) / len(df) if not df.empty else 0, 'retention_rate': self._calculate_retention_rate(df) } def _get_language_usage(self, df: pd.DataFrame) -> Dict: """Get language usage statistics""" if 'language' not in df.columns: return {} language_counts = df['language'].value_counts() total = len(df) return { 'distribution': language_counts.to_dict(), 'percentages': (language_counts / total * 100).round(2).to_dict(), 'diversity_score': len(language_counts) } def _get_sentiment_analysis(self, df: pd.DataFrame) -> Dict: """Get sentiment analysis""" if 'sentiment_score' not in df.columns: return {} sentiment_scores = df['sentiment_score'].dropna() return { 'average_sentiment': float(sentiment_scores.mean()) if not sentiment_scores.empty else 0, 'sentiment_distribution': self._categorize_sentiment(sentiment_scores), 'trend': self._calculate_sentiment_trend(sentiment_scores) } def _get_emotion_analysis(self, df: pd.DataFrame) -> Dict: """Get emotion analysis""" if 'emotion_data' not in df.columns: return {} emotions = [] for emotion_json in df['emotion_data'].dropna(): try: emotion_data = json.loads(emotion_json) if isinstance(emotion_data, dict): emotions.append(emotion_data) except: continue if not emotions: return {} # Aggregate emotions emotion_totals = defaultdict(float) for emotion_dict in emotions: for emotion, score in emotion_dict.items(): emotion_totals[emotion] += score return { 'emotion_distribution': dict(emotion_totals), 'dominant_emotion': max(emotion_totals.items(), key=lambda x: x[1])[0] if emotion_totals else None } def _get_response_time_analysis(self, df: pd.DataFrame) -> Dict: """Get response time analysis""" if 'response_time' not in df.columns: return {} response_times = df['response_time'].dropna() return { 'average_response_time': float(response_times.mean()) if not response_times.empty else 0, 'median_response_time': float(response_times.median()) if not response_times.empty else 0, 'p95_response_time': float(response_times.quantile(0.95)) if not response_times.empty else 0, 'slow_responses': len(response_times[response_times > 5.0]) if not response_times.empty else 0 } def _get_activity_patterns(self, df: pd.DataFrame) -> Dict: """Get user activity patterns""" if 'timestamp' not in df.columns: return {} df['timestamp'] = pd.to_datetime(df['timestamp']) df['hour'] = df['timestamp'].dt.hour df['day_of_week'] = df['timestamp'].dt.day_name() return { 'hourly_distribution': df['hour'].value_counts().to_dict(), 'daily_distribution': df['day_of_week'].value_counts().to_dict(), 'peak_hour': df['hour'].mode().iloc[0] if not df.empty else 0, 'peak_day': df['day_of_week'].mode().iloc[0] if not df.empty else 'Monday' } def _get_message_analysis(self, df: pd.DataFrame) -> Dict: """Get message analysis""" if 'message_length' not in df.columns: return {} message_lengths = df['message_length'].dropna() return { 'average_message_length': float(message_lengths.mean()) if not message_lengths.empty else 0, 'median_message_length': float(message_lengths.median()) if not message_lengths.empty else 0, 'long_messages': len(message_lengths[message_lengths > 500]) if not message_lengths.empty else 0, 'short_messages': len(message_lengths[message_lengths < 50]) if not message_lengths.empty else 0 } def _categorize_sentiment(self, sentiment_scores: pd.Series) -> Dict: """Categorize sentiment scores""" categories = {'positive': 0, 'neutral': 0, 'negative': 0} for score in sentiment_scores: if score > 0.1: categories['positive'] += 1 elif score < -0.1: categories['negative'] += 1 else: categories['neutral'] += 1 return categories def _calculate_sentiment_trend(self, sentiment_scores: pd.Series) -> str: """Calculate sentiment trend""" if len(sentiment_scores) < 2: return 'insufficient_data' recent_avg = sentiment_scores.tail(10).mean() older_avg = sentiment_scores.head(10).mean() if recent_avg > older_avg + 0.1: return 'improving' elif recent_avg < older_avg - 0.1: return 'declining' else: return 'stable' def _calculate_retention_rate(self, df: pd.DataFrame) -> float: """Calculate user retention rate""" if 'user_id' not in df.columns or 'timestamp' not in df.columns: return 0.0 df['timestamp'] = pd.to_datetime(df['timestamp']) # Users from first week first_week = df[df['timestamp'] >= df['timestamp'].min() + timedelta(days=7)] initial_users = set(first_week['user_id'].unique()) # Users active in last week last_week = df[df['timestamp'] >= df['timestamp'].max() - timedelta(days=7)] retained_users = set(last_week['user_id'].unique()) if not initial_users: return 0.0 return len(retained_users & initial_users) / len(initial_users) * 100 def _create_user_activity_chart(self, days: int) -> str: """Create user activity chart""" conn = sqlite3.connect(self.db_path) start_date = datetime.now() - timedelta(days=days) query = ''' SELECT DATE(timestamp) as date, COUNT(*) as activity_count FROM user_analytics WHERE timestamp >= ? GROUP BY DATE(timestamp) ORDER BY date ''' df = pd.read_sql_query(query, conn, params=[start_date]) conn.close() if df.empty: return None plt.figure(figsize=(12, 6)) plt.plot(df['date'], df['activity_count'], marker='o') plt.title('User Activity Over Time') plt.xlabel('Date') plt.ylabel('Activity Count') plt.xticks(rotation=45) plt.tight_layout() return self._save_chart_as_base64() def _save_chart_as_base64(self) -> str: """Save matplotlib chart as base64 string""" buffer = BytesIO() plt.savefig(buffer, format='png', dpi=150, bbox_inches='tight') buffer.seek(0) image_base64 = base64.b64encode(buffer.getvalue()).decode() plt.close() return image_base64 def cleanup_old_data(self, days: int = None): """Clean up old analytics data""" try: if days is None: days = self.retention_days cutoff_date = datetime.now() - timedelta(days=days) conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # Clean up each table tables = ['user_analytics', 'conversation_analytics', 'system_analytics', 'performance_analytics'] total_deleted = 0 for table in tables: cursor.execute(f'DELETE FROM {table} WHERE timestamp < ?', (cutoff_date,)) total_deleted += cursor.rowcount conn.commit() conn.close() logger.info(f"Cleaned up {total_deleted} old analytics records") except Exception as e: logger.error(f"Error cleaning up old data: {str(e)}")