student_system/dao/age_analysis.py

from fastapi import HTTPException

from model.classinfo import ClassInfo
from model.employment import Employment
from model.score import Score
from model.student import Student
from sqlalchemy.orm import Session

from sqlalchemy import case, func, text, and_


def analysis_student_age(db: Session, min_age: int | None = None, max_age: int | None = None):
    #按年龄区间（>=min 且 <=max）查询学生，区间可缺省。
    if min_age is not None and min_age < 0:
        raise HTTPException(status_code=400, detail="最小年龄不能小于0")
    elif min_age is not None and max_age is not None and min_age > max_age:
        raise HTTPException(status_code=400, detail="最小年龄不能大于最大年龄")
    elif min_age is None and max_age is None :
        data_age= db.query(Student).all()
    elif min_age is not None and max_age is None:
        data_age = db.query(Student).filter(Student.age >= min_age).all()
    elif max_age is not None and min_age is None:
        data_age = db.query(Student).filter(Student.age <= max_age).all()
    elif min_age is not None and max_age is not None:
        data_age = db.query(Student).filter(and_(Student.age >= min_age, Student.age <= max_age)).all()
    return data_age


def analysis_class_gender(db: Session):
    #每个班级总人数及男、女人数。
    data= (db.query(
            Student.class_id,
            func.count(Student.id).label("total"),
            func.sum(case((Student.gender == "男", 1), else_=0)).label("gender_male"),
            func.sum(case((Student.gender == "女", 1), else_=0)).label("gender_female"),
        ).group_by(Student.class_id).all())
    return data

def analysis_score_over(db: Session, score_over: float = 0):
    #每次考试成绩都在分数线以上的学生的编号、姓名和成绩。
    qualified_student_ids = (
        db.query(Score.student_id)
        .group_by(Score.student_id)
        .having(func.min(Score.score) >= score_over)
    )
    data=(
        db.query(Student.id, Student.name, Score.score)
        .join(Score, Student.id == Score.student_id)
        .filter(Student.id.in_(qualified_student_ids))
        .all()
    )
    return data

def analysis_score_not_qualified(db: Session, not_qualified_count: int = 1):
    #有指定次数以上不及格（<60）的学生的姓名、班级和不及格成绩明细。
    not_qualified_students = (db.query(Score.student_id)
                              .filter(Score.score < 60)
                              .group_by(Score.student_id)
                              .having(func.count(Score.student_id) >= not_qualified_count))
    data=(db.query(Student.id, Student.name, Student.class_id, Score.score)
          .join(Score, Student.id == Score.student_id)
          .filter(Student.id.in_(not_qualified_students), Score.score < 60).all())
    return data


def analysis_avg_score_order(db: Session, sort_type: str = "asc"):
    #每次考试每个班级的平均分，asc/desc 排序。
    if sort_type not in ("asc", "desc"):
        raise HTTPException(status_code=400, detail="sort 参数只能为 asc 或 desc")
    avg_score = func.avg(Score.score).label("avg_score")
    query = (
        db.query(Score.exam_seq, ClassInfo.class_name, avg_score)
        .join(Student, Score.student_id == Student.id)
        .join(ClassInfo, Student.class_id == ClassInfo.id)
        .group_by(Score.exam_seq, ClassInfo.class_name)
    )
    if sort_type == "asc":
        data = query.order_by(avg_score.asc()).all()
    else:
        data = query.order_by(avg_score.desc()).all()
    return data


def analysis_employment_salary_order(db: Session, n: int = 1):
    #就业薪资排名 Top N 的学生姓名、班级、就业时间和就业公司
    data=(db.query(
            Student.name,
            Student.class_id,
            Employment.employment_open_time,
            Employment.company_name,
            Employment.salary)
          .join(Employment, Employment.stu_id == Student.id)
          .order_by(Employment.salary.desc()).limit(n).all())
    return data

def analysis_student_employment_time(db: Session):
    #每个学生的就业时长（offer下发时间 - 就业开放时间，单位秒）。
    data=(
        db.query(
            Employment.stu_id,
            func.timestampdiff(
                text("SECOND"),
                Employment.employment_open_time,
                Employment.offer_time,
            ).label("time_diff"),
        )
        .filter(
            Employment.employment_open_time.isnot(None),
            Employment.offer_time.isnot(None),
        )
        .all()
    )
    return data

def avg_employment_time(db: Session):
    #每个班级的平均就业时长（仅统计有就业开放时间的学生，单位秒）
    return (
        db.query(
            Student.class_id,
            func.avg(
                func.timestampdiff(
                    text("SECOND"),
                    Employment.employment_open_time,
                    Employment.offer_time,
                )
            ).label("time_diff"),
        )
        .join(Employment, Employment.stu_id == Student.id)
        .filter(
            Employment.employment_open_time.isnot(None),
            Employment.offer_time.isnot(None),
        )
        .group_by(Student.class_id)
        .all()
    )