213 lines
11 KiB
Python
213 lines
11 KiB
Python
from model.statistics_model import StudentInfo, StudentScore, ClassInfo, EmploymentInfo
|
||
from datetime import date
|
||
from dateutil.relativedelta import relativedelta
|
||
from sqlalchemy import func # SQL 内置函数生成器（MIN/MAX/COUNT 等）
|
||
|
||
class StatisticsDao:
|
||
@staticmethod
|
||
def get_students_by_age_range_dao(n:int,m:int,min_age: int, max_age: int,db):
|
||
# 年龄 → 出生日期区间（数据库存 birthday，没有 age 字段）
|
||
today = date.today()
|
||
earliest_birthday = today - relativedelta(years=max_age) # 最大年龄 → 最早出生
|
||
latest_birthday = today - relativedelta(years=min_age) # 最小年龄 → 最晚出生
|
||
try:
|
||
q = db.query(StudentInfo).\
|
||
filter(StudentInfo.is_deleted == '0').\
|
||
filter(StudentInfo.birthday >= earliest_birthday).\
|
||
filter(StudentInfo.birthday <= latest_birthday)
|
||
total = q.count() # 总条数
|
||
req = q.offset((n - 1) * m).limit(m).all() # 分页
|
||
except Exception as e:
|
||
db.rollback()
|
||
raise e
|
||
else:
|
||
return req,total
|
||
|
||
@staticmethod
|
||
def get_students_by_class_id_dao(class_id:str,db):
|
||
try:
|
||
q = db.query(StudentInfo).filter(StudentInfo.class_id == class_id).\
|
||
filter(StudentInfo.is_deleted == '0')
|
||
total_count = q.count() # 总人数
|
||
male_count = q.filter(StudentInfo.gender == '男').count()
|
||
female_count = q.filter(StudentInfo.gender == '女').count()
|
||
except Exception as e:
|
||
db.rollback()
|
||
raise e
|
||
else:
|
||
return total_count, male_count , female_count
|
||
|
||
@staticmethod
|
||
def get_students_by_score_dao(n:int,m:int,score:float,db):
|
||
# GROUP BY 每个学生一行，HAVING MIN(score) > 阈值 → 所有成绩都过线
|
||
# SELECT MIN(score) AS score → 返回该学生最低分（schema StudentInfo2 对应字段）
|
||
try:
|
||
q = (db.query(
|
||
StudentInfo.student_name,
|
||
StudentInfo.student_id,
|
||
func.min(StudentScore.score).label('score') # 起别名，后面 i.score 访问
|
||
).join(StudentScore, StudentInfo.student_id == StudentScore.student_id)
|
||
.filter(StudentInfo.is_deleted == '0') # 逻辑删除：学生表
|
||
.filter(StudentScore.is_deleted == '0') # 逻辑删除：成绩表
|
||
.group_by(StudentInfo.student_id, StudentInfo.student_name) # only_full_group_by 要求
|
||
.having(func.min(StudentScore.score) > score)) # 组最低分 > 阈值 = 全过线
|
||
total = q.count()
|
||
req = q.offset((n - 1) * m).limit(m).all()
|
||
except Exception as e:
|
||
db.rollback()
|
||
raise e
|
||
else:
|
||
# Row 对象手动转字典，字段对齐 schema StudentInfo2
|
||
return [{"student_name":i.student_name,
|
||
'student_id':i.student_id,
|
||
"score":i.score} for i in req],total
|
||
|
||
@staticmethod
|
||
def get_student_by_no_pass_dao(n:int,m:int,fail_count:int,db):
|
||
try:
|
||
q = (db.query(
|
||
StudentInfo.student_id,
|
||
StudentInfo.student_name,
|
||
func.count(StudentScore.score_id).label('fail_count')
|
||
).join(StudentScore, StudentInfo.student_id == StudentScore.student_id)
|
||
.filter(StudentInfo.is_deleted == '0')
|
||
.filter(StudentScore.is_deleted == '0')
|
||
.filter(StudentScore.is_pass == '0')
|
||
.group_by(StudentInfo.student_id, StudentInfo.student_name)
|
||
.having(func.count(StudentScore.score_id) >= fail_count))
|
||
total = q.count()
|
||
req = q.offset((n - 1) * m).limit(m).all()
|
||
except Exception as e:
|
||
db.rollback()
|
||
raise e
|
||
else:
|
||
return [{"student_id": i.student_id,
|
||
"student_name": i.student_name,
|
||
"fail_count": i.fail_count} for i in req], total
|
||
|
||
@staticmethod
|
||
def get_class_exam_avg_score_dao(n:int,m:int,db):
|
||
# 统计每次考试每个班级的平均分，按平均分从高到低排序
|
||
# 三表联查：student_score →(student_id)→ student_info →(class_id)→ class_info
|
||
# 分组维度：exam_type + exam_date + course_id + class_id（一场考试×一个班级）
|
||
# 聚合：AVG(score) 排序：DESC
|
||
try:
|
||
q = (db.query(
|
||
StudentScore.course_id, # 课程编号（不同科目）
|
||
StudentInfo.class_id, # 班级编号，来自学生表
|
||
ClassInfo.class_name, # 班级名称，来自班级表
|
||
func.avg(func.ifnull(StudentScore.score, 0)).label('avg_score') # 每个学生缺考分按0算，再求班级平均分
|
||
).join(StudentInfo, StudentScore.student_id == StudentInfo.student_id) # 成绩→学生：取 class_id
|
||
.join(ClassInfo, StudentInfo.class_id == ClassInfo.class_id) # 学生→班级：取 class_name
|
||
.filter(StudentInfo.is_deleted == '0') # 过滤已删除学生
|
||
.filter(StudentScore.is_deleted == '0') # 过滤已删除成绩
|
||
.filter(ClassInfo.is_deleted == '0') # 过滤已停用班级
|
||
.group_by(StudentScore.course_id,
|
||
StudentInfo.class_id, # 每个班级单独成组
|
||
ClassInfo.class_name) # MySQL only_full_group_by 要求
|
||
.order_by(func.avg(func.ifnull(StudentScore.score,0)).desc())) # 平均分降序（从高到低）
|
||
total = q.count() # 分组后的总组合数
|
||
req = q.offset((n - 1) * m).limit(m).all() # 分页：从 (n-1)*m 开始取 m 条
|
||
except Exception as e:
|
||
db.rollback()
|
||
raise e
|
||
else:
|
||
# Row 对象转 dict，avg_score 保留两位小数对齐前端展示
|
||
return [{"course_id": i.course_id,
|
||
"class_id": i.class_id,
|
||
"class_name": i.class_name,
|
||
"avg_score": round(float(i.avg_score), 2)} for i in req], total
|
||
|
||
@staticmethod
|
||
def get_salary_top_dao(m:int,db): # 统计就业薪资最高的前五名学生的姓名，班级和就业时间，就业公司
|
||
try:
|
||
# 子查询：获取每个学生的最高薪资
|
||
sq = (db.query(
|
||
EmploymentInfo.student_id,
|
||
func.max(EmploymentInfo.salary).label('max_salary')
|
||
).filter(EmploymentInfo.is_deleted == '0')
|
||
.group_by(EmploymentInfo.student_id)
|
||
.order_by(func.max(EmploymentInfo.salary).desc())
|
||
.limit(m)
|
||
.subquery())
|
||
|
||
# 主查询：关联学生表、班级表和就业信息表
|
||
q = (db.query(
|
||
StudentInfo.student_name,
|
||
ClassInfo.class_name,
|
||
EmploymentInfo.part_time,
|
||
EmploymentInfo.company_name,
|
||
EmploymentInfo.salary
|
||
).join(EmploymentInfo, StudentInfo.student_id == EmploymentInfo.student_id)
|
||
.join(ClassInfo, StudentInfo.class_id == ClassInfo.class_id)
|
||
.join(sq,
|
||
(EmploymentInfo.student_id == sq.c.student_id) &
|
||
(EmploymentInfo.salary == sq.c.max_salary))
|
||
.filter(StudentInfo.is_deleted == '0')
|
||
.filter(ClassInfo.is_deleted == '0')
|
||
.filter(EmploymentInfo.is_deleted == '0'))
|
||
|
||
req = q.all()
|
||
except Exception as e:
|
||
db.rollback()
|
||
raise e
|
||
else:
|
||
return [{"student_name": i.student_name,
|
||
"class_name": i.class_name,
|
||
"part_time": i.part_time,
|
||
"company_name": i.company_name,
|
||
"salary": i.salary} for i in req]
|
||
|
||
@staticmethod
|
||
def get_time_size_dao(n:int,m:int,db): # 统计每个学生的就业时长(offer下发时间-就业开放时间)
|
||
try:
|
||
# 查询学生姓名、就业信息，并计算就业时长（天数）
|
||
q = (db.query(
|
||
StudentInfo.student_id,
|
||
StudentInfo.student_name,
|
||
EmploymentInfo.offer_date,
|
||
EmploymentInfo.resume_open_date,
|
||
func.datediff(EmploymentInfo.offer_date, EmploymentInfo.resume_open_date).label('time_size')
|
||
).join(EmploymentInfo, StudentInfo.student_id == EmploymentInfo.student_id)
|
||
.filter(StudentInfo.is_deleted == '0')
|
||
.filter(EmploymentInfo.is_deleted == '0')
|
||
.filter(EmploymentInfo.resume_open_date.isnot(None))) # 确保开放简历时间不为空
|
||
|
||
total = q.count()
|
||
req = q.offset((n - 1) * m).limit(m).all()
|
||
except Exception as e:
|
||
db.rollback()
|
||
raise e
|
||
else:
|
||
return [{"student_id": i.student_id,
|
||
"student_name": i.student_name,
|
||
"offer_date": i.offer_date,
|
||
"resume_open_date": i.resume_open_date,
|
||
"time_size": i.time_size} for i in req], total
|
||
|
||
@staticmethod
|
||
def get_class_avg_time_size_dao(n:int,m:int,db):
|
||
# 统计每个班级的平均就业时长(只统计进入就业阶段的学生)
|
||
try:
|
||
# 查询班级名称和平均就业时长
|
||
q = (db.query(
|
||
ClassInfo.class_id,
|
||
ClassInfo.class_name,
|
||
func.avg(func.datediff(EmploymentInfo.offer_date, EmploymentInfo.resume_open_date)).label('avg_time_size')
|
||
).join(StudentInfo, ClassInfo.class_id == StudentInfo.class_id)
|
||
.join(EmploymentInfo, StudentInfo.student_id == EmploymentInfo.student_id)
|
||
.filter(ClassInfo.is_deleted == '0')
|
||
.filter(StudentInfo.is_deleted == '0')
|
||
.filter(EmploymentInfo.is_deleted == '0')
|
||
.filter(EmploymentInfo.resume_open_date.isnot(None)) # 只统计有就业开放时间的学生
|
||
.group_by(ClassInfo.class_id, ClassInfo.class_name))
|
||
|
||
total = q.count()
|
||
req = q.offset((n - 1) * m).limit(m).all()
|
||
except Exception as e:
|
||
db.rollback()
|
||
raise e
|
||
else:
|
||
return [{"class_id": i.class_id,
|
||
"class_name": i.class_name,
|
||
"avg_time_size": round(float(i.avg_time_size), 2)} for i in req], total |