xzbautomarkapi/bgtask/problem.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from typing import Any, Dict, List

from sqlalchemy import update

from crud.paper import crud_question
from crud.problem import crud_class_error_statistic, crud_student_error_statistic
from db.asyncsession import LocalAsyncSession
from models.paper import PaperQuestion
from models.problem import StudentErrorQuestion
from schemas.problem import ClassErrorQuestionInDB, StudentErrorQuestionInDB


# 创建班级错题统计
async def bgtask_create_class_error_statistic(school_id: int, grade_id: int, task_id: int,
                                              task_name: str, task_type: str, class_id: int,
                                              student_count: int, paper_id: int, school_name: str,
                                              grade_name: str, class_name: str):
    print("[bgTask] Create Class-Error-Statistics Starting!")
    db = LocalAsyncSession()
    # 根据试卷ID获取所有的客观题
    total, db_questions = await crud_question.find_all(db,
                                                       filters=[PaperQuestion.pid == paper_id],
                                                       return_fields=["id"])
    errors = []
    for x in db_questions:
        temp = ClassErrorQuestionInDB(qid=x.id,
                                      school_id=school_id,
                                      grade_id=grade_id,
                                      class_id=class_id,
                                      student_count=student_count,
                                      task_id=task_id,
                                      task_type=task_type,
                                      task_name=task_name,
                                      school_name=school_name,
                                      grade_name=grade_name,
                                      class_name=class_name)
        errors.append(temp)
    try:
        await crud_class_error_statistic.insert_many(db, errors)
    except Exception as ex:
        print(f"[ERROR]-[bgTask] 创建错题失败！原因：{str(ex)}")
    print("[bgTask] Create Class-Error-Statistics Successfully!")


async def bgtask_create_student_error_statistic(student_errors: List[Dict[str, Any]] = None):
    """
    创建每个学生的错题统计记录
    """
    db = LocalAsyncSession()

    print("[bgTask] Create Student-Error-Statistics Starting!")
    for stu in student_errors:
        # 查询学生错题统计是否存在
        db_error = await crud_student_error_statistic.find_one(
            db, filters={"student_id": stu["student_id"]})

        # 如果不存在就新建，否则就更新
        if not db_error:
            temp = StudentErrorQuestionInDB(**stu)
            try:
                await crud_student_error_statistic.insert_one(db, temp)
            except Exception as ex:
                print(f"[ERROR]-[bgTask] 创建学生错题统计失败！原因：{str(ex)}")
        else:
            total_questions = db_error.total_questions + stu["total_questions"]
            error_ratio = round(db_error.total_errors / total_questions * 100, 2)
            stmt = (update(StudentErrorQuestion).where(
                StudentErrorQuestion.student_id == stu["student_id"]).values(
                    total_questions=total_questions, error_ratio=error_ratio))
            await crud_student_error_statistic.increase(db, stmt)

    print("[bgTask] Create Student-Error-Statistics Successfully!")