167 lines
5.8 KiB
Python
167 lines
5.8 KiB
Python
|
|
from typing import Optional
|
|
from sqlalchemy import and_, func, desc
|
|
from model.classes import Class
|
|
from model.student import Student
|
|
from model.student_score import Score
|
|
from model.work import Work
|
|
|
|
|
|
class Statistic:
|
|
def __init__(self,db):
|
|
self.db = db
|
|
|
|
def get_ages(self,age1:int,age2:Optional[int],condition:int):
|
|
|
|
stu = self.db.query(Student)
|
|
if not age2:
|
|
if condition == 1:
|
|
q=stu.filter(Student.age > age1).all()
|
|
|
|
elif condition == 2:
|
|
q=stu.filter(Student.age < age1).all()
|
|
|
|
elif condition == 3:
|
|
q=stu.filter(Student.age == age1).all()
|
|
|
|
else:
|
|
return f"输入错误，重新输入"
|
|
else:
|
|
if condition == 4:
|
|
q=stu.filter(and_(Student.age > age1, Student.age < age2)).all()
|
|
|
|
else:
|
|
return f"输入错误，重新输入"
|
|
return q
|
|
|
|
def get_class(self):
|
|
# 查询所有存在的班级cid（去重）
|
|
cid_list = self.db.query(Student.cid).distinct().all()
|
|
result = []
|
|
for row in cid_list:
|
|
cid = row.cid
|
|
# 当前班级全部学生
|
|
all_stu = self.db.query(Student).filter(Student.cid == cid).all()
|
|
total = len(all_stu)
|
|
if total == 0:
|
|
continue
|
|
# 男生
|
|
man_list = [s for s in all_stu if s.gender == "男"]
|
|
man_count = len(man_list)
|
|
# 女生
|
|
woman_list = [s for s in all_stu if s.gender == "女"]
|
|
woman_count = len(woman_list)
|
|
|
|
man_rate = man_count / total
|
|
woman_rate = woman_count / total
|
|
|
|
result.append({
|
|
"class_id": cid,
|
|
"total": total,
|
|
"male_count": man_count,
|
|
"male_rate": man_rate,
|
|
"female_count": woman_count,
|
|
"female_rate": woman_rate
|
|
})
|
|
return result
|
|
|
|
def get_grades(self, grades):
|
|
stu = self.db.query(Score).filter(Score.score > grades).distinct().all()
|
|
result = []
|
|
for st in stu:
|
|
result.append({
|
|
"sid": st.sid,
|
|
"score": st.score
|
|
})
|
|
|
|
return result
|
|
|
|
def get_bad_grades(self, number: int):
|
|
# 第一步：统计每个学生不及格(<60)的记录数量，筛选出不及格次数 >= number 的学生sid
|
|
sub_q = self.db.query(
|
|
Score.sid,
|
|
func.count(Score.score).label("bad_count")
|
|
).filter(Score.score < 60) \
|
|
.group_by(Score.sid) \
|
|
.having(func.count(Score.score) >= number) \
|
|
.subquery()
|
|
|
|
# 第二步：关联学生表+子查询，拿到满足条件学生的全部不及格成绩明细
|
|
bad_list = self.db.query(
|
|
Student.name,
|
|
Student.cid,
|
|
Score.score,
|
|
Score.exam_num
|
|
).join(sub_q, Student.sid == sub_q.c.sid) \
|
|
.join(Score, Student.sid == Score.sid) \
|
|
.filter(Score.score < 60) \
|
|
.all()
|
|
|
|
# 组装结果：按学生分组，一个学生一条，里面放成绩明细
|
|
result = {}
|
|
for name, cid, score, exam_num in bad_list:
|
|
key = (name, cid)
|
|
if key not in result:
|
|
result[key] = {
|
|
"name": name,
|
|
"cid": cid,
|
|
"bad_detail": [],
|
|
"bad_count": 0
|
|
}
|
|
result[key]["bad_detail"].append({"exam_num": exam_num, "score": score})
|
|
result[key]["bad_count"] += 1
|
|
return list(result.values())
|
|
|
|
def get_avg_grades(self, sort: int):
|
|
q = self.db.query(
|
|
Class.cid,
|
|
func.avg(Score.score).label("avg_score")
|
|
).join(Student,
|
|
Student.cid == Class.cid) \
|
|
.join(Score, Score.sid == Student.sid) \
|
|
.group_by(Class.cid)
|
|
|
|
data = q.all()
|
|
result = []
|
|
for item in data:
|
|
result.append({
|
|
"班级": item.cid,
|
|
"平均分": float(item.avg_score)
|
|
})
|
|
|
|
if sort == 1:
|
|
result.sort(key=lambda x: x["平均分"], reverse=True)
|
|
elif sort == 2:
|
|
result.sort(key=lambda x: x["平均分"], reverse=False)
|
|
else:
|
|
return {"code": 400, "msg": "输入错误，sort只能填1或2"}
|
|
return result
|
|
# #
|
|
def get_top(self,top):
|
|
stu=self.db.query(Student.name.label("student_name"),
|
|
Student.cid.label("class_id"),
|
|
Work.offer_time,
|
|
Work.company_name,
|
|
Work.salary).join(Student,Student.sid==Work.sid)\
|
|
.order_by(desc(Work.salary)).offset(top - 1).limit(1).first()
|
|
if not stu:
|
|
return f"错误排名"
|
|
name,cid,time,cname,salary = stu
|
|
return {"名字":name,"班级":cid,"就业时间":time,"公司名":cname,"薪水":salary}
|
|
|
|
def get_time(self):
|
|
result = []
|
|
stu=self.db.query(Work.sid,
|
|
Work.open_time,
|
|
Work.offer_time).all()
|
|
for item in stu:
|
|
time=item.offer_time-item.open_time
|
|
result.append({"学生id":item.sid,"就业时长":time})
|
|
return result
|
|
# def get_avg_time(self):
|
|
# result = []
|
|
# stu=Query(Class.id.label("class_id"),
|
|
# func.datediff(Student_employment.offer_time, Student_employment.open_time).join(Student_employment,Student_employment.sid==Student.sid)\
|
|
# .join(Student,Student.cid==Class.cid).group_by(Class.id).fillter().all()
|
|
|