Spaces:
Sleeping
Sleeping
File size: 4,947 Bytes
01757c2 | 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 132 | from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from ..database import get_db
from ..models import TokenUsage, Student
router = APIRouter(prefix="/api/analytics", tags=["Analytics"])
# GPT-4o-mini pricing (per 1M tokens)
INPUT_COST_PER_M = 0.15
OUTPUT_COST_PER_M = 0.60
EVENT_LABELS = {
"syllabus_analysis": "Syllabus Analysis",
"test_generation": "Test Paper Generation",
"answer_evaluation": "Answer Evaluation",
"performance_analysis": "Performance Analysis",
"targeted_test_generation": "Targeted Test Generation",
}
def calc_cost(input_tokens: int, output_tokens: int) -> float:
return round(
(input_tokens / 1_000_000) * INPUT_COST_PER_M
+ (output_tokens / 1_000_000) * OUTPUT_COST_PER_M,
6,
)
@router.get("/overview")
async def get_overview(db: AsyncSession = Depends(get_db)):
result = await db.execute(select(TokenUsage).order_by(TokenUsage.created_at.desc()))
rows = result.scalars().all()
total_input = sum(r.input_tokens for r in rows)
total_output = sum(r.output_tokens for r in rows)
total_tokens = sum(r.total_tokens for r in rows)
by_event: dict[str, dict] = {}
for r in rows:
et = r.event_type
if et not in by_event:
by_event[et] = {"event_type": et, "label": EVENT_LABELS.get(et, et),
"count": 0, "input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
by_event[et]["count"] += 1
by_event[et]["input_tokens"] += r.input_tokens
by_event[et]["output_tokens"] += r.output_tokens
by_event[et]["total_tokens"] += r.total_tokens
for et in by_event:
by_event[et]["cost_usd"] = calc_cost(by_event[et]["input_tokens"], by_event[et]["output_tokens"])
return {
"total_events": len(rows),
"total_input_tokens": total_input,
"total_output_tokens": total_output,
"total_tokens": total_tokens,
"estimated_cost_usd": calc_cost(total_input, total_output),
"by_event_type": sorted(by_event.values(), key=lambda x: x["total_tokens"], reverse=True),
"model": rows[0].model if rows else "—",
}
@router.get("/by-student")
async def get_by_student(db: AsyncSession = Depends(get_db)):
result = await db.execute(select(TokenUsage).where(TokenUsage.student_id.isnot(None)))
rows = result.scalars().all()
students_result = await db.execute(select(Student).order_by(Student.created_at.asc()))
students = students_result.scalars().all()
student_code_map = {s.id: f"STU-{i+1:03d}" for i, s in enumerate(students)}
by_student: dict[str, dict] = {}
for r in rows:
sid = r.student_id
if sid not in by_student:
by_student[sid] = {
"student_id": sid,
"student_name": r.student_name or "Unknown",
"student_code": student_code_map.get(sid, "—"),
"total_tokens": 0,
"input_tokens": 0,
"output_tokens": 0,
"event_count": 0,
"by_event": {},
}
by_student[sid]["total_tokens"] += r.total_tokens
by_student[sid]["input_tokens"] += r.input_tokens
by_student[sid]["output_tokens"] += r.output_tokens
by_student[sid]["event_count"] += 1
et = r.event_type
by_student[sid]["by_event"][et] = by_student[sid]["by_event"].get(et, 0) + r.total_tokens
for sid in by_student:
by_student[sid]["cost_usd"] = calc_cost(
by_student[sid]["input_tokens"], by_student[sid]["output_tokens"]
)
return sorted(by_student.values(), key=lambda x: x["total_tokens"], reverse=True)
@router.get("/events")
async def get_events(limit: int = 100, db: AsyncSession = Depends(get_db)):
result = await db.execute(
select(TokenUsage).order_by(TokenUsage.created_at.desc()).limit(limit)
)
rows = result.scalars().all()
students_result = await db.execute(select(Student).order_by(Student.created_at.asc()))
students = students_result.scalars().all()
student_code_map = {s.id: f"STU-{i+1:03d}" for i, s in enumerate(students)}
return [
{
"id": r.id,
"event_type": r.event_type,
"label": EVENT_LABELS.get(r.event_type, r.event_type),
"student_name": r.student_name,
"student_code": student_code_map.get(r.student_id, "—") if r.student_id else "—",
"subject": r.subject,
"grade": r.grade,
"input_tokens": r.input_tokens,
"output_tokens": r.output_tokens,
"total_tokens": r.total_tokens,
"cost_usd": calc_cost(r.input_tokens, r.output_tokens),
"model": r.model,
"extra_info": r.extra_info,
"created_at": r.created_at.isoformat(),
}
for r in rows
]
|