Spaces:
Sleeping
Sleeping
File size: 4,543 Bytes
03faf26 1cff1e5 03faf26 1cff1e5 03faf26 1cff1e5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | from datetime import datetime, timezone
from database import get_db
from models.collections import RESULTS, SESSIONS, USERS
from utils.helpers import str_objectid, str_objectids
async def get_student_history(user_id: str) -> list:
"""Get all interview reports for a student."""
db = get_db()
cursor = db[RESULTS].find({"user_id": user_id}).sort("completed_at", -1)
docs = await cursor.to_list(length=50)
results = []
for doc in docs:
results.append({
"session_id": doc.get("session_id"),
"overall_score": doc.get("overall_score", 0),
"total_questions": doc.get("total_questions", 0),
"completed_at": doc.get("completed_at", ""),
"role_title": doc.get("role_title", ""),
})
return results
async def get_admin_analytics() -> dict:
"""Get aggregated analytics for admin dashboard."""
db = get_db()
# Total students
total_students = await db[USERS].count_documents({"role": "student"})
# Users with in-progress interview sessions.
active_user_ids = await db[SESSIONS].distinct("user_id", {"status": "in_progress"})
live_users = len([uid for uid in active_user_ids if uid])
# New students created since start of current UTC day.
day_start = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0).isoformat()
new_users_today = await db[USERS].count_documents({"role": "student", "created_at": {"$gte": day_start}})
# Total interviews
total_interviews = await db[RESULTS].count_documents({})
# Average score
pipeline = [
{"$group": {"_id": None, "avg_score": {"$avg": "$overall_score"}}},
]
avg_result = await db[RESULTS].aggregate(pipeline).to_list(length=1)
avg_score = round(avg_result[0]["avg_score"], 1) if avg_result else 0
# Top performers
top_pipeline = [
{"$group": {
"_id": "$user_id",
"avg_score": {"$avg": "$overall_score"},
"interview_count": {"$sum": 1},
}},
{"$sort": {"avg_score": -1}},
{"$limit": 10},
]
top_results = await db[RESULTS].aggregate(top_pipeline).to_list(length=10)
top_performers = []
for r in top_results:
user = await db[USERS].find_one({"_id": __import__("bson").ObjectId(r["_id"])})
if not user:
# user_id might be stored as string
user = await db[USERS].find_one({"email": {"$exists": True}})
top_performers.append({
"user_id": r["_id"],
"name": user.get("name", "Unknown") if user else "Unknown",
"avg_score": round(r["avg_score"], 1),
"interview_count": r["interview_count"],
})
# Common weak areas
weakness_pipeline = [
{"$unwind": "$weaknesses"},
{"$group": {"_id": "$weaknesses", "count": {"$sum": 1}}},
{"$sort": {"count": -1}},
{"$limit": 10},
]
weakness_results = await db[RESULTS].aggregate(weakness_pipeline).to_list(length=10)
common_weak = [w["_id"] for w in weakness_results]
return {
"total_students": total_students,
"live_users": live_users,
"new_users_today": new_users_today,
"total_interviews": total_interviews,
"average_score": avg_score,
"top_performers": top_performers,
"common_weak_areas": common_weak,
}
async def get_student_analytics(user_id: str) -> dict:
"""Get analytics for a specific student."""
db = get_db()
results = await db[RESULTS].find({"user_id": user_id}).to_list(length=100)
if not results:
return {
"total_interviews": 0,
"average_score": 0,
"best_score": 0,
"worst_score": 0,
"weak_topics": [],
"strong_topics": [],
}
scores = [r.get("overall_score", 0) for r in results]
all_weaknesses = []
all_strengths = []
for r in results:
all_weaknesses.extend(r.get("weaknesses", []))
all_strengths.extend(r.get("strengths", []))
# Count frequencies
from collections import Counter
weak_counts = Counter(all_weaknesses)
strong_counts = Counter(all_strengths)
return {
"total_interviews": len(results),
"average_score": round(sum(scores) / len(scores), 1),
"best_score": max(scores),
"worst_score": min(scores),
"weak_topics": [w for w, _ in weak_counts.most_common(5)],
"strong_topics": [s for s, _ in strong_counts.most_common(5)],
}
|