xzbautomarkapi/admin/monitor/paper_consumer.py

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

import json
import os
import random
import time
from multiprocessing import Process, Queue
from shutil import move

from sqlalchemy import create_engine, and_,func
from sqlalchemy.orm import sessionmaker

from core.config import settings
from models.marktask import *
from models.paper import *
from models.problem import *

from utils.imgtool import crop_img_local, crop_img_remote
from utils.fileuploader import ossfile_uploader
from .recpaper import RecPaper
from utils.cv2img import get_std_points,rec_ans

engine = create_engine(settings.SYNC_MYSQL_URI)
session = sessionmaker(bind=engine)


def upload_simgs(simgs):
    """
    """
    urls = []
    for img in simgs:
        ossfile = os.path.split(img)[-1]
        url = ossfile_uploader.upload_from_file(img,ossfile,{"Content-Type":"image/png"})
        urls.append(url)
    return urls


def get_marktask(task_id):
    """
    """
    dbsession = session()
    task_id = int(task_id)
    task = dbsession.query(MarkTask).get(task_id)
    dbsession.close()
    if task:
        return task
    return None


def get_paper(pid):
    """
    """
    dbsession = session()
    pid = int(pid)
    paper = dbsession.query(Paper).get(pid)
    dbsession.close()
    if paper:
        return paper
    return None


def get_paper_pieces(pid):
    """
    """
    dbsession = session()
    pid = int(pid)
    ques = dbsession.query(PaperQuestion).filter(and_(PaperQuestion.pid == pid,PaperQuestion.usage.in_([1,2]))).all()
    dbsession.close()
    if ques:
        return ques
    return None


def get_student_marktask(task_id, sno):
    """
    """
    dbsession = session()
    task_id = int(task_id)
    smarktask = dbsession.query(StudentMarkTask).filter(
        and_(StudentMarkTask.task_id == task_id, StudentMarkTask.student_sno == sno)).scalar()
    dbsession.close()
    if smarktask:
        return smarktask
    return None


def update_student_marktask(task_id, sno, imgs):
    """
    """
    dbsession = session()
    task = dbsession.query(StudentMarkTask).filter(StudentMarkTask.task_id==task_id)\
        .filter(StudentMarkTask.student_sno==sno).one()
    task.pimgs = json.dumps(imgs)
    dbsession.commit()
    dbsession.close()


def get_student_answer(student_task_id,student_id,qid):
    """
    """
    dbsession = session()
    student_answer = dbsession.query(StudentAnswer).filter(
        and_(
            StudentAnswer.student_task_id==student_task_id,
            StudentAnswer.student_id==student_id,
            StudentAnswer.qid==qid,
        )).scalar()
    dbsession.close()
    if student_answer:
        return student_answer
    return None


def create_student_answer(**info):
    """
    """
    dbsession = session()
    answer = StudentAnswer(**info)
    dbsession.add(answer)
    dbsession.commit()
    dbsession.refresh(answer)
    dbsession.close()
    return answer

def update_student_answer(said,info):
    """更新学生答卷
    """
    dbsession = session()
    dbsession.query(StudentAnswer).filter(StudentAnswer.id==said).update(info)
    dbsession.commit()
    dbsession.close()

def update_task(tid,info):
    """更新阅卷任务
    """
    dbsession = session()
    dbsession.query(MarkTask).filter(MarkTask.id==tid).update(info)
    dbsession.commit()
    dbsession.close()

def update_student_marktask_status(said,info):
    """更新学生任务
    """
    dbsession = session()
    dbsession.query(StudentMarkTask).filter(StudentMarkTask.id==said).update(info)
    dbsession.commit()
    dbsession.close()

def get_uploaded_amount(task_id):
    """获取已上传数量
    """
    dbsession = session()
    allsmt = dbsession.query(func.count(StudentMarkTask.id)).filter(StudentMarkTask.status==2).scalar()
    dbsession.close()
    return allsmt

def update_class_error_questions(task_id,qid,answer,error_count,minus_answer):
    """更新班级错题记录
    """
    dbsession = session()
    db_class_error = dbsession.query(ClassErrorQuestion).filter(and_(ClassErrorQuestion.task_id==task_id,ClassErrorQuestion.qid==qid)).one()
    if db_class_error:
        #单题班级错题率
        error_count = db_class_error.error_count + error_count
        error_ratio = round(error_count / db_class_error.student_count * 100, 2)
        #错题分布
        answer_dist = json.loads(db_class_error.answer_dist) if db_class_error.answer_dist else {}
        if not answer_dist:
            answer_dist[answer] = 1
        else:
            answer_dist[answer] = answer_dist.get(answer, 0) + 1
            if minus_answer and answer_dist.get(minus_answer):
                answer_dist[minus_answer] = answer_dist[minus_answer] - 1

        db_class_error.error_count = error_count
        db_class_error.error_ratio = error_ratio
        db_class_error.answer_dist = json.dumps(answer_dist)

        dbsession.add(db_class_error)
        dbsession.commit()
        dbsession.refresh(db_class_error)
        dbsession.close()

def update_student_error_questions(task,smarktask,error_count):
    """更新学生错题统计
    """
    task_id = task.id
    student_task_id = smarktask.id
    student_id = smarktask.student_id

    dbsession = session()
    db_student_error = dbsession.query(StudentErrorQuestion).filter(
        and_(
            StudentErrorQuestion.task_id==task_id,
            StudentErrorQuestion.student_task_id==student_task_id,
            StudentErrorQuestion.student_id==student_id
        )).one_or_none()

    if db_student_error:
        work_error_count = db_student_error.work_error_count + error_count
        total_errors = work_error_count
        error_ratio = round(total_errors/db_student_error.total_questions,2)*100

        db_student_error.work_error_count = work_error_count
        db_student_error.total_errors = total_errors
        db_student_error.error_ratio = error_ratio

        dbsession.add(db_student_error)
        dbsession.commit()
        dbsession.refresh(db_student_error)
        dbsession.close()

def produce_papers(q):
    """
    """
    simgs = []
    while True:
        if not os.path.exists(settings.PAPERS_PATH):
            os.makedirs(settings.PAPERS_PATH)
        imglist = os.listdir(settings.PAPERS_PATH)
        imglist.sort()
        for pimg in imglist:
            if pimg.startswith("T"):
                pfile = os.path.join(settings.PAPERS_PATH, pimg)
            if os.path.isdir(pfile):
                ppimgs = os.listdir(pfile)
                if ppimgs:
                    ppimgs.sort()
                    task_id = os.path.split(pfile)[-1].split("-")[0].lstrip("T")
                    task = get_marktask(task_id)

                    if task:
                        pages = task.pages

                        for ppimg in ppimgs:
                            src = os.path.join(pfile, ppimg)
                            if os.path.isfile(src):
                                workspace = os.path.join(pfile, "workspace")
                                if not os.path.exists(workspace):
                                    os.makedirs(workspace)
                                target = os.path.join(workspace, ppimg)

                                move(src, target)
                                simgs.append(target)
                                if len(simgs) < pages:
                                    continue
                                else:
                                    q.put(simgs)
                                    simgs = []
        time.sleep(0.1)


def rec_papers(simgs, task):
    """
    """
    task_id = task.id
    pid = task.pid
    ans_points = json.loads(task.ans_points)

    # 识别考号和答案
    ans_img = simgs[0]
    std_points = get_std_points(ans_img)
    std_x = std_points[0]["x"]
    std_y = std_points[0]["y"]

    x = ans_points["x"] + std_x
    y = ans_points["y"] + std_y
    w = ans_points["w"]
    h = ans_points["h"]
    cut_point = (x, y, x + w, y + h)

    ans_img_card = crop_img_local(ans_img, cut_point)

    # 别考号和客观题
    qno_list = []
    # 获取标准切割的试题
    std_ques = get_paper_pieces(pid)
    for que in std_ques:
        if que.qtype in ["单选题","多选题"]:
            qno_list.append(que.qno)

    sno = random.choice(["010001","010002","010003","010004","010005"])

    # 单个学生阅卷任务
    smarktask = get_student_marktask(task_id, sno)

    if smarktask:
        pimgs = upload_simgs(simgs)
        update_student_marktask_status(smarktask.id,{"status":1,"pimgs":pimgs})
        # 生产学生作答库
        for que in std_ques:
            page = que.page
            qno = que.qno
            sqno = que.sqno
            qtype = que.qtype
            points = que.points
            std_ans = que.answer
            std_score = que.score
            
            marked_score = 0
            incorrect = 0
            status = 0

            answer = ""
            if qtype in ["单选题","多选题"]:
                std_ans_points = que.std_points
                answer = rec_ans(ans_img_card,std_ans_points)
                print(qno,answer,99999999999999)
                status = 1
                if answer != std_ans:
                    marked_score = 0
                    incorrect = 1
                else:
                    marked_score = std_score

            ans_imgs = []
            for point in points:
                x = point["x"] + std_x
                y = point["y"] + std_y
                w = point["w"]
                h = point["h"]
                cut_point = (x, y, x + w, y + h)
                url = crop_img_remote(simgs[page], cut_point, qno)
                ans_imgs.append(url)

            answerdict = {
                "student_id": smarktask.student_id,
                "student_sno": smarktask.student_sno,
                "student_name": smarktask.student_name,
                "class_id": task.class_id,
                "task_id": task_id,
                "task_name": task.name,
                "student_task_id": smarktask.id,
                "pid": pid,
                "pno": smarktask.pno,
                "pimgs": smarktask.pimgs,
                "pname": smarktask.pname,
                "qimg": ans_imgs[0],
                "qtype": qtype,
                "mtype": smarktask.mtype,
                "qno": qno,
                "sqno": sqno,
                "answer": answer,
                "score": que.score,
                "qid": que.id,
                "stem": que.stem,
                "score": std_score,
                "std_answer": std_ans,
                "status": status,
                "incorrect": incorrect,
                "marked_score": marked_score,
                "creator_id": 1,
                "creator_name": "admin",
                "editor_id": 1,
                "editor_name": "admin",
                "school_name": smarktask.school_name,
                "grade_name": smarktask.grade_name,
                "class_name": smarktask.class_name
            }
            error_count = 0
            minus_answer = ""
            student_error_count = 0

            student_answer = get_student_answer(smarktask.id,smarktask.student_id,que.id)
            if not student_answer:
                student_answer = create_student_answer(**answerdict)
                if student_answer.incorrect:
                    error_count += 1
                    student_error_count += 1
            else:
                if student_answer.incorrect != incorrect:
                    if student_answer.incorrect:
                        if error_count:
                            error_count += -1
                        else:
                            error_count = 0
                        #学生错题
                        if student_error_count:
                            student_error_count += -1
                        else:
                            student_error_count = 0
                    else:
                        error_count += 1
                        student_error_count += 1

                if qtype in ["单选题","多选题"]:
                    if student_answer.answer != answer:
                        minus_answer = student_answer.answer
                update_student_answer(student_answer.id,answerdict)

            if qtype in ["单选题","多选题"]:
                #更新班级错题
                update_class_error_questions(task_id,que.id,answer,error_count,minus_answer)
                #更新学生错题
                update_student_error_questions(task,smarktask,student_error_count)

        update_student_marktask_status(smarktask.id,{"status":2})
        uploaded_count = get_uploaded_amount(task_id)
        update_task(task_id,{"uploaded_amount":uploaded_count})

        for img in simgs:
            os.remove(img)


def consumer_papers(q):
    """
    """
    simgs = []
    while True:
        simgs = q.get()
        if simgs:
            print(f"start consumer simgs:{simgs}")
            pfile = simgs[0]
            task_id = pfile.split("workspace")[0].split("-")[0].split("/")[-1].lstrip("T")
            task = get_marktask(task_id)
            if task:
                rec_papers(simgs, task)

        time.sleep(0.1)


if __name__ == "__main__":
    q = Queue()
    p1 = Process(target=produce_papers, args=(q,))
    p1.start()
    for i in range(4):
        p2 = Process(target=consumer_papers, args=(q,))
        p2.start()