DoubaoTeam/dao/statistics.py

from schemas.statistics import NoExist
from sqlalchemy import func
from time import time
#`GET /statistics/students/age-over-30`
#查询所有超过 30 岁的学员信息

def age_over_thirty(Student,db):
    student_list=db.query(Student).filter(Student.age>30,Student.is_deleted==0).all()
    if not student_list :
        raise NoExist('无目标年龄大于30岁')
    # 把每个学生对象需要返回的字段，明确放进普通字典。
    result = []
    for student in student_list:
        result.append({
            "id": student.id,
            "student_no": student.student_no,
            "student_name": student.student_name,
            "class_id": student.class_id,
            "consultant_id": student.consultant_id,
            "native_place": student.native_place,
            "graduation_school": student.graduation_school,
            "major": student.major,
            "enrollment_date": student.enrollment_date,
            "graduation_date": student.graduation_date,
            "education": student.education,
            "age": student.age,
            "gender": student.gender,
            "create_at": student.create_at,
            "update_at": student.update_at,
            "is_deleted": student.is_deleted,
        })
    return result

def gender_count(Student,db):
    result=db.query(Student.class_id,Student.gender,func.count(Student.id)).filter(Student.is_deleted==0).group_by(Student.class_id,Student.gender).all()
    if not result :
        raise NoExist('班里招点人吧，要倒闭了')
    dict1 = {}
    for i,j,k in result:
        if i not in dict1 :
            dict1[i]={'男':0,'女':0,'总人数':0}
        dict1[i][j]+=k
        dict1[i]['总人数']+=k
    return dict1
#查询每次考试成绩都在 80 分以上的学生编号、姓名和成绩
def score_over_eighty(Score,Student,db):
    student_list=db.query(Student.id, Student.student_name)\
                  .join(Score,Score.student_id==Student.id).filter(Student.is_deleted==0,Score.is_deleted==0).group_by(Student.id,Student.student_name)\
                  .having(func.min(Score.score)>80).all()
    if not student_list :
        raise NoExist('全是学渣')
    list1 = []
    for i in student_list:
        student_list2 = db.query(Student.student_no, Student.student_name,Score.score).join(Score,Score.student_id==Student.id).filter(i.id==Student.id,Score.is_deleted==0).all()
        if student_list2 :
            for student_no, student_name, score in student_list2:
                list1.append({
                    "student_no": student_no,
                    "student_name": student_name,
                    "score": float(score) if score is not None else None,
                })
    return list1
#查询有两次以上不及格的学生姓名、班级和不及格成绩
def twice_error(Score,Student,db):
    student_list=db.query(Score.student_id).filter(Score.score<60,Score.is_deleted==0).group_by(Score.student_id).having(func.count(Score.id)>2).all()
    if not student_list :
        raise NoExist('全是学霸')
    list1=[]
    for i in student_list:
        result=db.query(Student.student_name,Student.class_id,Score.score).join(Score, Student.id==Score.student_id)\
        .filter(i.student_id==Score.student_id,Score.score<60,Student.is_deleted==0,Score.is_deleted==0).all()
        for student_name, class_id, score in result:
            list1.append({
                "student_name": student_name,
                "class_id": class_id,
                "score": float(score) if score is not None else None,
            })
    return list1

#统计每次考试每个班级的平均分，并按照平均分从高到低排序
def avg_class(Score,Student,db):
    student_list = db.query(Student.class_id,Score.exam_seq,func.avg(Score.score)).join(Score, Student.id == Score.student_id).filter(Score.is_deleted==0,Student.is_deleted==0).group_by(Student.class_id,Score.exam_seq).order_by(func.avg(Score.score).desc()).all()
    if not student_list :
        raise NoExist('招个人吧')
    result = []
    for class_id, exam_seq, average_score in student_list:
        result.append({
            "class_id": class_id,
            "exam_seq": exam_seq,
            "average_score": float(average_score) if average_score is not None else None,
        })
    return result

#统计就业薪资最高的前五名学生姓名、班级、就业时间、就业公司
def max_salary(Student,Employment,db):
    student_list = db.query(Student.student_name,Student.class_id,Employment.offer_date,Employment.company_name).join(Employment, Student.id == Employment.student_id).filter(Student.is_deleted==0,Employment.is_deleted==0).order_by(Employment.salary.desc()).limit(5).all()
    if not student_list :
        raise NoExist('没就业的啊兄弟？')
    result = []
    for student_name, class_id, offer_date, company_name in student_list:
        result.append({
            "student_name": student_name,
            "class_id": class_id,
            "offer_date": offer_date,
            "company_name": company_name,
        })
    return result

# - 统计每个学生的就业时长
# - 就业时长 = `offer_date - employment_open_date`
def student_market(Employment,db):
    student_list = db.query(Employment.student_id,func.datediff(Employment.offer_date,Employment.employment_open_date)).filter(Employment.is_deleted==0).all()
    result = []
    for student_id, duration_days in student_list:
        result.append({
            "student_id": student_id,
            "duration_days": duration_days,
        })
    return result

# - 统计每个班级的平均就业时长
# - 只统计有就业开放时间的学生
def avg_emptime(Employment,Student,db):
    student_list = (db.query(Student.class_id,\
                            func.avg(func.datediff(Employment.offer_date,Employment.employment_open_date)))\
                    .join(Employment, Student.id==Employment.student_id).filter(Student.is_deleted==0,Employment.is_deleted==0,Employment.employment_open_date.is_not(None)))\
    .group_by(Student.class_id).all()
    if not student_list :
        raise NoExist('一个就业都没有吗')
    result = []
    for class_id, average_days in student_list:
        result.append({
            "class_id": class_id,
            "average_days": float(average_days) if average_days is not None else None,
        })
    return result