1/dao/wl_statistics_dao.py

#统计分析 数据访问层（动态查询 + 聚合）
from typing import Optional

from sqlalchemy import and_, case, func
from sqlalchemy.orm import Session

from model.wl_class_model import Class_
from model.wl_emp_model import Emp
from model.wl_score_model import Score
from model.wl_student_model import Student
from scheme.wl_statistics_scheme import AgeCompareOp, SortOrder

# stu_gender 字段的实际取值
MALE_VALUE = "男"
FEMALE_VALUE = "女"




# ---- 2.6.1.1 动态年龄范围查询 ----
# 比较条件 -> SQLAlchemy 表达式工厂，动态拼接 where 条件
_AGE_OP_FACTORY = {
    AgeCompareOp.gt: lambda v1, v2: Student.stu_age > v1,
    AgeCompareOp.ge: lambda v1, v2: Student.stu_age >= v1,
    AgeCompareOp.lt: lambda v1, v2: Student.stu_age < v1,
    AgeCompareOp.le: lambda v1, v2: Student.stu_age <= v1,
    AgeCompareOp.eq: lambda v1, v2: Student.stu_age == v1,
    AgeCompareOp.between: lambda v1, v2: Student.stu_age.between(v1, v2),
}

class StatisticDao:

    @staticmethod
    def search_students_by_age(db: Session, op: AgeCompareOp,
                               value: int, value2: Optional[int] = None):
        """
        输入比较条件动态查询符合条件的学员信息
        :param db:数据库会话
        :param op:比较条件（如大于、小于、等于、区间等）
        :param value:对比值1
        :param value2:对比值2，选了between后的上界
        :return:
        """
        if op == AgeCompareOp.between:
            if value2 is None:
                raise ValueError("区间查询（between）需要同时提供上界 value2")
            if value > value2:
                raise ValueError("区间查询的下界不能大于上界")

        condition = _AGE_OP_FACTORY[op](value, value2)
        return db.query(Student) \
                 .filter(Student.is_deleted == 0, condition) \
                 .order_by(Student.stu_age) \
                 .all()

    @staticmethod
    def statistic_classes_count(db: Session):
        """每个班级的总人数 + 按性别细分的分布（含 0 人的空班级）"""
        result=[]
        for c in db.query(Class_).filter(Class_.is_deleted==0).all():
            students=[s for s in c.students if s.is_deleted == 0]# 班级里的学生，过滤逻辑删除后的
            result.append({
                "class_id":c.class_id,
                "total":len(students),
                "male_count":sum(1 for s in students if s.stu_gender==MALE_VALUE),
                "female_count":sum(1 for s in students if s.stu_gender==FEMALE_VALUE)
            })
        return result
        # 另一个版本的写法，效率更高
        # male = func.sum(case((Student.stu_gender == MALE_VALUE, 1), else_=0))
        # female = func.sum(case((Student.stu_gender == FEMALE_VALUE, 1), else_=0))
        #
        # return db.query(
        #     WlClass.class_id.label("class_id"),
        #     func.count(Student.stu_id).label("total"),
        #     male.label("male_count"),
        #     female.label("female_count"),
        # ).select_from(WlClass) \
        #  .outerjoin(
        #      Student,
        #      # 逻辑删除的过滤条件必须写在 ON 里，写进 where 会把空班级整行滤掉
        #      and_(Student.class_id == WlClass.class_id, Student.is_deleted == 0),
        #  ) \
        #  .filter(WlClass.is_deleted == 0) \
        #  .group_by(WlClass.class_id) \
        #  .order_by(WlClass.class_id) \
        #  .all()

    @staticmethod
    def query_score_by_line(db:Session,score_line:float):
        """
        查询每次考试都在分数线以上学生信息
        :param db:
        :param score_line: 分数线
        :return:
        """
        result=[]
        for s in db.query(Student).filter(Student.is_deleted==0).all():
            if all(sc.score>score_line for sc in s.scores):
                result.append({
                    "stu_no":s.stu_no,
                    "stu_name":s.stu_name,
                    "scores": [{"exam_order":i.exam_order,"score":i.score} for i in s.scores]
                })
        return result

    @staticmethod
    def query_by_fail_count(db:Session,fail_count:int):
        """
        查询有指定次数次以上不及格学生的信息
        :param db:
        :param fail_count: 指定次数
        :return:
        """
        result=[]
        students=db.query(Student).filter(Student.is_deleted==0).all()
        for stu in students:
            if len([sc for sc in stu.scores if sc.score<60])>fail_count:
                result.append({
                    "stu_name":stu.stu_name,
                    "class_id":stu.class_id,
                    "scores":[{"exam_order":i.exam_order,"score":i.score} for i in stu.scores]
                })
        return result

    @staticmethod
    def get_class_avg_score(db: Session, sort_order: SortOrder | None = None):
        """每次考试每个班级的平均分，支持按平均分动态升/降序"""
        avg_score = func.avg(Score.score).label("avg_score")

        q = db.query(
            Student.class_id.label("class_id"),
            Score.exam_order.label("exam_order"),
            avg_score,
        ).select_from(Score) \
         .join(Score.student) \
         .filter(Student.is_deleted == 0, Student.class_id.isnot(None)) \
         .group_by(Student.class_id, Score.exam_order)

        # 动态排序：传了方向就按平均分排，没传则按班级 + 考试序次排
        if sort_order == SortOrder.desc:
            q = q.order_by(avg_score.desc())
        elif sort_order == SortOrder.asc:
            q = q.order_by(avg_score.asc())
        else:
            q = q.order_by(Student.class_id, Score.exam_order)

        return q.all()

    @staticmethod
    def get_info_by_rank(db:Session,rank:int):
        """
        统计就业薪资排名TopN的学生信息
        :param db:
        :param rank: Top N 中的N
        :return:
        """
        q=(db.query(
            Student.stu_name,
            Student.class_id,
            Emp.offer_time,
            Emp.company_name,
            Emp.salary
        ).select_from(Student) \
         .join(Student.emp)) \
         .filter(Student.is_deleted==0, Emp.is_deleted==0) \
         .order_by(Emp.salary.desc())\
        .limit(rank)

        return q.all()
    @staticmethod
    def get_emp_time(db:Session):
        """统计每个学生的就业时长（天）= offer下发时间 - 就业开放时间
        未就业（offer_time，和emp_open_time至少一个为null）
        """
        emp_total_time = func.coalesce(
            func.datediff(Emp.offer_time, Emp.emp_open_time),
            '', # 未进入就业阶段或者未拿到offer
        ).label('emp_total_time')

        # 每个学生都要统计，所以用外连接
        q=db.query(
            Student.stu_name,
            emp_total_time,
        ).select_from(Student) \
        .outerjoin(Emp, and_(Student.stu_no == Emp.stu_no, Emp.is_deleted == 0)) \
        .filter(Student.is_deleted==0)

        return q.all()

    @staticmethod
    def get_class_avg_emp_time(db:Session, sort_order: SortOrder | None = None):
        """
        统计每个班级平均就业时长
        :param db:
        :param sort_order:排序（可选）
        :return:
        """
        emp_time = func.datediff(Emp.offer_time, Emp.emp_open_time)
        avg_emp_time = func.avg(emp_time).label("avg_emp_time")
        emp_stu_count = func.count(emp_time).label("emp_stu_count")# 统计有就业时长学生数量

        q = db.query(
            Class_.class_id.label("class_id"),
            avg_emp_time,
            emp_stu_count,
        ).select_from(Class_) \
         .outerjoin(
             Student,
             and_(Student.class_id == Class_.class_id, Student.is_deleted == 0),
         ) \
         .outerjoin(
             Emp,
             and_(
                 Emp.stu_no == Student.stu_no,
                 Emp.is_deleted == 0,
                 Emp.emp_open_time.isnot(None),
                 Emp.offer_time.isnot(None),
             ),
         ) \
         .group_by(Class_.class_id)


        if sort_order == SortOrder.desc:
            q = q.order_by(avg_emp_time.desc())
        elif sort_order == SortOrder.asc:
            q = q.order_by(avg_emp_time.asc())
        else:
            q = q.order_by(Class_.class_id)

        return [
            {
                "class_id": r.class_id,
                "avg_emp_time": round(float(r.avg_emp_time), 1) if r.avg_emp_time is not None else None,
                "emp_stu_count": r.emp_stu_count,
            }
            for r in q.all()
        ]