# admin_module.py - Administrative interface for English Helper import os import json import hashlib from datetime import datetime, timedelta from functools import wraps from flask import session, request, jsonify, redirect, url_for import sqlite3 from database import get_db_connection import logging logger = logging.getLogger(__name__) class AdminManager: def __init__(self): self.admin_credentials = self._load_admin_credentials() self.token_costs = { 'groq': {'input': 0.00000059, 'output': 0.00000079}, # per token 'gemini': {'input': 0.00000125, 'output': 0.00000375} # per token } def _load_admin_credentials(self): """Load admin credentials from environment variables (Hugging Face secrets)""" try: # Try to load from Hugging Face secrets format admin_user = os.environ.get('ADMIN_USERNAME', 'admin') admin_pass = os.environ.get('ADMIN_PASSWORD', 'admin123') # For security, hash the password admin_pass_hash = hashlib.sha256(admin_pass.encode()).hexdigest() return { 'username': admin_user, 'password_hash': admin_pass_hash, 'original_password': admin_pass # Store for initial comparison } except Exception as e: logger.error(f"Error loading admin credentials: {e}") # Fallback credentials return { 'username': 'admin', 'password_hash': hashlib.sha256('admin123'.encode()).hexdigest(), 'original_password': 'admin123' } def authenticate_admin(self, username, password): """Authenticate admin user""" try: if username != self.admin_credentials['username']: return False # Check password hash password_hash = hashlib.sha256(password.encode()).hexdigest() return password_hash == self.admin_credentials['password_hash'] except Exception as e: logger.error(f"Admin authentication error: {e}") return False def is_admin_logged_in(self): """Check if admin is logged in""" authenticated = session.get('admin_authenticated', False) username = session.get('admin_username') login_time = session.get('admin_login_time') # Debug logging logger.info(f"Admin auth check: authenticated={authenticated}, username={username}, login_time={login_time}") # Check if session has expired (24 hours) if authenticated and login_time: try: login_datetime = datetime.fromisoformat(login_time) if datetime.now() - login_datetime > timedelta(hours=24): logger.info("Admin session expired, logging out") self.logout_admin() return False except Exception as e: logger.error(f"Error checking session expiry: {e}") return authenticated def login_admin(self, username, password): """Admin login""" if self.authenticate_admin(username, password): session['admin_authenticated'] = True session['admin_username'] = username session['admin_login_time'] = datetime.now().isoformat() session.permanent = True # Make session permanent logger.info(f"Admin login successful: {username}") return True else: logger.warning(f"Admin login failed for username: {username}") return False def logout_admin(self): """Admin logout""" session.pop('admin_authenticated', None) session.pop('admin_username', None) session.pop('admin_login_time', None) def get_system_stats(self): """Get comprehensive system statistics""" try: conn = get_db_connection() cursor = conn.cursor() # User statistics cursor.execute("SELECT COUNT(*) FROM users") total_users = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM users WHERE created_at > datetime('now', '-7 days')") new_users_week = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM users WHERE created_at > datetime('now', '-1 day')") new_users_today = cursor.fetchone()[0] # Activity statistics cursor.execute("SELECT COUNT(*) FROM study_sessions") total_sessions = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM flashcards") total_flashcards = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM user_articles") total_articles = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM study_plans") total_study_plans = cursor.fetchone()[0] # Token usage statistics cursor.execute("SELECT SUM(tokens_used), COUNT(*) FROM token_usage") token_stats = cursor.fetchone() total_tokens = token_stats[0] if token_stats[0] else 0 total_api_calls = token_stats[1] if token_stats[1] else 0 # Calculate estimated costs estimated_cost = self._calculate_estimated_cost(cursor) # Recent activity cursor.execute(""" SELECT u.email, s.created_at, s.activity_type FROM study_sessions s JOIN users u ON s.user_id = u.id ORDER BY s.created_at DESC LIMIT 10 """) recent_activity = cursor.fetchall() conn.close() return { 'users': { 'total': total_users, 'new_week': new_users_week, 'new_today': new_users_today }, 'activity': { 'total_sessions': total_sessions, 'total_flashcards': total_flashcards, 'total_articles': total_articles, 'total_study_plans': total_study_plans }, 'api_usage': { 'total_tokens': total_tokens, 'total_calls': total_api_calls, 'estimated_cost': estimated_cost }, 'recent_activity': [ { 'user': activity[0], 'timestamp': activity[1], 'activity': activity[2] } for activity in recent_activity ] } except Exception as e: logger.error(f"Error getting system stats: {e}") return {} def _calculate_estimated_cost(self, cursor): """Calculate estimated API costs""" try: cursor.execute(""" SELECT api_provider, SUM(input_tokens), SUM(output_tokens) FROM token_usage GROUP BY api_provider """) usage_by_provider = cursor.fetchall() total_cost = 0 for provider, input_tokens, output_tokens in usage_by_provider: if provider in self.token_costs: costs = self.token_costs[provider] total_cost += (input_tokens * costs['input']) + (output_tokens * costs['output']) return round(total_cost, 4) except: return 0 def get_all_users(self, page=1, per_page=20): """Get paginated list of all users""" try: conn = get_db_connection() cursor = conn.cursor() offset = (page - 1) * per_page cursor.execute(""" SELECT u.id, u.email, u.created_at, u.email_confirmed, u.last_login, COUNT(DISTINCT s.id) as session_count, COUNT(DISTINCT f.id) as flashcard_count, COUNT(DISTINCT a.id) as article_count FROM users u LEFT JOIN study_sessions s ON u.id = s.user_id LEFT JOIN flashcards f ON u.id = f.user_id LEFT JOIN user_articles a ON u.id = a.user_id GROUP BY u.id ORDER BY u.created_at DESC LIMIT ? OFFSET ? """, (per_page, offset)) users = cursor.fetchall() # Get total count cursor.execute("SELECT COUNT(*) FROM users") total_users = cursor.fetchone()[0] conn.close() return { 'users': [ { 'id': user[0], 'email': user[1], 'created_at': user[2], 'email_confirmed': bool(user[3]), 'last_login': user[4], 'session_count': user[5], 'flashcard_count': user[6], 'article_count': user[7] } for user in users ], 'total': total_users, 'page': page, 'per_page': per_page, 'total_pages': (total_users + per_page - 1) // per_page } except Exception as e: logger.error(f"Error getting users: {e}") return {'users': [], 'total': 0} def delete_user(self, user_id): """Delete a user and all associated data""" try: conn = get_db_connection() cursor = conn.cursor() # Delete in order to respect foreign key constraints tables = [ 'study_plan_activities', 'study_plans', 'user_analytics', 'content_recommendations', 'user_interests', 'user_articles', 'study_sessions', 'flashcards', 'user_settings', 'users' ] for table in tables: cursor.execute(f"DELETE FROM {table} WHERE user_id = ?", (user_id,)) conn.commit() conn.close() return True except Exception as e: logger.error(f"Error deleting user {user_id}: {e}") return False def get_user_details(self, user_id): """Get detailed information about a specific user""" try: conn = get_db_connection() cursor = conn.cursor() # Basic user info cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,)) user = cursor.fetchone() if not user: return None # User settings cursor.execute("SELECT * FROM user_settings WHERE user_id = ?", (user_id,)) settings = cursor.fetchone() # Recent activity cursor.execute(""" SELECT activity_type, created_at, duration_minutes FROM study_sessions WHERE user_id = ? ORDER BY created_at DESC LIMIT 20 """, (user_id,)) recent_sessions = cursor.fetchall() # Token usage cursor.execute(""" SELECT api_provider, SUM(input_tokens), SUM(output_tokens), COUNT(*) FROM token_usage WHERE user_id = ? GROUP BY api_provider """, (user_id,)) token_usage = cursor.fetchall() conn.close() return { 'user': { 'id': user[0], 'email': user[1], 'created_at': user[2], 'email_confirmed': bool(user[3]), 'last_login': user[4] }, 'settings': dict(zip([col[0] for col in cursor.description], settings)) if settings else {}, 'recent_sessions': [ { 'activity': session[0], 'timestamp': session[1], 'duration': session[2] } for session in recent_sessions ], 'token_usage': [ { 'provider': usage[0], 'input_tokens': usage[1], 'output_tokens': usage[2], 'calls': usage[3] } for usage in token_usage ] } except Exception as e: logger.error(f"Error getting user details for {user_id}: {e}") return None def record_token_usage(self, user_id, api_provider, input_tokens, output_tokens, operation_type): """Record token usage for cost tracking""" try: conn = get_db_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO token_usage (user_id, api_provider, input_tokens, output_tokens, operation_type, created_at) VALUES (?, ?, ?, ?, ?, ?) """, (user_id, api_provider, input_tokens, output_tokens, operation_type, datetime.now().isoformat())) conn.commit() conn.close() return True except Exception as e: logger.error(f"Error recording token usage: {e}") return False def get_database_schema(self): """Get database schema information""" try: conn = get_db_connection() cursor = conn.cursor() # Get all tables cursor.execute("SELECT name FROM sqlite_master WHERE type='table'") tables = cursor.fetchall() schema_info = {} for table in tables: table_name = table[0] # Get table info cursor.execute(f"PRAGMA table_info({table_name})") columns = cursor.fetchall() # Get row count cursor.execute(f"SELECT COUNT(*) FROM {table_name}") row_count = cursor.fetchone()[0] schema_info[table_name] = { 'columns': [ { 'name': col[1], 'type': col[2], 'not_null': bool(col[3]), 'primary_key': bool(col[5]) } for col in columns ], 'row_count': row_count } conn.close() return schema_info except Exception as e: logger.error(f"Error getting database schema: {e}") return {} def get_system_health(self): """Get system health metrics""" try: import psutil import os # Memory usage memory = psutil.virtual_memory() # Disk usage disk = psutil.disk_usage('/') # Database size db_path = 'data/englishhelper.db' db_size = os.path.getsize(db_path) if os.path.exists(db_path) else 0 # Recent error logs (would implement proper logging) recent_errors = self._get_recent_errors() return { 'memory': { 'total': memory.total, 'used': memory.used, 'available': memory.available, 'percent': memory.percent }, 'disk': { 'total': disk.total, 'used': disk.used, 'free': disk.free, 'percent': disk.percent }, 'database': { 'size_bytes': db_size, 'size_mb': round(db_size / 1024 / 1024, 2) }, 'recent_errors': recent_errors, 'uptime': self._get_uptime() } except Exception as e: logger.error(f"Error getting system health: {e}") return {} def _get_recent_errors(self): """Get recent error logs (simplified)""" try: # This would typically read from log files # For now, return sample data return [ { 'timestamp': '2024-10-11 14:30:00', 'level': 'ERROR', 'message': 'API rate limit exceeded for user 123', 'module': 'groq_client' }, { 'timestamp': '2024-10-11 13:45:00', 'level': 'WARNING', 'message': 'High memory usage detected', 'module': 'system_monitor' } ] except: return [] def _get_uptime(self): """Get system uptime""" try: import psutil boot_time = psutil.boot_time() uptime_seconds = datetime.now().timestamp() - boot_time days = int(uptime_seconds // 86400) hours = int((uptime_seconds % 86400) // 3600) minutes = int((uptime_seconds % 3600) // 60) return f"{days}d {hours}h {minutes}m" except: return "Unknown" def check_system_alerts(self): """Check for system alerts and warnings""" alerts = [] try: # Check token usage limits conn = get_db_connection() cursor = conn.cursor() # Check daily token usage cursor.execute(""" SELECT SUM(tokens_used) FROM token_usage WHERE date(created_at) = date('now') """) daily_tokens = cursor.fetchone()[0] or 0 if daily_tokens > 100000: # Alert threshold alerts.append({ 'type': 'warning', 'message': f'High daily token usage: {daily_tokens:,} tokens', 'action': 'Monitor API costs' }) # Check error rates cursor.execute(""" SELECT COUNT(*) FROM token_usage WHERE created_at > datetime('now', '-1 hour') """) hourly_requests = cursor.fetchone()[0] or 0 if hourly_requests > 500: # High load threshold alerts.append({ 'type': 'info', 'message': f'High API request rate: {hourly_requests} requests/hour', 'action': 'Monitor performance' }) # Check database size health = self.get_system_health() if health.get('database', {}).get('size_mb', 0) > 100: # 100MB threshold alerts.append({ 'type': 'warning', 'message': f'Large database size: {health["database"]["size_mb"]}MB', 'action': 'Consider archiving old data' }) conn.close() return alerts except Exception as e: logger.error(f"Error checking system alerts: {e}") return [] # Decorator for admin-only routes def admin_required(f): @wraps(f) def decorated_function(*args, **kwargs): is_authenticated = admin_manager.is_admin_logged_in() # Enhanced logging for debugging from flask import session, request logger.info(f"Admin required check for {f.__name__}: authenticated={is_authenticated}") logger.info(f"Session keys: {list(session.keys())}") logger.info(f"Request URL: {request.url}") if not is_authenticated: logger.warning(f"Admin authentication failed for {f.__name__}") return jsonify({ 'error': 'Admin authentication required', 'authenticated': False, 'endpoint': f.__name__ }), 401 logger.info(f"Admin access granted to {f.__name__}") return f(*args, **kwargs) return decorated_function # Global admin manager instance admin_manager = AdminManager()