xzbautomarkapi/admin/api/endpoints/school/student.py

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

import copy
import os
from collections import defaultdict
from typing import Optional, Union

from fastapi import APIRouter, Query, Depends, Path, UploadFile, File
from openpyxl import load_workbook
from sqlalchemy import or_, and_, desc, asc
from sqlalchemy.ext.asyncio import AsyncSession
from starlette.background import BackgroundTasks

from admin.api.endpoints.school.utils import check_row, check_filetype
from bgtask.tasks import bgtask_update_class_teacher_student
from core.config import settings
from core.security import hashed_password
from crud.school import crud_school, crud_grade, crud_class
from crud.user import crud_student
from models.user import Admin, Student
from schemas.base import ReturnField
from schemas.school.student import StudentList, StudentDetail, NewStudent, StudentInDB, UpdateStudent
from utils.depends import get_async_db, get_current_user

router = APIRouter()


# 学生列表
@router.get("/students",
            response_model=StudentList,
            response_model_exclude_none=True,
            summary="学生列表")
async def get_students(page: Optional[int] = None,
                       size: Optional[int] = None,
                       sno: str = Query("", description="学号"),
                       name: str = Query("", description="学生姓名"),
                       tid: int = Query(0, description="学生ID"),
                       sid: int = Query(0, description="学校ID"),
                       gid: int = Query(0, description="年级ID"),
                       cid: int = Query(0, description="班级ID"),
                       order: str = Query("-id", description="排序字段,用逗号分隔,升降序以-判断"),
                       res: ReturnField = Query("", description="返回字段，取值: id,name,sno,phone"),
                       db: AsyncSession = Depends(get_async_db),
                       current_user: Admin = Depends(get_current_user)):
    # 过滤条件
    _q = {}
    if sno:
        _q["sno"] = sno
    if name:
        _q["name"] = name
    if tid:
        _q["id"] = tid
    if sid:
        _q["school_id"] = sid
    if gid:
        _q["grade_id"] = gid
    if cid:
        _q["class_id"] = cid
    if ((page is not None) and page >= 1) and ((size is not None) and size >= 1):
        offset = (page - 1) * size
    else:
        offset = size = None

    # 排序字段
    order_fields = []
    if order:
        for x in order.split(","):
            field = x.strip()
            if field:
                if field.startswith("-"):
                    order_fields.append(desc(getattr(Student, field[1:])))
                else:
                    order_fields.append(asc(getattr(Student, field)))

    total, items = await crud_student.find_all(db,
                                               filters=_q,
                                               offset=offset,
                                               limit=size,
                                               order_by=order_fields,
                                               return_fields=res)
    for x in items:
        if not x.age:
            x.age = ""
    return {"total": total, "data": items}


# 创建学生
@router.post("/students",
             response_model=StudentDetail,
             response_model_exclude_none=True,
             summary="创建学生")
async def create_student(info: NewStudent,
                         bgtask: BackgroundTasks,
                         db: AsyncSession = Depends(get_async_db),
                         current_user: Admin = Depends(get_current_user)):
    # 判断学校是否存在
    db_school = await crud_school.find_one(db,
                                           filters={"id": info.school_id},
                                           return_fields=["name", "category"])
    if not db_school:
        return {"errcode": 404, "mess": "学校不存在!"}

    # 判断年级是否存在
    db_grade = await crud_grade.find_one(db, filters={"id": info.grade_id}, return_fields=["name"])
    if not db_grade:
        return {"errcode": 404, "mess": "年级不存在!"}

    # 判断班级是否存在
    db_class = await crud_class.find_one(db,
                                         filters={"id": info.class_id},
                                         return_fields=["id", "name"])
    if not db_class:
        return {"errcode": 404, "mess": "班级不存在!"}

    # 判断是否存在同年级同学号
    existed = await crud_student.count(db,
                                       filters=[
                                           or_(
                                               and_(Student.school_id == info.school_id,
                                                    Student.sno == info.sno),
                                               Student.phone == info.phone)
                                       ])
    if existed:
        return {"errcode": 400, "mess": "学号或手机号重复!"}

    # 开始创建
    obj_in = StudentInDB(**info.dict(by_alias=True),
                         period=db_school.category,
                         school_name=db_school.name,
                         grade_name=db_grade.name,
                         class_name=db_class.name,
                         username=info.phone,
                         password=hashed_password(info.phone[-6:]),
                         creator_id=current_user.id,
                         creator_name=current_user.username,
                         editor_id=current_user.id,
                         editor_name=current_user.username)
    db_obj = await crud_student.insert_one(db, obj_in)

    # 异步更新班级的学生数量
    bgtask.add_task(bgtask_update_class_teacher_student, db_class.id, "add", 0, 1)

    return {"data": db_obj}


@router.get("/students/{sid}",
            response_model=StudentDetail,
            response_model_exclude_none=True,
            summary="学生详情")
async def get_student(sid: int = Path(..., description="学生ID"),
                      db: AsyncSession = Depends(get_async_db),
                      current_user: Admin = Depends(get_current_user)):
    db_obj = await crud_student.find_one(db, filters={"id": sid})
    return {"data": db_obj}


# 更新学生
@router.put("/students/{sid}",
            response_model=StudentDetail,
            response_model_exclude_none=True,
            summary="更新学生")
async def update_student(info: UpdateStudent,
                         bgtask: BackgroundTasks,
                         sid: int = Path(..., description="学生ID"),
                         db: AsyncSession = Depends(get_async_db),
                         current_user: Admin = Depends(get_current_user)):

    # 判断提交参数是否为空
    info_dict = info.dict(exclude_none=True)
    if not info_dict:
        return {"errcode": 400, "mess": "提交参数为空!"}

    # 判断学生是否存在
    db_obj = await crud_student.find_one(db, filters={"id": sid})
    if not db_obj:
        return {"errcode": 404, "mess": "学生不存在!"}

    # 判断学校是否存在
    if ("school_id" in info_dict) and (db_obj.school_id != info_dict["school_id"]):
        db_school = await crud_school.find_one(db,
                                               filters={"id": info.school_id},
                                               return_fields=["name", "category"])
        if not db_school:
            return {"errcode": 404, "mess": "学校不存在!"}
        else:
            info.school_name = db_school.name
            info.period = db_school.category

    # 判断年级是否存在
    if ("grade_id" in info_dict) and (db_obj.grade_id != info_dict["grade_id"]):
        db_grade = await crud_grade.find_one(db,
                                             filters={"id": info.grade_id},
                                             return_fields=["name"])
        if not db_grade:
            return {"errcode": 404, "mess": "年级不存在!"}
        else:
            info.grade_name = db_grade.name

    # 判断班级是否存在
    if ("class_id" in info_dict) and (db_obj.class_id != info_dict["class_id"]):
        db_class = await crud_class.find_one(db,
                                             filters={"id": info.class_id},
                                             return_fields=["name"])
        if not db_class:
            return {"errcode": 404, "mess": "班级不存在!"}
        else:
            info.class_name = db_class.name
            bgtask.add_task(bgtask_update_class_teacher_student, db_obj.class_id, "del", 0, 1)
            bgtask.add_task(bgtask_update_class_teacher_student, db_class.id, "add", 0, 1)

    # 判断手机号是否重复
    if ("phone" in info_dict) and (db_obj.phone != info_dict["phone"]):
        db_student = await crud_student.find_one(
            db, filters=[Student.id != sid, Student.phone == info.phone])
        if db_student:
            return {"errcode": 400, "mess": "手机号重复!"}
        info.username = info.phone[-6:]

    # 更新
    info.editor_id = current_user.id
    info.editor_name = current_user.username
    db_obj = await crud_student.update(db, db_obj, info)
    return {"data": db_obj}


# 删除学生
@router.delete("/students/{sid}",
               response_model=StudentDetail,
               response_model_exclude_none=True,
               summary="删除学生")
async def delete_student(bgtask: BackgroundTasks,
                         sid: Union[int,
                                    str] = Path(...,
                                                description="学生ID，批量删除传用逗号分隔ID(str)，单个删除传ID(int)"),
                         db: AsyncSession = Depends(get_async_db),
                         current_user: Admin = Depends(get_current_user)):
    # 查询所有学生
    if isinstance(sid, int):
        deleted_count = 1
        _q = [Student.id == sid]
    else:
        sids = [int(x.strip()) for x in sid.split(',') if x.strip()]
        deleted_count = len(sids)
        _q = [Student.id.in_(sids)]
    total, db_objs = await crud_student.find_all(db, filters=_q)
    # 删除
    for item in db_objs:
        await crud_student.delete(db, obj_id=item.id)
        bgtask.add_task(bgtask_update_class_teacher_student, item.class_id, "del", student_amount=1)
    return {
        "data": total,
        "mess": "success" if total == deleted_count else f"成功:{total}，失败: {deleted_count - total}"
    }


# 批量导入学生
@router.post("/students/bulk",
             response_model=StudentDetail,
             response_model_exclude_none=True,
             summary="批量导入学生")
async def import_student(bgtask: BackgroundTasks,
                         datafile: UploadFile = File(..., description="数据文件"),
                         db: AsyncSession = Depends(get_async_db),
                         current_user: Admin = Depends(get_current_user)):
    # 判断文件格式
    if not check_filetype(datafile.filename, ".xlsx"):
        return {"errcode": 400, "mess": "文件格式错误!"}

    # 把文件写入磁盘，再加载回来
    disk_file = os.path.join(settings.UPLOADER_PATH, datafile.filename)
    content = await datafile.read()
    with open(disk_file, "wb") as f:
        f.write(content)

    # 返回结果
    phones = []
    snos = []
    errors = []
    success = 0
    students = []
    counter = 0
    class_students = defaultdict(int)
    schools = {}
    grades = {}
    classes = {}

    # 使用openpyxl读取文件
    wb = load_workbook(disk_file)
    ws = wb.worksheets[0]
    for row in ws.iter_rows(min_row=2, max_col=ws.max_column, max_row=ws.max_row, values_only=True):
        row = await check_row(row, 9)
        if row is None:    # 空行
            continue
        elif not row:    # 字段不完整
            errors.append(f"第{counter}行: 某些单元格为空!")
            continue

        # 判断学校是否存在
        if row[6] not in schools:
            db_school = await crud_school.find_one(db,
                                                   filters={"name": row[6]},
                                                   return_fields=["id", "category"])
            if not db_school:
                errors.append(f"第{counter}行: 学校不存在!")
                continue
            else:
                schools[row[6]] = {"id": db_school.id, "category": db_school.category}

        # 判断年级是否存在
        grade_key = f"{row[6]}-{row[7]}"
        if grade_key not in grades:
            db_grade = await crud_grade.find_one(db,
                                                 filters={
                                                     "school_id": schools[row[6]]["id"],
                                                     "name": row[7]
                                                 },
                                                 return_fields=["id"])
            if not db_grade:
                errors.append(f"第{counter}行: 年级不存在!")
                continue
            else:
                grades[grade_key] = db_grade.id

        # 判断是否存在同名学生
        if (row[5] in phones) or (row[1] in snos):
            errors.append(f"第{counter}行: 学号或手机号重复!")
            continue
        else:
            phones.append(row[5])
            snos.append(row[1])
            # 判断数据库是否存在重复的学号或手机号
            db_student = await crud_student.count(
                db,
                filters=[
                    or_(and_(Student.school_id == schools[row[6]]["id"], Student.sno == row[1]),
                        Student.phone == row[5])
                ])
            if db_student:
                errors.append(f"第{counter}行: 学号或手机号重复!")
                continue

        # 判断班级是否存在
        class_key = f"{grade_key}-{row[8]}"
        if class_key not in classes:
            db_class = await crud_class.find_one(db,
                                                 filters={
                                                     "school_id": schools[row[6]]["id"],
                                                     "grade_id": grades[grade_key],
                                                     "name": row[8]
                                                 },
                                                 return_fields=["id"])
            if not db_class:
                errors.append(f"第{counter}行: 班级不存在!")
                continue
            else:
                classes[class_key] = db_class.id
        class_students[classes[class_key]] += 1

        # 创建学生
        obj_in = StudentInDB(sno=row[1],
                             username=str(row[5]).strip(),
                             password=hashed_password(str(row[5]).strip()[-6:]),
                             name=row[2],
                             sex=1 if (not row[3] or row[3] == "男") else 0,
                             age=row[4] if row[4] else 0,
                             phone=row[5],
                             sid=schools[row[6]]["id"],
                             school_name=row[6],
                             period=schools[row[6]]["category"],
                             gid=grades[grade_key],
                             grade_name=row[7],
                             cid=classes[class_key],
                             class_name=row[8],
                             creator_id=current_user.id,
                             creator_name=current_user.username,
                             editor_id=current_user.id,
                             editor_name=current_user.username)
        students.append(obj_in)
        success += 1
        counter += 1

        if counter == 50:
            try:
                await crud_student.insert_many(db, copy.deepcopy(students))
            except Exception as ex:
                print(f"[student] INSERT-ERROR: {str(ex)}, students: {students}")
                return {"errcode": 1, "mess": "入库失败，请联系技术人员解决！"}
            else:
                students.clear()
                counter = 0
    if counter:
        try:
            await crud_student.insert_many(db, students)
        except Exception as ex:
            print(f"[student] INSERT-ERROR: {str(ex)}, students: {students}")
            return {"errcode": 1, "mess": "入库失败，请联系技术人员解决！"}

    # 异步更新班级的教师数量
    for k, v in class_students.items():
        bgtask.add_task(bgtask_update_class_teacher_student, int(k), "add", student_amount=1)

    # 删除上传文件
    os.remove(disk_file)

    return {"data": {"success": success, "fail": len(errors), "errors": errors}}