312 lines
12 KiB
Python
312 lines
12 KiB
Python
# 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]:
|
|
"""
|
|
# 查询每次考试成绩都在输入分数线（如80分）以上的学生的编号、姓名和成绩 -- 参数: 分数
|
|
"""
|
|
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_id, [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]:
|
|
"""
|
|
# 2、查询有输入指定次数（如两次）以上不及格的学生的姓名、班级和不及格成绩明细 -- 参数: 不及格次数
|
|
"""
|
|
test_max_cnt = len(db.query(StuScore.exam_attempt)
|
|
.filter(StuScore.is_deleted == 0)
|
|
.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]:
|
|
"""
|
|
# 统计每次考试每个班级的平均分，并支持按分数从高到低或从低到高动态排序 -- 参数: 1:低到高, 2:高到低
|
|
"""
|
|
avg_score = func.avg(StuScore.exam_score)
|
|
order_score = asc(avg_score) if order_choose == 1 else desc(avg_score)
|
|
finally_data = (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 finally_data:
|
|
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]:
|
|
"""
|
|
# 1、统计就业薪资排名 Top N（动态输入 N）的学生的姓名、班级、就业时间和就业公司。 -- 参数: top_N
|
|
"""
|
|
finally_data = (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 finally_data
|
|
]
|
|
|
|
return ans
|
|
|
|
@staticmethod
|
|
def find_students_employment_duration(db: Session) -> List[StudentEmpDurationItem]:
|
|
"""
|
|
# 2、统计每个学生的就业时长（计算公式：offer下发时间 - 就业开放时间） -- 参数: 无
|
|
"""
|
|
time = func.datediff(StudentEmployManage.send_offer_time, StudentEmployManage.emp_open_time)
|
|
|
|
finally_data = (db.query(
|
|
StuInfo.id,
|
|
StuInfo.cls_id,
|
|
StuInfo.name,
|
|
time.label("time")
|
|
).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))
|
|
.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 finally_data
|
|
]
|
|
|
|
return ans
|
|
|
|
@staticmethod
|
|
def find_class_avg_employment_duration(db: Session) -> List[ClassAvgEmpDurationItem]:
|
|
"""
|
|
# 3、统计每个班级的平均就业时长（仅统计进入就业阶段，即有就业开放时间的学生） -- 参数: 无<br>
|
|
"""
|
|
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
|