#统计分析 数据访问层（动态查询 + 聚合） from os.path import join 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): 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): 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): 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): q=(db.query( Student.stu_name, Student.class_id, Emp.offer_time, Emp.company_name ).select_from(Student) \ .join(Student.emp)) \ .filter(Student.is_deleted==0) \ .order_by(Emp.salary.desc())\ .limit(rank) return q.all() @staticmethod def get_emp_time(db:Session): """每个学生的就业时长（天）= offer下发时间 - 就业开放时间 """ emp_total_time = func.coalesce( # 未就业学生左外连接后 Emp 列为 NULL，DATEDIFF 也返回 NULL，统一兜成 0 func.datediff(Emp.offer_time, Emp.emp_open_time), 0, ).label('emp_total_time') q=db.query( Student.stu_name, emp_total_time, ).select_from(Student) \ .outerjoin(Student.emp) \ .filter(Student.is_deleted==0) return q.all() @staticmethod def get_class_avg_emp_time(db:Session, sort_order: SortOrder | None = None): """每个班级的平均就业时长（天）= offer下发时间 - 就业开放时间 口径： - 以班级表为主表左外连接，保证没有任何就业学生的班级也出现在结果里 - 只统计已进入就业阶段（emp_open_time 非空）且已拿到 offer（offer_time 非空）的学生 - 该班可统计的人数为 0 时 AVG 返回 NULL，出参里 avg_emp_time 给 None，即"无就业学生" """ emp_time = func.datediff(Emp.offer_time, Emp.emp_open_time) # AVG 和 COUNT(表达式) 都会自动忽略 NULL，所以没 offer 时间的学生不参与分子也不占分母 avg_emp_time = func.avg(emp_time).label("avg_emp_time") # 用 COUNT(表达式) 而不是 COUNT(*)，它正好是平均值的分母 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, # 逻辑删除的过滤条件必须写在 ON 里，写进 where 会把空班级整行滤掉 and_(Student.class_id == Class_.class_id, Student.is_deleted == 0), ) \ .outerjoin( Emp, and_( Emp.stu_id == Student.stu_id, Emp.emp_open_time.isnot(None), Emp.offer_time.isnot(None), ), ) \ .group_by(Class_.class_id) # 动态排序逻辑与 get_class_avg_score 保持一致 if sort_order == SortOrder.desc: # MySQL 里 DESC 时 NULL 排最后，"无就业学生"的班级自然落到末尾 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 出来是 Decimal，转 float 并按天保留 1 位小数；None 表示该班无就业学生 "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() ]