从0到1搭建教务系统：选课与成绩录入模块源码解析

class Course(models.Model):
course_id = models.CharField(max_length=20, unique=True, verbose_name="课程编号")
name = models.CharField(max_length=100, verbose_name="课程名称")
teacher = models.ForeignKey(User, on_delete=models.CASCADE, related_name="taught_courses", verbose_name="授课教师")
credit = models.FloatField(verbose_name="学分")
capacity = models.IntegerField(verbose_name="选课容量")
enrolled_count = models.IntegerField(default=0, verbose_name="已选人数")
start_time = models.DateTimeField(verbose_name="开课时间")
end_time = models.DateTimeField(verbose_name="结课时间")

def __str__(self):
return f"{self.course_id} - {self.name}"

class StudentCourse(models.Model):
student = models.ForeignKey(User, on_delete=models.CASCADE, related_name="selected_courses", verbose_name="学生")
course = models.ForeignKey(Course, on_delete=models.CASCADE, related_name="enrolled_students", verbose_name="课程")
select_time = models.DateTimeField(auto_now_add=True, verbose_name="选课时间")

def check_time_conflict(student, new_course):
# 获取学生已选课程的时间范围
existing_courses = StudentCourse.objects.filter(student=student).values_list('course__start_time', 'course__end_time')

for start, end in existing_courses:
# 时间冲突判断：新课程时间与已有课程时间存在重叠
if (new_course.start_time < end) and (new_course.end_time > start):
return True
return False

def select_course(student, course_id):
try:
course = Course.objects.get(course_id=course_id)
except Course.DoesNotExist:
return "课程不存在"

# 校验课程容量
if course.enrolled_count >= course.capacity:
return "课程已满员"

# 校验时间冲突
if check_time_conflict(student, course):
return "选课时间冲突"

# 原子操作：避免并发选课导致超容
with transaction.atomic():
course.enrolled_count += 1
course.save()
StudentCourse.objects.create(student=student, course=course)

class Meta:
unique_together = ("student", "course")

def batch_import_grades(request, course_id):
if request.method != "POST":
return JsonResponse({"status": "error", "message": "仅支持POST请求"})

try:
course = Course.objects.get(course_id=course_id)
except Course.DoesNotExist:
return JsonResponse({"status": "error", "message": "课程不存在"})

# 验证当前用户是否为课程教师
if request.user != course.teacher:
return JsonResponse({"status": "error", "message": "无权限录入该课程成绩"})

file = request.FILES.get("grade_file")
if not file:
return JsonResponse({"status": "error", "message": "请上传成绩文件"})

# 读取Excel文件
df = pd.read_excel(file)
required_columns = ["学号", "成绩"]
if not all(col in df.columns for col in required_columns):
return JsonResponse({"status": "error", "message": "文件格式错误，需包含'学号'和'成绩'列"})

success_count = 0
error_count = 0
error_messages = []

for index, row in df.iterrows():
student_id = row["学号"]
score = row["成绩"]

try:
student = User.objects.get(username=student_id)
# 检查学生是否选过该课程
if not StudentCourse.objects.filter(student=student, course=course).exists():
error_messages.append(f"行{index+2}: 学生{student_id}未选该课程")
error_count += 1
continue

# 更新或创建成绩记录
grade, created = Grade.objects.update_or_create(
student=student,
course=course,
defaults={"score": score}
)
success_count += 1
except User.DoesNotExist:
error_messages.append(f"行{index+2}: 学生{student_id}不存在")
error_count += 1