max_code_test_gitea_2/dao/statistics_dao.py

# dao/statistics_dao.py
# 本文件封装对 所有相关表的 所有数据库操作（统计查询）
from collections import defaultdict

from fastapi.openapi.utils import status_code_ranges
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session, joinedload
from sqlalchemy import func, case, and_, join, asc, desc

from model import StudentEmployManage
from model.stu_model import StuInfo
from model.stu_score_model import StuScore
from typing import Optional, List, Dict
from scheme.statistics_scheme import ClassGenderDistributionItem, ExamScoreItem, StuInfoAllExamAboveScore, FailingStudentItem, \
    ClassExamAvgScoreItem, TopSalaryStudentItem, StudentEmpDurationItem, ClassAvgEmpDurationItem, StudentAgeDetailItem


class StatisticsDao:
    """用户数据访问对象，所有方法均为静态方法，方便调用"""

    @staticmethod
    def get_class_gender_distribution(db: Session) -> List[ClassGenderDistributionItem]:

        all_cls_gender_stats = (
            db.query(
                StuInfo.cls_id.label("cls_id"),
                func.count(StuInfo.id).label("total_cnt"),
                func.sum(case((StuInfo.gender == '男', 1), else_=0)).label("man_cnt"),
                func.sum(case((StuInfo.gender == '女', 1), else_=0)).label("female_cnt")
            )
            .filter(StuInfo.is_deleted == 0)
            .group_by(StuInfo.cls_id)
        )

        return [
            ClassGenderDistributionItem(
                cls_id=i.cls_id,
                total_cnt=i.total_cnt,
                man_cnt=i.man_cnt,
                female_cnt=i.female_cnt,
            ) for i in all_cls_gender_stats
        ]

    # 查询每次考试成绩都在输入分数线（如80分）以上的学生的编号、姓名和成绩。
    @staticmethod
    def get_stu_info_all_exams_above_score(db: Session, score: float) -> List[StuInfoAllExamAboveScore]:
        stu_id_list_query = (
            db.query(StuScore.stu_id)
            .filter(StuScore.is_deleted == 0)
            .group_by(StuScore.stu_id)
            .having(func.min(StuScore.exam_score) >= score)
            .all()
        )
        stu_id_list = [stu[0] for stu in stu_id_list_query]

        records = (db.query(
            StuScore.stu_id,
            StuScore.exam_attempt,
            StuScore.exam_score,
            StuInfo.name,
        )
                   .join(StuInfo, StuScore.stu_id == StuInfo.id)
                   .filter(
            StuScore.is_deleted == 0,
            StuInfo.is_deleted == 0,
            StuScore.stu_id.in_(stu_id_list)
        )
                   .order_by(StuScore.stu_id, StuScore.exam_attempt)
                   .all()
                   )

        stats_dict = defaultdict(lambda: {"name": "", "scores": []})
        for row in records:
            stats_dict[row.stu_id]["name"] = row.name
            stats_dict[row.stu_id]["scores"].append(
                ExamScoreItem(exam_attempt=row.exam_attempt, exam_score=float(row.exam_score))
            )

        return [
            StuInfoAllExamAboveScore(
                stu_id=stu_id,
                stu_name=name_score["name"],
                stu_score=name_score["scores"]
            )
            for stu_id, name_score in stats_dict.items()
        ]

    @staticmethod
    def get_stu_dict_who_fail(db: Session) -> Dict[str, List[StuScore]]:
        stu_fail_dict = defaultdict(list)
        data = (db.query(StuScore)
                .options(joinedload(StuScore.student))  # 预加载模式： 让 SQLAlchemy 在第一次查成绩表时，通过 JOIN 语句一次性把学生信息查出来。
                .filter(and_(StuScore.is_deleted == 0, StuScore.exam_score <= 60))
                .all())
        for row in data:
            stu_fail_dict[row.stu_id].append(row)
        return stu_fail_dict

    @staticmethod
    def find_students_fail_count(db: Session, fail_cnt: int) -> List[FailingStudentItem]:
        test_max_cnt = len(db.query(StuScore.exam_attempt).group_by(StuScore.exam_attempt).all())
        if fail_cnt > test_max_cnt:
            raise HTTPException(status_code = 400, detail = f'最大次数为{test_max_cnt}')

        stu_fail_dict = StatisticsDao.get_stu_dict_who_fail(db)
        ans = []

        for stu_id, scores_info_list in stu_fail_dict.items():
            if len(scores_info_list) < fail_cnt:
                continue

            stu_info = scores_info_list[0].student
            ans.append(
                FailingStudentItem(
                    stu_name=stu_info.name,
                    stu_cls_id=stu_info.cls_id,
                    stu_fail_cnt=len(scores_info_list),
                    stu_score=[
                        ExamScoreItem(
                            exam_attempt=score_info.exam_attempt,
                            exam_score=score_info.exam_score
                        )

                        for score_info in scores_info_list
                    ]
                )
            )

        return ans

    # 统计每次考试每个班级的平均分，并支持按分数从高到低或从低到高动态排序。
    @staticmethod
    def find_class_exam_averages(db: Session, order_choose: int) -> List[ClassExamAvgScoreItem]:
        avg_score = func.avg(StuScore.exam_score)
        order_score = asc(avg_score) if order_choose == 1 else desc(avg_score)
        aaa = (db.query(
            StuInfo.cls_id,
            StuScore.exam_attempt,
            func.avg(StuScore.exam_score).label("avg_score"),
            func.count(StuScore.stu_id).label("stu_cnt")
        )
               .join(StuScore, StuScore.stu_id == StuInfo.id)
               .filter(StuScore.is_deleted == 0, StuInfo.is_deleted == 0)
               .group_by(StuInfo.cls_id, StuScore.exam_attempt)
               .order_by(StuScore.exam_attempt, order_score))

        # 班级 - 考试场次 - 平均分
        ans = []
        for row in aaa:
            ans.append(
                ClassExamAvgScoreItem(
                    exam_attempt=row.exam_attempt,
                    cls_id=row.cls_id,
                    avg_score=round(row.avg_score, 2),
                    stu_cnt=row.stu_cnt,
                )
            )

        return ans

    @staticmethod
    def find_top_salary_students(db: Session, top_n: int) -> List[TopSalaryStudentItem]:
        aaa = (db.query(
            StudentEmployManage.send_offer_time,
            StudentEmployManage.emp_company,
            StudentEmployManage.salary,
            StuInfo.name,
            StuInfo.cls_id
        ).join(StuInfo, StuInfo.id == StudentEmployManage.stu_id)
               .filter(StudentEmployManage.is_deleted == 0, StudentEmployManage.is_deleted == 0)
               .order_by(StudentEmployManage.salary.desc())
               .limit(top_n)
               )

        ans = [
            TopSalaryStudentItem(
                stu_name=row.name,
                cls_id=row.cls_id,
                offer_time=row.send_offer_time,
                emp_company_name=row.emp_company,
                salary=row.salary
            )
            for row in aaa
        ]

        return ans

    @staticmethod
    def find_students_employment_duration(db: Session) -> List[StudentEmpDurationItem]:
        time = func.datediff(StudentEmployManage.send_offer_time, StudentEmployManage.emp_open_time)
        aaa = (db.query(
            StuInfo.id,
            StuInfo.cls_id,
            StuInfo.name,
            time.label("time")
        ).join(StudentEmployManage, StuInfo.id == StudentEmployManage.stu_id)
               .filter(StuInfo.is_deleted == 0,
                       StuInfo.is_deleted == 0,
                       StudentEmployManage.send_offer_time.isnot(None),
                       StudentEmployManage.emp_open_time.isnot(None))
               .order_by(time).all()
               )

        ans = [
            StudentEmpDurationItem(
                stu_id=row.id,
                stu_name=row.name,
                cls_id=row.cls_id,
                offer_time=row.time
            )
            for row in aaa
        ]

        return ans

    @staticmethod
    def find_class_avg_employment_duration(db: Session) -> List[ClassAvgEmpDurationItem]:
        duration_time = func.datediff(StudentEmployManage.send_offer_time, StudentEmployManage.emp_open_time)
        class_avg_query = (db.query(
            StuInfo.cls_id,
            func.avg(duration_time).label("avg_duration_days"),
            func.count(StuInfo.id).label("employed_cnt")
        ).join(StudentEmployManage, StuInfo.id == StudentEmployManage.stu_id)
                           .filter(StuInfo.is_deleted == 0,
                                   StudentEmployManage.is_deleted == 0,
                                   StudentEmployManage.send_offer_time.isnot(None),
                                   StudentEmployManage.emp_open_time.isnot(None))
                           .group_by(StuInfo.cls_id)
                           .order_by(func.avg(duration_time))
                           )

        ans = [
            ClassAvgEmpDurationItem(
                cls_id=row.cls_id,
                avg_duration_days=round(row.avg_duration_days, 2),
                employed_cnt=row.employed_cnt
            )
            for row in class_avg_query
        ]

        return ans

    @staticmethod
    def find_students_by_age_condition(db: Session,
                                       operator: str,
                                       age: int,
                                       min_age: int,
                                       max_age: int) -> List[StudentAgeDetailItem]:
        """
        通过年纪限制 符合条件的学生
        """
        base_query = db.query(StuInfo).filter(StuInfo.is_deleted == 0)
        if operator == "between":
            filter_query = base_query.filter(StuInfo.age.between(min_age, max_age))
        elif operator == "ge" or operator == ">=":
            filter_query = base_query.filter(StuInfo.age >= age)
        elif operator == "le" or operator == "<=":
            filter_query = base_query.filter(StuInfo.age <= age)
        elif operator == "gt" or operator == ">":
            filter_query = base_query.filter(StuInfo.age > age)
        elif operator == "lt" or operator == "<":
            filter_query = base_query.filter(StuInfo.age < age)
        elif operator == "eq" or operator == "=":
            filter_query = base_query.filter(StuInfo.age == age)

        finally_data = filter_query.all()

        ans = [
            StudentAgeDetailItem(
                stu_id=row.id,
                stu_name=row.name,
                cls_id=row.cls_id,
                age=row.age,
                gender=row.gender
            )
            for row in finally_data
        ]

        return ans