322 lines
10 KiB
Python
322 lines
10 KiB
Python
from collections import defaultdict
|
||
from dataclasses import dataclass
|
||
from math import ceil
|
||
from typing import Literal
|
||
|
||
from django.db.models import Count
|
||
from ninja.errors import HttpError
|
||
|
||
from account.models import RoleChoices, User
|
||
from task.models import Task, TaskTypeChoices
|
||
|
||
from .models import Submission
|
||
|
||
GradebookTaskType = Literal["tutorial", "challenge"]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class GradebookFilters:
|
||
classname: str
|
||
task_type: GradebookTaskType | None = None
|
||
username: str | None = None
|
||
include_all_tasks: bool = False
|
||
|
||
|
||
def _score(value):
|
||
rounded = round(float(value or 0), 2)
|
||
if rounded == 0:
|
||
return 0.0
|
||
return rounded
|
||
|
||
|
||
def _grade_for_rank(rank: int, student_count: int) -> str:
|
||
if rank <= ceil(0.30 * student_count):
|
||
return "A"
|
||
if rank <= ceil(0.70 * student_count):
|
||
return "B"
|
||
if rank <= ceil(0.90 * student_count):
|
||
return "C"
|
||
if rank <= ceil(0.95 * student_count):
|
||
return "D"
|
||
return "E"
|
||
|
||
|
||
def _assessment_grade_for_rank(rank: int, student_count: int) -> str:
|
||
# 考核等级：A10% B20% C40% D20% E10%，按最高三次提交均分排名
|
||
if rank <= ceil(0.10 * student_count):
|
||
return "A"
|
||
if rank <= ceil(0.30 * student_count):
|
||
return "B"
|
||
if rank <= ceil(0.70 * student_count):
|
||
return "C"
|
||
if rank <= ceil(0.90 * student_count):
|
||
return "D"
|
||
return "E"
|
||
|
||
|
||
# 考核分数：各等级对应的分数段（高分→低分），段内按名次线性铺开
|
||
# 最高分封顶 99，不出现满分 100
|
||
ASSESSMENT_SCORE_BANDS = {
|
||
"A": (92, 99),
|
||
"B": (84, 91),
|
||
"C": (76, 83),
|
||
"D": (68, 75),
|
||
"E": (60, 67),
|
||
}
|
||
|
||
|
||
def _assessment_score(index_in_grade: int, grade_size: int, grade: str) -> int:
|
||
low, high = ASSESSMENT_SCORE_BANDS[grade]
|
||
if grade_size <= 1:
|
||
return high
|
||
# index_in_grade 从 0 开始：段内第 1 名拿段顶，最后 1 名拿段底
|
||
return round(high - (index_in_grade / (grade_size - 1)) * (high - low))
|
||
|
||
|
||
def _task_sort_key(task):
|
||
type_order = 0 if task.task_type == TaskTypeChoices.TUTORIAL else 1
|
||
return (type_order, task.display, task.id)
|
||
|
||
|
||
def _task_type_label(task_type: str) -> str:
|
||
return "教程" if task_type == TaskTypeChoices.TUTORIAL else "挑战"
|
||
|
||
|
||
def _csv_number(value):
|
||
if value == "":
|
||
return ""
|
||
number = float(value)
|
||
if number.is_integer():
|
||
return str(int(number))
|
||
return f"{number:.2f}".rstrip("0").rstrip(".")
|
||
|
||
|
||
def _classes():
|
||
return list(
|
||
User.objects.filter(role__in=(RoleChoices.NORMAL, RoleChoices.ADMIN))
|
||
.exclude(classname="")
|
||
.values_list("classname", flat=True)
|
||
.distinct()
|
||
.order_by("classname")
|
||
)
|
||
|
||
|
||
def _task_csv_header(task):
|
||
return f"{_task_type_label(task['task_type'])}{task['display']}-{task['title']}"
|
||
|
||
|
||
def build_gradebook(filters: GradebookFilters):
|
||
classname = filters.classname.strip() if filters.classname else ""
|
||
if not classname:
|
||
raise HttpError(400, "请选择班级")
|
||
if filters.task_type not in (None, "tutorial", "challenge"):
|
||
raise HttpError(400, "无效的任务类型")
|
||
|
||
classes = _classes()
|
||
class_students = list(
|
||
User.objects.filter(
|
||
role__in=(RoleChoices.NORMAL, RoleChoices.ADMIN), classname=classname
|
||
)
|
||
.order_by("username", "id")
|
||
.only("id", "username", "classname")
|
||
)
|
||
class_student_ids = [student.id for student in class_students]
|
||
student_count = len(class_student_ids)
|
||
coverage_threshold_count = ceil(student_count * 0.5) if student_count else 0
|
||
|
||
task_submission_qs = Submission.objects.filter(user_id__in=class_student_ids)
|
||
if filters.task_type:
|
||
task_submission_qs = task_submission_qs.filter(task__task_type=filters.task_type)
|
||
|
||
submitted_counts = {
|
||
row["task_id"]: row["submitted_count"]
|
||
for row in task_submission_qs.values("task_id").annotate(
|
||
submitted_count=Count("user_id", distinct=True)
|
||
)
|
||
}
|
||
task_map = {
|
||
task.id: task
|
||
for task in Task.objects.filter(id__in=submitted_counts.keys()).only(
|
||
"id",
|
||
"display",
|
||
"title",
|
||
"task_type",
|
||
)
|
||
}
|
||
|
||
all_tasks = []
|
||
for task in sorted(task_map.values(), key=_task_sort_key):
|
||
submitted_count = submitted_counts[task.id]
|
||
included = student_count > 0 and submitted_count >= coverage_threshold_count
|
||
all_tasks.append(
|
||
{
|
||
"id": task.id,
|
||
"display": task.display,
|
||
"title": task.title,
|
||
"task_type": task.task_type,
|
||
"submitted_count": submitted_count,
|
||
"coverage": _score(submitted_count / student_count)
|
||
if student_count
|
||
else 0.0,
|
||
"included": included,
|
||
}
|
||
)
|
||
|
||
tasks = [task for task in all_tasks if filters.include_all_tasks or task["included"]]
|
||
visible_task_ids = [task["id"] for task in tasks]
|
||
included_task_ids = {task["id"] for task in tasks if task["included"]}
|
||
|
||
best_by_pair = {}
|
||
if class_student_ids and visible_task_ids:
|
||
submissions = (
|
||
Submission.objects.filter(
|
||
user_id__in=class_student_ids,
|
||
task_id__in=visible_task_ids,
|
||
)
|
||
.select_related("task")
|
||
.only("id", "user_id", "task_id", "score", "created", "task__task_type")
|
||
.order_by("user_id", "task_id", "-score", "-created", "-id")
|
||
)
|
||
for submission in submissions:
|
||
best_by_pair.setdefault((submission.user_id, submission.task_id), submission)
|
||
|
||
rows = []
|
||
for student in class_students:
|
||
scores = {}
|
||
tutorial_total = 0.0
|
||
challenge_total = 0.0
|
||
submitted_scores = []
|
||
submitted_task_count = 0
|
||
missing_task_count = 0
|
||
|
||
for task in tasks:
|
||
submission = best_by_pair.get((student.id, task["id"]))
|
||
if submission:
|
||
cell_score = _score(submission.score)
|
||
scores[task["id"]] = {
|
||
"score": cell_score,
|
||
"submitted": True,
|
||
"submission_id": submission.id,
|
||
}
|
||
else:
|
||
cell_score = 0.0
|
||
scores[task["id"]] = {
|
||
"score": 0.0,
|
||
"submitted": False,
|
||
"submission_id": None,
|
||
}
|
||
|
||
if not task["included"]:
|
||
continue
|
||
if submission:
|
||
submitted_task_count += 1
|
||
submitted_scores.append(cell_score)
|
||
else:
|
||
missing_task_count += 1
|
||
if task["task_type"] == TaskTypeChoices.TUTORIAL:
|
||
tutorial_total += cell_score
|
||
else:
|
||
challenge_total += cell_score
|
||
|
||
tutorial_total = _score(tutorial_total)
|
||
challenge_total = _score(challenge_total)
|
||
top3 = sorted(submitted_scores, reverse=True)[:3]
|
||
top3_score = _score(sum(top3) / len(top3)) if top3 else 0.0
|
||
total_score = _score(tutorial_total + challenge_total)
|
||
average_score = (
|
||
_score(total_score / len(included_task_ids)) if included_task_ids else None
|
||
)
|
||
rows.append(
|
||
{
|
||
"user_id": student.id,
|
||
"username": student.username,
|
||
"classname": student.classname,
|
||
"rank": 0,
|
||
"grade": "E",
|
||
"assessment_grade": "E",
|
||
"assessment_score": 60,
|
||
"top3_score": top3_score,
|
||
"scores": scores,
|
||
"tutorial_total": tutorial_total,
|
||
"challenge_total": challenge_total,
|
||
"total_score": total_score,
|
||
"average_score": average_score,
|
||
"submitted_task_count": submitted_task_count,
|
||
"missing_task_count": missing_task_count,
|
||
}
|
||
)
|
||
|
||
rows.sort(key=lambda row: (-row["total_score"], row["username"]))
|
||
for index, row in enumerate(rows, start=1):
|
||
row["rank"] = index
|
||
row["grade"] = _grade_for_rank(index, student_count)
|
||
|
||
# 考核等级：按最高三次提交均分单独排名后定级（不改变上面的总分排名）
|
||
assessment_order = sorted(
|
||
rows, key=lambda row: (-row["top3_score"], row["username"])
|
||
)
|
||
for index, row in enumerate(assessment_order, start=1):
|
||
row["assessment_grade"] = _assessment_grade_for_rank(index, student_count)
|
||
|
||
# 考核分数：在各等级段内按名次（沿用最高三次均分排序）线性铺开到对应分数段
|
||
grade_groups = defaultdict(list)
|
||
for row in assessment_order:
|
||
grade_groups[row["assessment_grade"]].append(row)
|
||
for grade, group in grade_groups.items():
|
||
grade_size = len(group)
|
||
for index_in_grade, row in enumerate(group):
|
||
row["assessment_score"] = _assessment_score(
|
||
index_in_grade, grade_size, grade
|
||
)
|
||
|
||
username = filters.username.strip().lower() if filters.username else ""
|
||
if username:
|
||
rows = [row for row in rows if username in row["username"].lower()]
|
||
|
||
return {
|
||
"classname": classname,
|
||
"classes": classes,
|
||
"task_count": len(tasks),
|
||
"included_task_count": len(included_task_ids),
|
||
"student_count": student_count,
|
||
"coverage_threshold_count": coverage_threshold_count,
|
||
"tasks": tasks,
|
||
"rows": rows,
|
||
}
|
||
|
||
|
||
def gradebook_csv_rows(gradebook):
|
||
tasks = gradebook["tasks"]
|
||
yield [
|
||
"排名",
|
||
"平时等级",
|
||
"考核等级",
|
||
"考核分数",
|
||
"用户名",
|
||
"班级",
|
||
*[_task_csv_header(task) for task in tasks],
|
||
"教程合计",
|
||
"挑战合计",
|
||
"总分",
|
||
"平均分",
|
||
"已提交任务数",
|
||
"未提交任务数",
|
||
]
|
||
|
||
for row in gradebook["rows"]:
|
||
yield [
|
||
row["rank"],
|
||
row["grade"],
|
||
row["assessment_grade"],
|
||
row["assessment_score"],
|
||
row["username"],
|
||
row["classname"],
|
||
*[_csv_number(row["scores"][task["id"]]["score"]) for task in tasks],
|
||
_csv_number(row["tutorial_total"]),
|
||
_csv_number(row["challenge_total"]),
|
||
_csv_number(row["total_score"]),
|
||
"" if row["average_score"] is None else _csv_number(row["average_score"]),
|
||
row["submitted_task_count"],
|
||
row["missing_task_count"],
|
||
]
|