from utils .auth import auth_required, requires_role from flask import Blueprint, jsonify, g from db.model import User, Course, Enrollment, Chat, db from sqlalchemy import select, func, desc, and_ from datetime import datetime, timedelta from constants import UserRole admin = Blueprint('admin', __name__) @admin.route('/stats/users', methods=['GET']) @auth_required() @requires_role([UserRole.ADMIN]) def get_user_stats(): """ Get total users and authors count. Only accessible by admin users. """ try: # Get total users total_users = db.session.execute( select(func.count()).select_from(User) ).scalar() # Get authors (users who have created courses) distinct_authors_count = db.session.execute( select(func.count(func.distinct(Course.authorID))) ).scalar() return jsonify({ 'stats': { 'totalUsers': total_users, 'totalAuthors': distinct_authors_count } }), 200 except Exception as e: return jsonify({'message': f'An error occurred: {str(e)}'}), 500 @admin.route('/stats/enrollments', methods=['GET']) @auth_required() @requires_role([UserRole.ADMIN]) def get_enrollment_stats(): """ Get course enrollment and discussion statistics. Only accessible by admin users. """ try: # Get enrollment and user counts enrollment_stats = db.session.execute( select( func.count(Enrollment.id).label('total_enrollments'), func.count(func.distinct(Enrollment.userID)).label('enrolled_users') ) .select_from(Enrollment) ).first() # Get course-wise enrollment counts course_stats = db.session.execute( select( Course.name, func.count(Enrollment.id).label('enrollment_count') ) .join(Course, Course.id == Enrollment.courseID) .group_by(Course.id) .order_by(desc('enrollment_count')) ).all() return jsonify({ 'stats': { 'totalEnrollments': enrollment_stats.total_enrollments, 'totalEnrolledUsers': enrollment_stats.enrolled_users, 'courseEnrollments': [{ 'courseName': stat.name, 'enrollmentCount': stat.enrollment_count } for stat in course_stats] } }), 200 except Exception as e: return jsonify({'message': f'An error occurred: {str(e)}'}), 500 @admin.route('/stats/discussions', methods=['GET']) @auth_required() @requires_role([UserRole.ADMIN]) def get_discussion_stats(): """ Get chat room activity statistics. Only accessible by admin users. """ try: # Get activity for last 24 hours twenty_four_hours_ago = datetime.now() - timedelta(hours=24) # Get active rooms and their stats active_rooms = db.session.execute( select( Course.name, func.count(Chat.id).label('message_count'), func.count(func.distinct(Chat.userID)).label('active_users') ) .join(Course, Course.id == Chat.courseID) .where(Chat.chatDate >= twenty_four_hours_ago) .group_by(Course.id) .order_by(desc('message_count')) ).all() # Get total active rooms total_active_rooms = len(active_rooms) # Get most active room most_active_room = None if active_rooms: most_active = active_rooms[0] most_active_room = { 'name': most_active.name, 'messageCount': most_active.message_count, 'activeUsers': most_active.active_users } # Get total active users across all rooms total_active_users = db.session.execute( select(func.count(func.distinct(Chat.userID))) .where(Chat.chatDate >= twenty_four_hours_ago) ).scalar() return jsonify({ 'stats': { 'totalActiveRooms': total_active_rooms, 'totalActiveUsers': total_active_users, 'mostActiveRoom': most_active_room, 'activeRooms': [{ 'roomName': room.name, 'messageCount': room.message_count, 'activeUsers': room.active_users } for room in active_rooms] } }), 200 except Exception as e: return jsonify({'message': f'An error occurred: {str(e)}'}), 500