第二版_完整
This commit is contained in:
jianqi
@@ -0,0 +1,123 @@
|
||||
# dao/cls_mgmt_dao.py
|
||||
# 本文件封装对 ClsMgmt 表的所有数据库操作（增、删、改、查）
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from model.cls_mgmt_model import ClsMgmt
|
||||
from scheme.cls_mgmt_scheme import ClsMgmtResponse, ClsMgmtCreate
|
||||
from scheme.users import UserCreate, UserUpdate
|
||||
from typing import Optional, List
|
||||
|
||||
class ClsMgmtDAO:
|
||||
"""用户数据访问对象，所有方法均为静态方法，方便调用"""
|
||||
|
||||
@staticmethod
|
||||
def get_all(db: Session, skip: int = 0, limit: int = 100):
|
||||
"""
|
||||
获取所有班级（支持分页）
|
||||
:param db: 数据库会话
|
||||
:param skip: 偏移量（跳过前 skip 条）
|
||||
:param limit: 最大返回条数
|
||||
:return: 用户对象列表
|
||||
"""
|
||||
return db.query(ClsMgmt).offset(skip).limit(limit).all()
|
||||
|
||||
# -> Optional[ClsMgmt]
|
||||
# @staticmethod
|
||||
# def get_by_id(db: Session, class_id: str) :
|
||||
# """
|
||||
# 根据主键 ID 获取单个班级
|
||||
# :return: 用户对象或 None
|
||||
# """
|
||||
# return db.query(ClsMgmt).filter(ClsMgmt.id == class_id).first()
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(db: Session, class_id: str) -> Optional[ClsMgmt]:
|
||||
"""
|
||||
根据主键 ID 获取单个班级
|
||||
:return: 班级对象或 None
|
||||
"""
|
||||
return db.query(ClsMgmt).filter(ClsMgmt.id == class_id).first()
|
||||
|
||||
|
||||
|
||||
@staticmethod
|
||||
def generate_class_id(cls_start_date) -> str:
|
||||
"""
|
||||
根据开课时间生成班级id
|
||||
规则：取年后两位 + 月 + 日，共6位数字
|
||||
例如：date(2026, 8, 14) → "260814"
|
||||
"""
|
||||
# strftime("%Y%m%d") → "20260814"，取后6位 → "260814"
|
||||
return cls_start_date.strftime("%Y%m%d")[-6:]
|
||||
|
||||
@staticmethod
|
||||
def exists(db: Session, class_id: str) -> bool:
|
||||
"""
|
||||
根据班级id检查是否已存在
|
||||
"""
|
||||
existing = db.query(ClsMgmt).filter(
|
||||
ClsMgmt.id == class_id,
|
||||
ClsMgmt.is_deleted == 0
|
||||
).first()
|
||||
return existing is not None
|
||||
|
||||
@staticmethod
|
||||
def create(db: Session, cls_data: "ClsMgmtCreate") -> "ClsMgmt":
|
||||
"""
|
||||
新增班级
|
||||
id 根据开课时间自动生成，不再自增
|
||||
"""
|
||||
class_id = ClsMgmtDAO.generate_class_id(cls_data.cls_start_date)
|
||||
new_cls = ClsMgmt(
|
||||
id=class_id, # 手动赋值班级id
|
||||
cls_start_date=cls_data.cls_start_date,
|
||||
head_tea_id=cls_data.head_tea_id,
|
||||
lecturer_id=cls_data.lecturer_id,
|
||||
is_deleted=0,
|
||||
# 如果 created_at / updated_at 没有默认值，取消注释：
|
||||
# created_at=datetime.now(),
|
||||
# updated_at=datetime.now(),
|
||||
)
|
||||
db.add(new_cls)
|
||||
db.commit()
|
||||
db.refresh(new_cls)
|
||||
return new_cls
|
||||
|
||||
@staticmethod
|
||||
def delete_class(db: Session, class_id: str):
|
||||
"""
|
||||
删除班级（软删除）
|
||||
不真正删除数据，而是将 is_deleted 标记为 1
|
||||
"""
|
||||
del_cls = db.query(ClsMgmt).filter(
|
||||
ClsMgmt.id == class_id,
|
||||
ClsMgmt.is_deleted == 0 # 只查未删除的
|
||||
).first()
|
||||
|
||||
if del_cls:
|
||||
del_cls.is_deleted = 1 # ← 关键：软删除，标记为已删除
|
||||
db.commit()
|
||||
return f"班级{class_id}删除成功"
|
||||
return f"班级{class_id}不存在"
|
||||
|
||||
@staticmethod
|
||||
def update_cls(db: Session, cls_data, class_id: str):
|
||||
"""
|
||||
修改班级班主任和授课老师信息
|
||||
"""
|
||||
update_cls = db.query(ClsMgmt).filter(
|
||||
ClsMgmt.id == class_id,
|
||||
ClsMgmt.is_deleted == 0 # 只查未删除的
|
||||
).first()
|
||||
|
||||
if update_cls:
|
||||
update_cls.head_tea_id=cls_data.head_tea_id
|
||||
update_cls.lecturer_id=cls_data.lecturer_id
|
||||
db.commit()
|
||||
return update_cls
|
||||
return f"班级{class_id}不存在"
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,276 @@
|
||||
# dao/statistics_dao.py
|
||||
# 本文件封装对 所有相关表的 所有数据库操作（统计查询）
|
||||
from collections import defaultdict
|
||||
|
||||
from dns.entropy import between
|
||||
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 model.users import User
|
||||
from scheme.users import UserCreate, UserUpdate
|
||||
from typing import Optional, List, Dict
|
||||
from scheme.statistics_scheme import StatsResponse, ExamScoreItem, StuInfoAllExamAboveScore, FailingStudentItem, \
|
||||
ClassExamAvgScoreItem, TopSalaryStudentItem, StudentEmpDurationItem, ClassAvgEmpDurationItem, StudentAgeDetailItem
|
||||
|
||||
|
||||
class StatisticsDao:
|
||||
"""用户数据访问对象，所有方法均为静态方法，方便调用"""
|
||||
|
||||
@staticmethod
|
||||
def get_cls_stats(db: Session) -> List[StatsResponse]:
|
||||
|
||||
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 [
|
||||
StatsResponse(
|
||||
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]:
|
||||
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_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]:
|
||||
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]:
|
||||
avg_score = func.avg(StuScore.exam_score)
|
||||
order_score = asc(avg_score) if order_choose == 1 else desc(avg_score)
|
||||
aaa = (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 aaa:
|
||||
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]:
|
||||
aaa = (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 aaa
|
||||
]
|
||||
|
||||
return ans
|
||||
|
||||
@staticmethod
|
||||
def find_students_employment_duration(db: Session) -> List[StudentEmpDurationItem]:
|
||||
time = func.datediff(StudentEmployManage.send_offer_time, StudentEmployManage.emp_open_time)
|
||||
aaa = (db.query(
|
||||
StuInfo.id,
|
||||
StuInfo.cls_id,
|
||||
StuInfo.name,
|
||||
time.label("time")
|
||||
).join(StudentEmployManage, StuInfo.id == StudentEmployManage.stu_id)
|
||||
.filter(StuInfo.is_deleted == 0,
|
||||
StuInfo.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 aaa
|
||||
]
|
||||
|
||||
return ans
|
||||
|
||||
@staticmethod
|
||||
def find_class_avg_employment_duration(db: Session) -> List[ClassAvgEmpDurationItem]:
|
||||
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
|
||||
Reference in New Issue
Block a user