test/dao/analysis.py

from fastapi import Query,Depends
from sqlalchemy import func, case
from model.student_info import Student
from model.employments import Employments
from model.scores import Scores
from sqlalchemy.orm import Session

#student_info学生基本信息
#动态年龄范围查询：支持用户输入年龄阈值及比较条件（如大于、小于、等于、区间等），动态查询符合条件的学员信息
def student_info_age(db: Session,min_age:int ,max_age:int ,operator:str,):
    if operator not in("大于","大于等于","小于","小于等于","区间"):
        raise ValueError("输入比较格式不正确，请在规定范围内选择输入")
    else:
        q = db.query(Student)
        if operator == "大于":
            q = q.filter(Student.age > min_age)
        elif operator == "大于等于":
            q = q.filter(Student.age >= min_age)
        elif operator == "小于":
            q = q.filter(Student.age < min_age)
        elif operator == "小于等于":
            q = q.filter(Student.age <= min_age)
        else:  # 区间
            q = q.filter(Student.age >= min_age, Student.age <= max_age)
        print(q.all())
        return q.all()

#多维度班级统计：统计每个班级的总人数，以及按性别（男、女）细分的人数分布
def student_info_class(db: Session):
    s1 = (db.query(Student.class_name,
                   func.count(Student.id).label("总人数"),
                   func.sum(case((Student.gender=="男",1),else_=0)).label("男"),
                   func.sum(case((Student.gender=="女",1),else_=0)).label("女"))
          .group_by(Student.class_name).all())

    return [
        {
            "class_name": row.class_name,
            "总人数": row.总人数,
            "男": row.男,
            "女": row.女,
        }for row in s1
    ]



    #student_score学生成绩
#查询每次考试成绩都在输入分数线（如80分）以上的学生的编号、姓名和成绩。
def student_score(db: Session,s_n : int):
    s1 = db.query(Scores.id).group_by(Scores.id).having(func.min(Scores.score)>=s_n).all()
    s1_ids = [row[0] for row in s1]
    if not s1_ids:
        return []
    s2 = (db.query(Student.id,Student.name,Scores.exam_round,Scores.score)
          .join(Student, Scores.id == Student.id)
          .filter(Scores.id.in_(s1_ids)).all())
    return [{"id": i.id, "name": i.name, "exam_round": i.exam_round, "score": i.score}
        for i in s2]
#查询有输入指定次数（如两次(包含两次)）以上不及格的学生的姓名、班级和不及格成绩明细。
def student_score60(db: Session,n : int ):
    s1 = (db.query(Scores.id).group_by(Scores.id)
    .having(func.sum(case((Scores.score < 60, 1), else_=0)) >=n)
          .all())
    s1_ids = [row[0] for row in s1]
    if not s1_ids:
        return []
    s2 = (db.query(Student.name,Student.class_name,Scores.exam_round,Scores.score)
          .join(Student, Scores.id == Student.id)
          .filter(Scores.id.in_(s1_ids)).all())
    return [{"id": i.name, "name": i.class_name, "exam_round": i.exam_round, "score": i.score}
        for i in s2]
#统计每次考试每个班级的平均分，并支持按分数从高到低或从低到高动态排序。
def student_score_class(db: Session,order: str = "desc"):
    s1 = (db.query(Scores.exam_round,
            Student.class_name,
            func.avg(Scores.score).label("平均分"),).join(Student, Scores.id == Student.id)
          .group_by(Student.class_name,Scores.exam_round))
    if order == "desc":
        res = s1.order_by(func.avg(Scores.score).desc()).all()
    else:
        res = s1.order_by(func.avg(Scores.score).asc()).all()
    return [row._asdict() for row in res]


    #student_employments
#查询查询就业薪资排名前n的学生
def student_employments_salary(db: Session,n:int):
    s1 = db.query(Employments).order_by(Employments.sal.desc()).limit(n).all()
    return s1

#统计每个学生的就业时长
def student_employments_time(db: Session):
    time = []
    all_ = db.query(Employments).all()
    for employments in all_:
        if employments.offer_review  is None:
            time.append({"name":employments.student.name,
                         "class_name": employments.student.class_name,
                         "status": "未就业"})
        elif employments.offer_open and employments.offer_review:
            time.append({"name":employments.student.name,
                         "class_name": employments.student.class_name,
                         "time": (employments.offer_review-employments.offer_open).days})
    return time

#统计每个班级的平均就业时长（仅统计进入就业阶段，即有就业开放时间的学生）
def student_employments_class_time(db: Session):
    s1 = (db.query(Student.class_name,
                   func.avg(
                       func.unix_timestamp(Employments.offer_review) -
                       func.unix_timestamp(Employments.offer_open))
                   .label("avg_time"))
                    .join(Student, Employments.id == Student.id)
                    .filter(Employments.offer_review.isnot(None))
                    .group_by(Student.class_name).all())
    time = []
    for i in s1:
        days = round(float(i.avg_time) / 86400, 2) if i.avg_time else 0
        time.append({
            "class_name": i.class_name,
            "平均时长（天）": days
        })
    return time