111 lines
5.1 KiB
Python
111 lines
5.1 KiB
Python
from sqlalchemy import func, case
|
|
from model.students import Student
|
|
from model.Score import Score
|
|
from model.employment import EmploymentBase,EmploymentOffer
|
|
from model.c_lass import Classinfo
|
|
|
|
def students_age(db,op:str,
|
|
age_value: int = None,
|
|
age_min: int = None,
|
|
age_max: int = None):
|
|
|
|
a=db.query(Student).filter(Student.is_deleted == 0)
|
|
if op == ">":
|
|
a=a.filter(Student.age > age_value)
|
|
elif op == "<":
|
|
a=a.filter(Student.age < age_value)
|
|
elif op == "=":
|
|
a=a.filter(Student.age == age_value)
|
|
elif op == ">=":
|
|
a=a.filter(Student.age >= age_value)
|
|
elif op == "<=":
|
|
a=a.filter(Student.age <= age_value)
|
|
elif op == "between":
|
|
if age_min is None or age_max is None:
|
|
raise ValueError("区间查询必须提供 age_min和age_max")
|
|
a=a.filter(Student.age.between(age_min, age_max))
|
|
else:
|
|
raise ValueError("请使用 >, <, =, >=, <=, between")
|
|
return a.all()
|
|
# 多维度班级统计**：统计每个班级的总人数，以及按性别（男、女）细分的人数分布。
|
|
def class_statistics(db):
|
|
a = (db.query(Student.class_id,func.count(Student.stu_id).label("total"),
|
|
func.sum(case((Student.gender == "男", 1), else_=0)).label("male"),
|
|
func.sum(case((Student.gender == "女", 1), else_=0)
|
|
).label("female"))
|
|
.filter(Student.is_deleted==0).group_by(Student.class_id).all())
|
|
list1 = []
|
|
for i in a:
|
|
list1.append({"class_id": i.class_id,"total": i.total,"male": i.male,"female": i.female})
|
|
return list1
|
|
|
|
# - 查询每次考试成绩都在输入分数线（如80分）以上的学生的编号、姓名和成绩。
|
|
def score_above(db,score_input):
|
|
result = []
|
|
for stu in db.query(Student).filter(Student.is_deleted == 0).all():
|
|
if stu.score and all(s.score >= score_input for s in stu.score):
|
|
for s in stu.score:
|
|
result.append((stu.stu_id, stu.stu_name, s.exam_id, s.score))
|
|
return result
|
|
# - 查询有输入指定次数（如两次）以上不及格的学生的姓名、班级和不及格成绩明细。
|
|
def score_fail(db, fail_times, fail_score=60):
|
|
result = []
|
|
for stu in db.query(Student).filter(Student.is_deleted == 0).all():
|
|
fails = [s for s in stu.score if s.score < fail_score]
|
|
if len(fails) >= fail_times:
|
|
for s in fails:
|
|
result.append((stu.stu_name, stu.class_id, s.exam_order, s.score))
|
|
return result
|
|
|
|
# - 统计每次考试每个班级的平均分，并支持按分数从高到低或从低到高动态排序。
|
|
def avg_scores(db, order="desc"):
|
|
avg = func.avg(Score.score).label("avg_score")
|
|
a = (db.query(Score.exam_id,Student.class_id,avg)
|
|
.join(Score.student).filter(Student.is_deleted == 0)
|
|
.group_by(Score.exam_id,Student.class_id))
|
|
a = a.order_by(avg.desc() if order == "desc" else avg.asc())
|
|
return a.all()
|
|
|
|
# - 统计就业薪资排名 Top N（动态输入 N）的学生的姓名、班级、就业时间和就业公司。
|
|
def top_salary(db, n):
|
|
a = (db.query(Student.stu_name,
|
|
Student.class_id,
|
|
EmploymentBase.job_time,
|
|
EmploymentBase.company_name,
|
|
EmploymentBase.salary).join(Student, EmploymentBase.stu_id == Student.stu_id)
|
|
.filter(EmploymentBase.is_deleted == 0, Student.is_deleted == 0)
|
|
.order_by(EmploymentBase.salary.desc()).offset(n-1).limit(1).first())
|
|
if not a:
|
|
return None
|
|
return {"name": a.stu_name,"class_id": a.class_id,"job_time": str(a.job_time) if a.job_time else None,
|
|
"company": a.company_name,"salary": a.salary}
|
|
|
|
# - 统计每个学生的就业时长（计算公式：offer下发时间 - 就业开放时间）。
|
|
def stu_every(db):
|
|
a = (db.query(Student.stu_id,
|
|
Student.stu_name,
|
|
EmploymentBase.employment_open_time,
|
|
EmploymentOffer.offer_time)
|
|
.join(EmploymentBase, EmploymentBase.stu_id == Student.stu_id)
|
|
.join(EmploymentOffer, EmploymentOffer.stu_id == Student.stu_id)
|
|
.filter(Student.is_deleted == 0,EmploymentBase.is_deleted == 0,
|
|
EmploymentOffer.is_deleted == 0).all())
|
|
return [{"stu_id": sid,"name": name,
|
|
"days": (offer_t - open_t).days if open_t and offer_t else None}
|
|
for sid, name, open_t, offer_t in a]
|
|
|
|
# - 统计每个班级的平均就业时长（仅统计进入就业阶段，即有就业开放时间的学生）。
|
|
def class_avg(db):
|
|
days = func.datediff(EmploymentOffer.offer_time,
|
|
EmploymentBase.employment_open_time).label("days")
|
|
a = (db.query(Student.class_id,
|
|
func.avg(days).label("avg_days"),
|
|
func.count().label("count"))
|
|
.join(EmploymentBase, EmploymentBase.stu_id == Student.stu_id)
|
|
.join(EmploymentOffer, EmploymentOffer.stu_id == Student.stu_id)
|
|
.filter(Student.is_deleted == 0,
|
|
EmploymentBase.is_deleted == 0,
|
|
EmploymentOffer.is_deleted == 0,
|
|
EmploymentBase.employment_open_time.isnot(None)).group_by(Student.class_id).all())
|
|
return [{"class_id": i.class_id, "avg_days": round(float(i.avg_days), 2)} for i in a]
|