xzbautomarkapi/bgtask/tasks.py

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

import json
import os
from typing import List, Dict, Any

from pydantic import Field
from sqlalchemy import update, text, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from common.const import GRADE_ORDER, OBJECTIVE_QUESTION_TYPES
from crud.base import ModelType
from crud.marktask import crud_task, crud_student_task, crud_student_answer
from crud.problem import crud_class_error_statistic, crud_student_push_record,\
    crud_student_error_statistic,crud_class_push_record
from crud.school import crud_grade, crud_system, crud_class
from crud.user import crud_teacher, crud_student
from db.asyncsession import LocalAsyncSession
from models.marktask import StudentMarkTask, StudentAnswer
from models.problem import ClassErrorQuestion, StudentErrorPushRecord, StudentErrorQuestion, ClassErrorPushRecord
from models.school import SchoolGrade, SchoolClass
from models.user import Teacher, Student
from schemas.app.task import BgUpdateTaskReviewInfo, BgUpdateStudentTaskInfo
from schemas.problem import StudentErrorPushRecordInDB, UpdateClassErrorQuestionItem, UpdateStudentErrorQuestionItem
from schemas.school.school import NewGrade


async def bgtask_delete_related_object(sid: int = 0, gid: int = 0, cid: int = 0):
    db = LocalAsyncSession()

    # 根据学校删除年级、班级、教师、学生等
    if sid and (not gid) and (not cid):
        # 1、删除年级
        await crud_grade.delete(db, where_clauses=[SchoolGrade.school_id == sid])
        # 2、删除班级
        await crud_class.delete(db, where_clauses=[SchoolClass.school_id == sid])
        # TODO: 删除教师、学生、作业、考试、阅卷任务等
        await crud_teacher.delete(db, where_clauses=[Teacher.school_id == sid])
        await crud_student.delete(db, where_clauses=[Student.school_id == sid])
    elif gid and (not cid):    # 如果是年级被删除，则删除该年级下的所有班级
        await crud_class.delete(db, where_clauses=[SchoolClass.grade_id == gid])
        # TODO: 删除学生、作业、考试、阅卷任务等
        await crud_student.delete(db, where_clauses=[Student.grade_id == gid])
    elif cid:    # 如果班级被删除了，则删除该班级下的所有学生
        # TODO: 删除学生、作业、考试、阅卷任务等
        await crud_student.delete(db, where_clauses=[Student.class_id == cid])
    else:
        pass


async def bgtask_delete_remote_file(filepath: str):
    urls = [x.strip() for x in filepath.split(";")]
    try:
        for url in urls:
            os.remove(url)
    except Exception as ex:
        print(f"[Error] Delete file: {str(ex)}")


async def bgtask_rank_score(task_id: int = Field(..., description="阅卷任务ID")):
    """
    更新学生排名
    """
    db = LocalAsyncSession()

    # 根据 task_id 查询 StudentMarkTask
    stmt = f"""SELECT id,
    CASE
    WHEN @prevRank = score THEN @curRank
    WHEN @prevRank := score THEN @curRank := @curRank + 1
    END AS rank
    FROM { StudentMarkTask.__tablename__ },
    (SELECT @curRank :=0, @prevRank := NULL) rb
    WHERE task_id={ task_id } AND is_completed=true AND score>0
    ORDER BY score desc;
    """
    current_rank = 1
    ranks = await db.execute(stmt)
    for item in ranks:
        current_rank += 1
        stmt = (update(StudentMarkTask).where(StudentMarkTask.id == item[0]).values(rank=item[1]))
        await crud_student_task.execute_v2(db, stmt)

    # 为0分记录设置排名
    stmt = (update(StudentMarkTask).where(StudentMarkTask.task_id == task_id,
                                          StudentMarkTask.score == 0).values(rank=current_rank))
    await crud_student_task.execute_v2(db, stmt)

    print(f"Calculate Rank OK!!")


async def bgtask_update_marktask(task_id: int, smt_has_marked: bool, smt_passed: bool,
                                 smt_score: float, diff_score: float):
    """
    更新阅卷任务
    smt_has_marked：是否已被批阅过？
    smt_passed：在更新之前是否及格？
    smt_score：StudentMarkTask的最终分数
    diff_score: 重复批阅时的分数差
    """
    print("Async Update MarkTask Starting!")
    print(
        f"the params: task_id={task_id}, smt_has_marked={smt_has_marked},smt_passed={smt_passed}, smt_score={smt_score},diff_score={diff_score} "
    )
    db = LocalAsyncSession()

    # 查询MarkTask是否存在
    db_obj = await crud_task.find_one(db, filters={"id": task_id})

    # 更新已批阅人数 / 及格人数
    # 及格人数更新逻辑
    # 1、如果StudentMarkTask本身是及格的，当新分数 <60 时，及格人数 -1；
    # 2、如果StudentMarkTask本身是不及格的，当新分数 >=60 时，及格人数 +1；
    mt_marked_amount = db_obj.marked_amount
    pass_amount = db_obj.pass_amount
    if not smt_has_marked:
        mt_marked_amount += 1    # 批阅人数
        if smt_score >= 60:
            pass_amount = db_obj.pass_amount + 1
    else:
        if smt_passed:
            if smt_score < 60:
                pass_amount = db_obj.pass_amount - 1
        else:
            if smt_score >= 60:
                pass_amount = db_obj.pass_amount + 1
    print("------------------")
    print(f"the mt_marked_amount={mt_marked_amount}")
    print(f"the pass_amount={pass_amount}")
    pass_rate = round(pass_amount / mt_marked_amount, 2)
    print(f"the pass_rate={pass_rate}")

    # 最高分 / 最低分
    stmt = select(func.max(StudentMarkTask.score), func.min(StudentMarkTask.score)).select_from(StudentMarkTask) \
        .where(StudentMarkTask.task_id == task_id, StudentMarkTask.is_completed == 1)
    high_score, low_score = list(await crud_student_task.execute_v2(db, stmt))[0]
    print("the high/low score: ", high_score, low_score)

    # 平均分
    avg_score = round((db_obj.avg_score * db_obj.marked_amount + diff_score) / mt_marked_amount, 2)

    # 批阅状态，0=未开始，1=进行中，2=已完成
    status = 2 if mt_marked_amount == db_obj.student_amount else 1

    # 更新
    mt = BgUpdateTaskReviewInfo(marked_amount=mt_marked_amount,
                                high_score=high_score,
                                low_score=low_score,
                                avg_score=avg_score,
                                pass_amount=pass_amount,
                                pass_rate=pass_rate,
                                status=status)
    await crud_task.update(db, db_obj, mt)

    # 更新StudentMarkTask排名
    # 1、如果StudentMarkTask未被批阅完成：
    #   1.1 增加MarkTask的批阅数量；
    #   1.2 当新增后的批阅数量等于学生数量时，表示MarkTask被批阅完成，则开始更新排名；
    # 2、如果StudentMarkTask批阅已完成，则属于重复批阅，如果两次批阅间存在分数差，则更新排名；
    if mt_marked_amount == db_obj.student_amount:    # MarkTask批阅完成时，更新排名
        await bgtask_rank_score(task_id)
    else:
        if diff_score and db_obj.status == 2:    # 如果重复批阅有分数差，且MarkTask已批阅完成时，更新排名
            await bgtask_rank_score(task_id)
    print("Async Update MarkTask Successfully!")


async def bgtask_update_student_marktask(task_id: int = Field(..., description="学生阅卷任务ID"),
                                         qtype: int = Field(..., description="试题类型,0：客观题,1: 主观题"),
                                         score: int = Field(..., description="多次批阅之间的得分差"),
                                         has_marked: bool = Field(..., description="是否已被批阅过"),
                                         editor_id: int = Field(..., description="最后编辑人ID"),
                                         editor_name: str = Field(..., description="最后编辑人姓名")):
    """
    数据更新流程：
    1、根据task_id更新StudentMarkTask，
    2、如果该学生的所有试题已经被批阅完，则根据StudentMarkTask.task_id 更新 MarkTask 的最高分/最低分/平均分...等信息。
    3、如果MarkTask批阅完成，则对StudentMarkTask进行排名。
    """
    print("Async Update StudentMarkTask Starting!")
    # new db session
    db = LocalAsyncSession()

    # ------------------ 更新学生阅卷任务-----------------------------------
    student_task = await crud_student_task.find_one(db, filters={"id": task_id})
    has_completed = student_task.is_completed    # 原始批阅状态
    smt_is_passed = True if student_task.score >= 60 else False    # 是否及格

    # 如果试题属于重复批阅，则不增加批阅数量
    if not has_marked:
        smt_marked_amount = student_task.marked_amount + 1
    else:
        smt_marked_amount = student_task.marked_amount

    # 学生任务更新信息
    smt_is_completed = smt_marked_amount == student_task.question_amount
    smt = BgUpdateStudentTaskInfo(marked_amount=smt_marked_amount,
                                  is_completed=smt_is_completed,
                                  score=student_task.score + score,
                                  editor_id=editor_id,
                                  editor_name=editor_name)
    # 按题型统计得分
    if qtype not in OBJECTIVE_QUESTION_TYPES:
        smt.subjective_score = student_task.subjective_score + score
    else:
        smt.objective_score = student_task.objective_score + score

    # 开始更新
    student_task = await crud_student_task.update(db, student_task, smt)
    # ------------------ 更新学生阅卷任务结束 ---------------------------

    # 如果单个学生被批阅完成，更新MarkTask信息
    if smt_is_completed:
        diff_score = student_task.score if not has_completed else score

        await bgtask_update_marktask(task_id=student_task.task_id,
                                     smt_has_marked=has_completed,
                                     smt_passed=smt_is_passed,
                                     smt_score=student_task.score,
                                     diff_score=diff_score)
    print("Async Update StudentMarkTask Successfully!")


async def bgtask_create_grade(sid: int, category: int):
    db = LocalAsyncSession()
    # 年级列表
    school_system = await crud_system.find_one(db,
                                               filters={"id": category},
                                               return_fields={"grades"})
    grade_set = set(school_system.grades.split(","))

    # 查询学校是否存在
    _, db_grades = await crud_grade.find_all(db, filters={"school_id": sid})
    db_grade_set = set([x.name for x in db_grades])
    diff_grades = grade_set - db_grade_set
    new_grades = [NewGrade(sid=sid, name=x, order=GRADE_ORDER[x]) for x in diff_grades]
    try:
        await crud_grade.insert_many(db, new_grades)
    except Exception as ex:
        print(f"AutoCreateGradeError: {str(ex)}")
    return


async def bgtask_delete_student_task_question(task_id: int):
    db = LocalAsyncSession()
    # 查询学生任务ID列表
    _, student_tasks = await crud_student_task.find_all(db,
                                                        filters={"task_id": task_id},
                                                        return_fields=["id"])
    student_task_ids = ",".join([str(x.id) for x in student_tasks])
    # 删除学生任务和答题
    await crud_student_task.delete(db, where_clauses=[StudentMarkTask.task_id == task_id])
    await crud_student_answer.delete(db, where_clauses=[StudentAnswer.task_id == task_id])
    # 删除学生错题统计
    print(8888888888888888)
    await crud_student_error_statistic.delete(
        db, where_clauses=[StudentErrorQuestion.task_id == task_id])
    # 删除班级错题统计
    print(9999999999999999)
    await crud_class_error_statistic.delete(db,
                                            where_clauses=[ClassErrorQuestion.task_id == task_id])


async def bgtask_update_class_teacher_student(cid: int,
                                              op: str = "add",
                                              teacher_amount: int = 0,
                                              student_amount: int = 0):
    """
    更新班级的教师数量和学生数量, op: 操作类型，add=增加, del=减少
    """
    db = LocalAsyncSession()

    if op == "del":
        teacher_amount = -teacher_amount
        student_amount = -student_amount

    stmt = (update(SchoolClass).where(SchoolClass.id == cid).values(
        teacher_amount=SchoolClass.teacher_amount + teacher_amount,
        student_amount=SchoolClass.student_amount + student_amount))
    await crud_class.execute_v2(db, stmt)
    print(
        f"Async Update Class OK! student_amount={student_amount}, teacher_amount={teacher_amount}")


async def update_student_error_statistic(student_id: int, task_type: str, error_count: int):
    """
    更新学生错题统计数据
    """
    # new db session
    db = LocalAsyncSession()

    db_error = await crud_student_error_statistic.find_one(db, filters={"student_id": student_id})
    update_dict = {"total_errors": db_error.total_errors + error_count}
    if task_type == "work":
        key = "work_error_count"
    else:
        key = "exam_error_count"

    update_dict[key] = getattr(db_error, key) + error_count
    update_dict["error_ratio"] = round(update_dict["total_errors"] / db_error.total_questions * 100,
                                       2)
    # 更新
    item = UpdateStudentErrorQuestionItem(**update_dict)
    await crud_student_error_statistic.update(db, db_error, item)


# 更新错题统计数据
async def bgtask_update_class_error_statistic(task_id: int, class_id: int, qid: List[int],
                                              count: int, old_score: float, new_score: float,
                                              has_marked: bool):
    # new db session
    db = LocalAsyncSession()
    # update
    db_error = await crud_class_error_statistic.find_one(db,
                                                         filters={
                                                             "task_id": task_id,
                                                             "class_id": class_id,
                                                             "qid": qid
                                                         })
    update_dict = {}
    print("Async Update Class-Error-Statistic Starting!")
    # 如果是错题，则更新错题数量和错题率
    if count:
        error_count = db_error.error_count + count
        error_ratio = round(error_count / db_error.student_count * 100, 2)
        update_dict["error_count"] = error_count
        update_dict["error_ratio"] = error_ratio

    # 更新答案分布
    if not db_error.answer_dist:
        answer_dist = {}
    else:
        answer_dist = json.loads(db_error.answer_dist)
    # 更新分数的人数
    if old_score != 0:
        answer_dist[old_score] -= 1
    if new_score not in answer_dist:
        answer_dist[new_score] = 1
    else:
        answer_dist[new_score] += 1
    if answer_dist:
        update_dict["answer_dist"] = json.dumps(answer_dist)

    # 更新
    if update_dict:
        try:
            item = UpdateClassErrorQuestionItem(**update_dict)
            await crud_class_error_statistic.update(db, db_error, item)
        except Exception as ex:
            print(f"UpdateClassError: {ex}")
    print("Async Update Class-Error-Statistic Successfully!")


# 批量更新错题统计数据
async def bgtask_batch_update_class_error_statistic(task_id: int, class_id: int,
                                                    questions: List[Dict[str, Any]]):
    """
    批量更新班级错题统计
    :param task_id: 阅卷任务ID, int
    :param class_id: 班级ID, int
    :param questions: 错题统计数据，list，形式：[{"qid": 1, "answer": "A"},...]
    """
    # new db session
    db = LocalAsyncSession()
    for q in questions:
        # update
        db_error = await crud_class_error_statistic.find_one(db,
                                                             filters={
                                                                 "task_id": task_id,
                                                                 "class_id": class_id,
                                                                 "qid": q["qid"]
                                                             })
        update_dict = {}
        print("Async Update[batch] Class-Error-Statistic Starting!")
        # 如果是错题，则更新错题数量和错题率
        error_count = db_error.error_count + 1
        error_ratio = round(error_count / db_error.student_count * 100, 2)
        update_dict["error_count"] = error_count
        update_dict["error_ratio"] = error_ratio

        # 更新答案分布
        answer_dist = {}
        if not db_error.answer_dist:
            answer_dist[q["answer"]] = 1
        else:
            answer_dist = json.loads(db_error.answer_dist)
            answer_dist[q["answer"]] = answer_dist.get(q["answer"], 0) + 1
        if answer_dist:
            update_dict["answer_dist"] = json.dumps(answer_dist)
        # 更新
        if update_dict:
            try:
                item = UpdateClassErrorQuestionItem(**update_dict)
                await crud_class_error_statistic.update(db, db_error, item)
            except Exception as ex:
                print(f"UpdateClassError: {ex}")
        print("Async Update[batch] Class-Error-Statistic Successfully!")


# 创建学生错题推送记录
async def bgtask_create_student_push_record(task: ModelType, db_class: ModelType,
                                            total_question: int, push_record_id: int,
                                            students: List[ModelType], current_user: Teacher):
    db = LocalAsyncSession()
    # 学生推送列表
    push_records = []

    # 查询错题ID
    total, db_errors = await crud_class_error_statistic.find_all(
        db,
        filters=[
            ClassErrorQuestion.task_id == task.id,
            ClassErrorQuestion.class_id == db_class.id,
            ClassErrorQuestion.error_count > 0,
        ],
        return_fields=["qid"])
    push_error_ids = [x.qid for x in db_errors]

    # 为每个学生创建推送记录
    for stu in students:
        obj_in = StudentErrorPushRecordInDB(student_id=stu.id,
                                            student_sno=stu.sno,
                                            student_name=stu.name,
                                            task_id=task.id,
                                            task_name=task.name,
                                            task_type=task.mtype,
                                            question_count=total_question,
                                            push_record_id=push_record_id,
                                            push_error_count=total,
                                            push_error_ids=push_error_ids,
                                            creator_id=current_user.id,
                                            creator_name=current_user.name,
                                            editor_id=current_user.id,
                                            editor_name=current_user.name)
        push_records.append(obj_in)

    await crud_student_push_record.insert_many(db, push_records)


# 克隆学生错题推送记录
async def bgtask_clone_student_push_record(db: AsyncSession, cls_prd: ModelType,
                                           students: List[ModelType]):
    # 从班级推送记录中查出一条记录
    db_prd = await crud_student_push_record.find_one(db, filters={"push_record_id": cls_prd.id})

    push_records = []
    for stu in students:
        obj_in = StudentErrorPushRecordInDB(student_id=stu.id,
                                            student_sno=stu.sno,
                                            student_name=stu.name,
                                            task_id=db_prd.task_id,
                                            task_name=db_prd.task_name,
                                            task_type=db_prd.task_type,
                                            question_count=db_prd.total_question,
                                            push_record_id=db_prd.push_record_id,
                                            push_error_count=db_prd.total,
                                            push_error_ids=db_prd.push_error_ids,
                                            creator_id=db_prd.creator_id,
                                            creator_name=db_prd.creator_name,
                                            editor_id=db_prd.editor_id,
                                            editor_name=db_prd.editor_name)
        push_records.append(obj_in)
    await crud_student_push_record.insert_many(db, push_records)


# 删除学生错题推送记录
async def bgtask_delete_student_push_record(db: AsyncSession,
                                            cls_prid: int,
                                            students: List[int] = None):
    _q = [StudentErrorPushRecord.push_record_id == cls_prid]
    if students:
        _q = [StudentErrorPushRecord.student_id.in_(students)]

    try:
        await crud_student_push_record.delete(db, where_clauses=_q)
    except Exception as ex:
        print(f"[ERROR] 删除学生推送列表失败! 错误信息: {str(ex)}")