429 lines
14 KiB
Python
429 lines
14 KiB
Python
|
||
from sqlalchemy import func
|
||
from sqlalchemy.exc import SQLAlchemyError
|
||
from sqlalchemy.orm import Session
|
||
from datetime import date
|
||
|
||
from model import Student, Classes, Score, Employment
|
||
from schema.student_schema import AgeQuery
|
||
|
||
|
||
class BusinessException(Exception):
|
||
"""业务异常类，DAO层捕获数据库异常后抛出，由API层处理"""
|
||
pass
|
||
|
||
|
||
class DuplicateStudentNo(BusinessException):
|
||
"""学号已存在。继承 BusinessException，API 层可以单独挑出来返回 409。"""
|
||
pass
|
||
|
||
#=============把一个学生对象变成字典形式================
|
||
def student_dict(student):
|
||
if student is None:
|
||
return {}
|
||
return {"s_id": student.sid,
|
||
"student_no": student.student_no,
|
||
"student_name":student.student_name,
|
||
"class_id":student.class_id,
|
||
"flag":student.flag,
|
||
"advisor_id":student.advisor_id,
|
||
"native_place":student.native_place,
|
||
"school":student.school,
|
||
"major":student.major,
|
||
"enrollment_time":student.enrollment_time,
|
||
"graduation_time":student.graduation_time,
|
||
"education":student.education,
|
||
"age":student.age,
|
||
"gender":student.gender,
|
||
"state":student.state}
|
||
|
||
|
||
def create_student(db: Session,
|
||
student_no: str,
|
||
student_name: str,
|
||
class_id: int = None,
|
||
flag: int = 1,
|
||
advisor_id: int = None,
|
||
native_place: str = None,
|
||
school: str = None,
|
||
major: str = None,
|
||
enrollment_time: date = None,
|
||
graduation_time: date = None,
|
||
education: str = None,
|
||
age: int = None,
|
||
gender: str = None,
|
||
state: str = None
|
||
):
|
||
"""
|
||
创建学生
|
||
"""
|
||
try:
|
||
# 学号是唯一约束，先查一下，给出比 IntegrityError 更好懂的提示
|
||
existing = db.query(Student).filter(Student.student_no == student_no).first()
|
||
if existing is not None:
|
||
raise DuplicateStudentNo(f"学号 {student_no} 已存在")
|
||
|
||
new_student = Student(
|
||
student_no=student_no,
|
||
student_name=student_name,
|
||
class_id=class_id,
|
||
flag=flag,
|
||
advisor_id=advisor_id,
|
||
native_place=native_place,
|
||
school=school,
|
||
major=major,
|
||
enrollment_time=enrollment_time,
|
||
graduation_time=graduation_time,
|
||
education=education,
|
||
age=age,
|
||
gender=gender,
|
||
state=state,
|
||
)
|
||
db.add(new_student)
|
||
db.commit()
|
||
db.refresh(new_student)
|
||
return new_student
|
||
except SQLAlchemyError as e:
|
||
db.rollback()
|
||
raise BusinessException(f"创建学生失败: {str(e)}")
|
||
|
||
#================查询学生==================
|
||
def search_student(db: Session, student_no: str,
|
||
student_name: str,
|
||
class_id: int = None,
|
||
page: int = 1,
|
||
size:int=5):
|
||
try:
|
||
query = db.query(Student).filter(Student.flag == 1)
|
||
|
||
if student_no is not None:
|
||
query = query.filter(Student.student_no == student_no)
|
||
if student_name:
|
||
query = query.filter(Student.student_name.like(f"%{student_name}%"))
|
||
if class_id is not None:
|
||
query = query.filter(Student.class_id == class_id)
|
||
|
||
total = query.count()
|
||
offset = (page - 1) * size
|
||
students = query.order_by(Student.sid.desc()).offset(offset).limit(size).all()
|
||
|
||
items = [student_dict(s) for s in students]
|
||
pages = (total + size - 1) // size if size > 0 else 0
|
||
|
||
return {
|
||
"items": items,
|
||
"total": total,
|
||
"page": page,
|
||
"page_size": size,
|
||
"pages": pages,
|
||
}
|
||
except SQLAlchemyError as e:
|
||
raise BusinessException(f"查询学生列表失败: {str(e)}")
|
||
#=================ID查询学生详情====================
|
||
|
||
def get_student(db: Session, sid: int):
|
||
"""
|
||
根据ID查询学生，自动过滤逻辑删除
|
||
"""
|
||
try:
|
||
student = db.query(Student).filter(
|
||
Student.sid == sid,
|
||
Student.flag == 1
|
||
).first()
|
||
return student
|
||
except SQLAlchemyError as e:
|
||
raise BusinessException(f"查询学生失败: {str(e)}")
|
||
|
||
|
||
#===============更新学生信息================
|
||
def update_student(db: Session, sid: int,
|
||
student_no: str = None,
|
||
student_name: str = None,
|
||
class_id: int = None,
|
||
flag: int = None,
|
||
advisor_id: int = None,
|
||
native_place: str = None,
|
||
school: str = None,
|
||
major: str = None,
|
||
enrollment_time: date = None,
|
||
graduation_time: date = None,
|
||
education: str = None,
|
||
age: int = None,
|
||
gender: str = None,
|
||
state: str = None):
|
||
"""
|
||
更新学生信息：只改传了值的字段，没传（None）的保持原样。
|
||
"""
|
||
try:
|
||
student = db.query(Student).filter(Student.sid == sid).first()
|
||
if student is None:
|
||
return None
|
||
|
||
# 判断的是"传进来的参数"，是否是None
|
||
if student_no is not None:
|
||
student.student_no = student_no
|
||
if student_name is not None:
|
||
student.student_name = student_name
|
||
if class_id is not None:
|
||
student.class_id = class_id
|
||
if flag is not None:
|
||
student.flag = flag
|
||
if advisor_id is not None:
|
||
student.advisor_id = advisor_id
|
||
if native_place is not None:
|
||
student.native_place = native_place
|
||
if school is not None:
|
||
student.school = school
|
||
if major is not None:
|
||
student.major = major
|
||
if enrollment_time is not None:
|
||
student.enrollment_time = enrollment_time
|
||
if graduation_time is not None:
|
||
student.graduation_time = graduation_time
|
||
if education is not None:
|
||
student.education = education
|
||
if age is not None:
|
||
student.age = age
|
||
if gender is not None:
|
||
student.gender = gender
|
||
if state is not None:
|
||
student.state = state
|
||
|
||
db.commit()
|
||
db.refresh(student)
|
||
return student
|
||
except SQLAlchemyError as e:
|
||
db.rollback()
|
||
raise BusinessException(f"学生更新信息失败: {str(e)}")
|
||
|
||
#==============逻辑删除学生==================
|
||
def delete_student(db: Session, sid: int):
|
||
try:
|
||
student = db.query(Student).filter(Student.sid == sid).first()
|
||
if student is None or student.flag == 0:
|
||
return None
|
||
student.flag = 0
|
||
db.commit() # 进行任务提交
|
||
db.refresh(student)
|
||
return student
|
||
except SQLAlchemyError as e:
|
||
db.rollback()
|
||
raise BusinessException(f"学生删除失败: {str(e)}")
|
||
|
||
|
||
#==============性别查询学生==================
|
||
def get_class_gender(db: Session):
|
||
class_list = db.query(Classes).all()
|
||
res = []
|
||
for cls in class_list:
|
||
cid = cls.cid
|
||
# 当前班级有效学生总数
|
||
class_total = (
|
||
db.query(func.count(Student.sid))
|
||
.filter(Student.class_id == cid, Student.flag == 1)
|
||
.scalar()
|
||
)
|
||
|
||
# 当前班级男生数量
|
||
male_count = (
|
||
db.query(func.count(Student.sid))
|
||
.filter(Student.class_id == cid, Student.flag == 1, Student.gender == "男")
|
||
.scalar()
|
||
)
|
||
|
||
# 当前班级女生数量
|
||
female_count = (
|
||
db.query(func.count(Student.sid))
|
||
.filter(Student.class_id == cid, Student.flag == 1, Student.gender == "女")
|
||
.scalar()
|
||
)
|
||
|
||
res.append({
|
||
"class_name": cls.class_name,
|
||
"class_total": class_total,
|
||
"boys_count": male_count,
|
||
"girls_count": female_count
|
||
})
|
||
return res
|
||
|
||
#==============查询每次考试高于自定义分数线的学生==================
|
||
def get_higher_score(db:Session,score):
|
||
all_students = (db.query(Student)
|
||
.filter(Student.flag==1)
|
||
.all())
|
||
higherStu=[]
|
||
for student in all_students:
|
||
all_scores = []
|
||
for stu_score in student.scores:
|
||
if stu_score.score<score:
|
||
break
|
||
all_scores.append(stu_score.score)
|
||
else:
|
||
if len(all_scores) > 0:
|
||
higherStu.append({
|
||
'stu_id':student.sid,
|
||
'stu_name':student.student_name,
|
||
'stu_score':all_scores
|
||
})
|
||
return higherStu
|
||
|
||
#==============查询不及格次数大于指定次数的学生==================
|
||
def get_fail_more_than(db: Session, fail_times: int):
|
||
# 查询有效学生
|
||
all_students = (db.query(Student)
|
||
.filter(Student.flag == 1)
|
||
.all())
|
||
result = []
|
||
for student in all_students:
|
||
#不及格成绩列表
|
||
fail_score_list = []
|
||
for stu_score in student.scores:
|
||
if stu_score.score < 60:
|
||
fail_score_list.append(stu_score.score)
|
||
if len(fail_score_list) >= fail_times:
|
||
result.append({
|
||
'stu_id': student.sid,
|
||
'stu_name': student.student_name,
|
||
'class_name': student.classes.class_name,
|
||
'fail_scores': fail_score_list
|
||
})
|
||
return result
|
||
|
||
|
||
#==============查询所有学生就业时长==================
|
||
def work_time(db:Session):
|
||
all_times = []
|
||
#所有工作信息
|
||
worksInfo = db.query(Employment).filter(Employment.flag == 1).all()
|
||
#所有学生信息
|
||
all_students = (db.query(Student).filter(Student.flag==1).all())
|
||
for work in worksInfo:
|
||
if work.offer_time is not None and work.employment_start_time is not None and work.employment_start_time >= work.offer_time:
|
||
delta = work.employment_start_time - work.offer_time
|
||
days = delta.days # 整数天数
|
||
for i in all_students:
|
||
if i.sid==work.student_id:
|
||
all_times.append({'name':i.student_name,'id': work.student_id, 'work_time': days})
|
||
return all_times
|
||
|
||
#==============统计平均分,可以切换排序方式==================
|
||
def count_avg(db: Session,order_type: str = "desc"):
|
||
avg_score = func.avg(Score.score).label("avg_score")
|
||
query = (
|
||
db.query(
|
||
Score.score.label("score"),
|
||
Classes.cid.label("class_id"),
|
||
Classes.class_name.label("class_name"),
|
||
avg_score,
|
||
)
|
||
.join(Student, Score.student_id == Student.sid)
|
||
.join(Classes, Student.class_id == Classes.cid)
|
||
.group_by(Score.score, Classes.cid, Classes.class_name)
|
||
)
|
||
|
||
# 动态排序
|
||
if order_type == "asc":
|
||
query = query.order_by(avg_score.asc())
|
||
else:
|
||
query = query.order_by(avg_score.desc())
|
||
rows = query.all()
|
||
return [
|
||
{
|
||
"score": row.score,
|
||
"class_id": row.class_id,
|
||
"class_name": row.class_name,
|
||
"avg_score": float(row.avg_score) if row.avg_score is not None else 0.0,
|
||
}
|
||
for row in rows
|
||
]
|
||
|
||
#==============按年龄区间自定义比较==================
|
||
def get_students_by_age(db: Session, query:AgeQuery):
|
||
q = db.query(Student).filter(Student.flag==1)
|
||
|
||
op = query.operator
|
||
if op == "gt":
|
||
q = q.filter(Student.age > query.value)
|
||
elif op == "lt":
|
||
q = q.filter(Student.age < query.value)
|
||
elif op == "eq":
|
||
q = q.filter(Student.age == query.value)
|
||
elif op == "ge":
|
||
q = q.filter(Student.age >= query.value)
|
||
elif op == "le":
|
||
q = q.filter(Student.age <= query.value)
|
||
elif op == "between":
|
||
q = q.filter(Student.age.between(query.min_value, query.max_value))
|
||
return q.all()
|
||
|
||
#==============按薪资排名==================
|
||
def get_top_n_salary(db: Session, rank: int):
|
||
rows = (
|
||
db.query(
|
||
Student.student_name.label("student_name"),
|
||
Classes.class_name.label("class_name"),
|
||
Employment.offer_time.label("offer_time"),
|
||
Employment.company_name.label("company"),
|
||
Employment.salary.label("salary"),
|
||
)
|
||
.join(Student, Employment.student_id == Student.sid)
|
||
.join(Classes, Student.class_id == Classes.cid)
|
||
.order_by(Employment.salary.desc())
|
||
.limit(rank)
|
||
.all()
|
||
)
|
||
|
||
return [
|
||
{
|
||
"student_name": row.student_name,
|
||
"class_name": row.class_name,
|
||
"offer_time": row.offer_time,
|
||
"company": row.company,
|
||
"salary": float(row.salary) if row.salary is not None else 0.0,
|
||
}
|
||
for row in rows
|
||
]
|
||
|
||
#==============每个班的平均就业时长==================
|
||
def avg_work_time_by_class(db: Session):
|
||
rows = (
|
||
db.query(
|
||
Classes.cid.label("class_id"),
|
||
Classes.class_name.label("class_name"),
|
||
Employment.employment_start_time,
|
||
Employment.offer_time,
|
||
)
|
||
.join(Student, Employment.student_id == Student.sid)
|
||
.join(Classes, Student.class_id == Classes.cid)
|
||
.filter(Employment.employment_start_time.isnot(None))
|
||
.filter(Employment.offer_time.isnot(None))
|
||
.filter(Employment.flag == 1, Student.flag == 1)
|
||
.filter(Employment.employment_start_time >= Employment.offer_time)
|
||
.all()
|
||
)
|
||
|
||
# 普通字典：{class_id: {"class_name": ..., "days_list": [...]}}
|
||
class_data = {}
|
||
|
||
for row in rows:
|
||
delta = row.employment_start_time - row.offer_time
|
||
days = delta.total_seconds() / 86400
|
||
if row.class_id not in class_data:
|
||
class_data[row.class_id] = {
|
||
"class_name": row.class_name,
|
||
"days_list": [],
|
||
}
|
||
class_data[row.class_id]["days_list"].append(days)
|
||
# 组装结果
|
||
result = []
|
||
for class_id, data in class_data.items():
|
||
days_list = data["days_list"]
|
||
avg_days = round(sum(days_list) / len(days_list), 2)
|
||
result.append({
|
||
"class_id": class_id,
|
||
"class_name": data["class_name"],
|
||
"days": avg_days,
|
||
})
|
||
|
||
return result
|