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

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import copy
import os
from typing import Optional

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

from admin.api.endpoints.school.utils import check_row
from core.config import settings
from core.security import hashed_password
from crud.school import crud_class, crud_school, crud_grade
from crud.sysdata.role import crud_role
from crud.user import crud_teacher
from models.user import SysUser
from schemas.base import OrderByField, ReturnField
from schemas.school.teacher import (TeacherList, TeacherDetail, NewTeacher,
                                    TeacherInDB, UpdateTeacher)
from utils.depends import get_async_db, get_current_user

router = APIRouter()


# 教师列表
@router.get("/teachers",
            response_model=TeacherList,
            response_model_exclude_none=True,
            summary="教师列表")
async def get_teachers(
    page: Optional[int] = None,
    size: Optional[int] = None,
    name: str = Query("", description="教师姓名"),
    tid: int = Query(0, description="教师ID"),
    sid: int = Query(0, description="学校ID"),
    gid: int = Query(0, description="年级ID"),
    cid: str = Query("", description="班级ID"),
    rid: int = Query(0, description="职务(角色ID)"),
    sub: str = Query("", description="任教科目"),
    order: OrderByField = Query("-created_at",
                                description="排序字段,用逗号分隔,升降序以-判断,默认-created_at"),
    res: ReturnField = Query("", description="返回字段，默认列表展示字段, 自定义id,name,phone"),
    db: AsyncSession = Depends(get_async_db),
    current_user: SysUser = Depends(get_current_user)):
    # 过滤条件
    filters = []
    if name:
        filters.append(text(f"name = '{name}'"))
    if tid:
        filters.append(text(f"id = {tid}"))
    if sid:
        filters.append(text(f"school_id = {sid}"))
    if gid:
        filters.append(text(f"grade_id = {gid}"))
    if cid:
        filters.append(text(f"class_id LIKE '%{cid}%'"))
    if rid:
        filters.append(text(f"role_id = {rid}"))
    if sub:
        filters.append(text(f"subject = '{sub}'"))
    if ((page is not None) and page >= 1) and ((size is not None)
                                               and size >= 1):
        offset = (page - 1) * size
    else:
        offset = size = None

    # 排序
    if isinstance(order, str):
        order = [text(order)]
    total, items = await crud_teacher.find_all(db,
                                               filters=filters,
                                               offset=offset,
                                               limit=size,
                                               order_by=order,
                                               return_fields=res)
    return {"total": total, "data": items}


# 创建教师
@router.post("/teachers",
             response_model=TeacherDetail,
             response_model_exclude_none=True,
             summary="创建教师")
async def create_teacher(info: NewTeacher,
                         db: AsyncSession = Depends(get_async_db),
                         current_user: SysUser = Depends(get_current_user)):

    # 判断学校是否存在
    db_school = await crud_school.find_one(db, filters={"id": info.school_id})
    if not db_school:
        return {"errcode": 404, "mess": "学校不存在!"}

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

    # 判断角色是否存在
    db_role = await crud_role.find_one(db, filters={"id": info.role_id})
    if not db_role:
        return {"errcode": 404, "mess": "角色不存在!"}

    # 判断是否存在同年级同名教师
    db_teacher = await crud_teacher.find_one(db, filters={"phone": info.phone})
    if db_teacher:
        return {"errcode": 400, "mess": "手机号重复!"}

    # 判断班级是否存在
    class_ids = [x.strip() for x in info.class_id.split(",") if x.strip()]
    total, db_class = await crud_class.find_all(
        db, filters=[text(f"id IN ({info.class_id})")])
    if total != len(class_ids):
        return {"errcode": 400, "mess": "班级不存在!"}

    # 开始创建
    obj_in = TeacherInDB(**info.dict(by_alias=True),
                         school_name=db_school.name,
                         grade_name=db_grade.name,
                         class_name=",".join([x.name for x in db_class]),
                         role_name=db_role.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_teacher.insert_one(db, obj_in)
    return {"data": db_obj}


@router.get("/teachers/{tid}",
            response_model=TeacherDetail,
            response_model_exclude_none=True,
            summary="教师详情")
async def get_teacher(tid: int = Path(..., description="教师ID"),
                      db: AsyncSession = Depends(get_async_db),
                      current_user: SysUser = Depends(get_current_user)):
    db_obj = await crud_teacher.find_one(db, filters={"id": tid})
    return {"data": db_obj}


# 更新教师
@router.put("/teachers/{tid}",
            response_model=TeacherDetail,
            response_model_exclude_none=True,
            summary="更新教师")
async def update_teacher(info: UpdateTeacher,
                         tid: int = Path(..., description="教师ID"),
                         db: AsyncSession = Depends(get_async_db),
                         current_user: SysUser = Depends(get_current_user)):
    # 判断提交参数是否为空
    info_dict = info.dict(exclude_none=True)
    if not info_dict:
        return {"errcode": 400, "mess": "提交参数为空!"}

    # 判断教师是否存在
    db_obj = await crud_teacher.find_one(db, filters={"id": tid})
    if not db_obj:
        return {"errcode": 404, "mess": "教师不存在!"}

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

    # 判断学校是否存在
    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})
        if not db_school:
            return {"errcode": 404, "mess": "学校不存在!"}
        else:
            info.school_name = db_school.name

    # 判断年级是否存在
    if ("grade_id" in info_dict) and (db_obj.grade_id != info_dict["grade_id"]):
        db_grade = await crud_grade.find_one(db, {"id": info.grade_id})
        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"]):
        class_ids = [x.strip() for x in info.class_id.split(",") if x.strip()]
        total, db_class = await crud_class.find_all(
            db, filters=[text(f"id IN ({info.class_id})")])
        if total != len(class_ids):
            return {"errcode": 400, "mess": "班级不存在!"}
        else:
            info.class_name = ",".join([x.name for x in db_class])

    # 判断角色是否存在
    if ("role_id" in info_dict) and (db_obj.role_id != info_dict["role_id"]):
        db_role = await crud_role.find_one(db, filters={"id": info.role_id})
        if not db_role:
            return {"errcode": 404, "mess": "角色不存在!"}
        else:
            info.role_name = db_role.name

    # 更新
    info.editor_id = current_user.id
    info.editor_name = current_user.username
    db_obj = await crud_teacher.update(db, db_obj, info)

    return {"data": db_obj}


# 删除教师
@router.delete("/teachers/{tid}",
               response_model=TeacherDetail,
               response_model_exclude_none=True,
               summary="删除教师")
async def delete_teacher(bg_task: BackgroundTasks,
                         tid: int = Path(..., description="教师ID"),
                         db: AsyncSession = Depends(get_async_db),
                         current_user: SysUser = Depends(get_current_user)):
    existed = await crud_teacher.count(db, {"id": tid})
    if not existed:
        return {"errcode": 404, "mess": "教师不存在!"}
    else:
        await crud_teacher.delete(db, obj_id=tid)
        # TODO: 删除关联数据
        # bg_task.add_task(delete_related_object, db, cid=id)
        return {"data": None}


# 批量导入教师
@router.post("/teachers/bulk",
             response_model=TeacherDetail,
             response_model_exclude_none=True,
             summary="批量导入教师")
async def import_teacher(datafile: UploadFile = File(..., description="数据文件"),
                         db: AsyncSession = Depends(get_async_db),
                         current_user: SysUser = Depends(get_current_user)):
    # 判断文件格式
    if not datafile.filename.endswith(".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)

    # 返回结果
    errors = []
    success = 0
    teachers = []
    counter = 0

    # 使用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, 10)
        if row is None:    # 空行
            continue
        elif not row:    # 字段不完整
            errors.append(f"第{row[0]}行: 某些单元格为空!")
            continue

        # 判断学校是否存在
        db_school = await crud_school.find_one(db, filters={"name": row[5]})
        if not db_school:
            errors.append(f"第{row[0]}行: 学校不存在!")
            continue

        # 判断年级是否存在
        db_grade = await crud_grade.find_one(db,
                                             filters={
                                                 "school_id": db_school.id,
                                                 "name": row[6]
                                             })
        if not db_grade:
            errors.append(f"第{row[0]}行: 年级不存在!")
            continue

        # 判断班级是否存在
        class_names = [x.strip() for x in row[7].split(",") if x.strip()]
        total, db_class = await crud_class.find_all(
            db,
            filters=[
                text(
                    f"school_id = {db_school.id} AND grade_id = {db_grade.id} AND name IN ('{row[7]}')"
                )
            ])
        if total != len(class_names):
            errors.append(f"第{row[0]}行: 班级不存在!")
            continue
        class_ids = ",".join([str(x.id) for x in db_class])

        # 判断角色是否存在
        db_role = await crud_role.find_one(db, filters={"name": row[8]})
        if not db_role:
            errors.append(f"第{row[0]}行: 角色不存在!")
            continue

        # 判断是否存在同名教师
        existed = await crud_teacher.count(db, filters={"phone": row[4]})
        if existed:
            errors.append(f"第{row[0]}行: 手机号重复!")
            continue

        # 创建教师对象
        obj_in = TeacherInDB(username=row[4],
                             password=hashed_password(row[4][-6:]),
                             name=row[1],
                             sex=1 if row[2] == "男" else 0,
                             age=row[3],
                             phone=row[4],
                             sid=db_school.id,
                             school_name=row[5],
                             gid=db_grade.id,
                             grade_name=row[6],
                             cid=class_ids,
                             class_name=row[7],
                             rid=db_role.id,
                             role_name=row[8],
                             sub=row[9],
                             creator_id=current_user.id,
                             creator_name=current_user.username,
                             editor_id=current_user.id,
                             editor_name=current_user.username)
        teachers.append(obj_in)
        success += 1
        counter += 1
        if counter == 50:
            await crud_teacher.insert_many(db, copy.deepcopy(teachers))
            teachers.clear()
            counter = 0
    if counter:
        await crud_teacher.insert_many(db, teachers)

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

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