96 lines
3.4 KiB
Python
96 lines
3.4 KiB
Python
"""学习分析。"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from database import get_db
|
|
from dependencies import get_current_user, user_permissions
|
|
from models import AttemptSession, ErrorEntry, User
|
|
from schemas import AnalyticsOut, DiagnosisOut, MetricItem
|
|
from services.mastery_engine import action_for, snapshot
|
|
|
|
|
|
router = APIRouter(tags=["analytics"])
|
|
|
|
|
|
def build_analytics(db: Session, user_id: int) -> AnalyticsOut:
|
|
"""按用户聚合学习分析：本人分析与管理员查看指定用户共用同一份逻辑。"""
|
|
sessions = (
|
|
db.query(AttemptSession)
|
|
.filter(AttemptSession.user_id == user_id)
|
|
.order_by(AttemptSession.id.asc())
|
|
.all()
|
|
)
|
|
total_answers = sum(s.total for s in sessions)
|
|
total_correct = sum(s.correct for s in sessions)
|
|
accuracy = total_correct / total_answers * 100 if total_answers else 0.0
|
|
|
|
last = sessions[-1] if sessions else None
|
|
errors_count = (
|
|
db.query(ErrorEntry).filter(ErrorEntry.user_id == user_id).count()
|
|
)
|
|
knowledge_rows = snapshot(db, user_id)
|
|
weak_count = sum(1 for k in knowledge_rows if k.mastery < 60)
|
|
|
|
metrics = [
|
|
MetricItem(
|
|
label="历史正确率",
|
|
value=f"{accuracy:.1f}%" if total_answers else "0%",
|
|
),
|
|
MetricItem(
|
|
label="最近一次",
|
|
value=f"{last.correct}/{last.total}" if last else "—",
|
|
),
|
|
MetricItem(label="错题数量", value=str(errors_count)),
|
|
MetricItem(label="需重点干预", value=str(weak_count)),
|
|
]
|
|
|
|
error_counts: dict[str, int] = {}
|
|
for row in (
|
|
db.query(ErrorEntry.knowledge_name, ErrorEntry.knowledge_names)
|
|
.filter(ErrorEntry.user_id == user_id)
|
|
.all()
|
|
):
|
|
try:
|
|
names = json.loads(row[1] or "[]")
|
|
if not isinstance(names, list) or not names:
|
|
names = [row[0]]
|
|
except (TypeError, ValueError):
|
|
names = [row[0]]
|
|
for name in {str(name) for name in names}:
|
|
error_counts[name] = error_counts.get(name, 0) + 1
|
|
|
|
diagnosis = [
|
|
DiagnosisOut(
|
|
name=k.name,
|
|
mastery=k.mastery,
|
|
error_count=error_counts.get(k.name, 0),
|
|
action=action_for(k.mastery),
|
|
)
|
|
for k in sorted(knowledge_rows, key=lambda x: x.mastery)
|
|
]
|
|
return AnalyticsOut(metrics=metrics, diagnosis=diagnosis)
|
|
|
|
|
|
@router.get("/analytics", response_model=AnalyticsOut)
|
|
def get_analytics(
|
|
user_id: int | None = None,
|
|
db: Session = Depends(get_db),
|
|
user: User = Depends(get_current_user),
|
|
):
|
|
"""默认返回本人学习分析；管理员可用 user_id 查看指定用户（用户管理子页面）。"""
|
|
permissions = user_permissions(db, user)
|
|
target_id = user.id
|
|
if user_id is not None and user_id != user.id:
|
|
if "users:manage" not in permissions:
|
|
raise HTTPException(status_code=403, detail="没有权限查看该用户的学习分析")
|
|
target_id = user_id
|
|
elif "learning:use" not in permissions and "users:manage" not in permissions:
|
|
raise HTTPException(status_code=403, detail="没有权限执行该操作")
|
|
if db.get(User, target_id) is None:
|
|
raise HTTPException(status_code=404, detail="用户不存在")
|
|
return build_analytics(db, target_id)
|