diff --git a/api/statistics_api.py b/api/statistics_api.py
index 118503f..bdcb4fe 100644
--- a/api/statistics_api.py
+++ b/api/statistics_api.py
@@ -6,32 +6,35 @@ from typing import List, Optional
from database import get_db
from dao.statistics_dao import StatisticsDao
-from scheme.statistics_scheme import ClassGenderDistributionItem, StuInfoAllExamAboveScore, FailingStudentItem, ClassExamAvgScoreItem, \
- TopSalaryStudentItem, StudentEmpDurationItem, ClassAvgEmpDurationItem, StudentAgeDetailItem
+from scheme.statistics_scheme import ClassGenderDistributionItem, \
+TopSalaryStudentItem, StudentEmpDurationItem, ClassAvgEmpDurationItem, StudentAgeDetailItem
router = APIRouter()
#### 2.6.1 基本信息动态统计
# - **动态年龄范围查询**：支持用户输入年龄阈值及比较条件（如大于、小于、等于、区间等），动态查询符合条件的学员信息。
-@router.get("/api/stats/students/age_filter",
+@router.get("/students/age_filter",
response_model=List[StudentAgeDetailItem],
summary="动态年龄范围查询",
description="支持用户输入年龄阈值及比较条件（如大于、小于、等于、区间等），动态查询符合条件的学员信息。")
def get_students_by_age_condition(db: Session = Depends(get_db),
- operator: str = Query(..., description="比较条件: gt/gte/lt/lte/eq/between"),
+ operator: str = Query(...,
+ description="""比较条件: gt/ge/lt/le/eq/between
或者:>/>=/"""),
age: Optional[int] = Query(None, ge=0, description="单值比较时的年龄(gt/lt/eq等用)"),
- min_age: Optional[int] = Query(None, ge=0,description="区间比较时的最小年龄(between用)"),
- max_age: Optional[int] = Query(None, ge=0,description="区间比较时的最大年龄(between用)")
+ min_age: Optional[int] = Query(None, ge=0,
+ description="区间比较时的最小年龄(between用)"),
+ max_age: Optional[int] = Query(None, ge=0,
+ description="区间比较时的最大年龄(between用)")
):
if operator == "between":
if min_age is None or max_age is None:
- raise HTTPException(status_code=400, detail="最大/最小值 都得输入")
+ raise HTTPException(status_code=400, detail="between功能: 最大/最小值 都得输入")
if min_age > max_age:
- raise HTTPException(status_code=400, detail="最小值 要 小于 最大值！")
+ raise HTTPException(status_code=400, detail="between功能: 最小值 要 小于 最大值！")
else:
if age is None:
- raise HTTPException(status_code=400, detail="age 不能不写！")
+ raise HTTPException(status_code=400, detail="非between功能: age 不能不写！")
data = StatisticsDao.find_students_by_age_condition(db = db,
operator = operator,
@@ -41,9 +44,9 @@ def get_students_by_age_condition(db: Session = Depends(get_db),
return data
# **多维度班级统计**：统计每个班级的总人数，以及按性别（男、女）细分的人数分布。
-@router.get("/api/stats/class/gender_distribution",
+@router.get("/class/gender_distribution",
response_model=List[ClassGenderDistributionItem],
- summary="班级性别统计",
+ summary="班级人数及性别统计",
description="统计每个班级的总人数，以及按性别（男、女）细分的人数分布。")
def get_class_stats(db: Session = Depends(get_db)):
data = StatisticsDao.get_class_gender_distribution(db)
@@ -61,21 +64,26 @@ def get_class_stats(db: Session = Depends(get_db)):
def get_students_all_exams_above_score(
db: Session = Depends(get_db),
choose: int = Query(..., ge=1, le=3, description="功能选择(1-3): 参考提示"),
- player_num: int | float = Query(..., ge=0, le=100, description="输入的具体参数,参考提示")
+ param: int | float = Query(..., description="输入的具体参数,参考提示")
):
if choose == 1:
# 查询每次考试成绩都在输入分数线（如80分）以上的学生的编号、姓名和成绩 -- 参数: 分数
- if player_num > 100 or player_num < 0:
+ if param > 100 or param < 0:
raise HTTPException(status_code=400, detail="成绩需要在[0-100]区间")
- data = StatisticsDao.get_stu_info_all_exams_above_score(db, player_num)
+ data = StatisticsDao.get_stu_info_all_exams_above_score(db, param)
+
elif choose == 2:
# 2、查询有输入指定次数（如两次）以上不及格的学生的姓名、班级和不及格成绩明细 -- 参数: 不及格次数
- data = StatisticsDao.find_students_fail_count(db, player_num)
+ data = StatisticsDao.find_students_fail_count(db, param)
+
elif choose == 3:
# 统计每次考试每个班级的平均分，并支持按分数从高到低或从低到高动态排序 -- 参数: 1:低到高, 2:高到低
- data = StatisticsDao.find_class_exam_averages(db, player_num)
+ if param not in [1, 2]:
+ raise HTTPException(status_code=400, detail="排序只有两种: 1:低到高, 2:高到低")
+
+ data = StatisticsDao.find_class_exam_averages(db, param)
return data
@@ -119,9 +127,37 @@ def get_class_exams_avg_scores(
return data
"""
+#### 2.6.3 就业数据统计: all
+@router.get("/employment_time",
+ summary="就业数据统计",
+ description= """
+ 1、统计就业薪资排名 Top N（动态输入 N）的学生的姓名、班级、就业时间和就业公司。 -- 参数: top_N
+ 2、统计每个学生的就业时长（计算公式：offer下发时间 - 就业开放时间） -- 参数: 无
+ 3、统计每个班级的平均就业时长（仅统计进入就业阶段，即有就业开放时间的学生） -- 参数: 无
+ """)
+def get_students_all_exams_above_score(
+ db: Session = Depends(get_db),
+ choose: int = Query(..., ge=1, le=3, description="功能选择(1-3): 参考提示"),
+ player_num: int = Query(10, ge=0, le=100, description="功能1 输入的具体参数,参考提示(其他功能无效)")
+):
+
+ if choose == 1:
+ # 1、统计就业薪资排名 Top N（动态输入 N）的学生的姓名、班级、就业时间和就业公司。 -- 参数: top_N
+ data = StatisticsDao.find_top_salary_students(db, player_num)
+ elif choose == 2:
+ # 2、统计每个学生的就业时长（计算公式：offer下发时间 - 就业开放时间） -- 参数: 无
+ data = StatisticsDao.find_students_employment_duration(db)
+ elif choose == 3:
+ # 3、统计每个班级的平均就业时长（仅统计进入就业阶段，即有就业开放时间的学生） -- 参数: 无
+ data = StatisticsDao.find_class_avg_employment_duration(db)
+
+ return data
+
+# 2.6.3 就业数据统计 单独版
+"""
#### 2.6.3 就业数据统计
# - 统计就业薪资排名 Top N（动态输入 N）的学生的姓名、班级、就业时间和就业公司。
-@router.get("/api/stats/employment/top_salaries",
+@router.get("/employment/top_salaries",
response_model=List[TopSalaryStudentItem],
summary="就业薪资top_N",
description="统计就业薪资排名 Top N（动态输入 N）的学生的姓名、班级、就业时间和就业公司")
@@ -132,7 +168,7 @@ def get_top_salary_students(
return data
# - 统计每个学生的就业时长（计算公式：offer下发时间 - 就业开放时间）。
-@router.get("/api/stats/employment/job_seeking_duration",
+@router.get("/employment/job_seeking_duration",
response_model=List[StudentEmpDurationItem],
summary="统计就业时长",
description="统计每个学生的就业时长（计算公式：offer下发时间 - 就业开放时间）")
@@ -141,10 +177,11 @@ def get_students_job_seeking_duration(db: Session = Depends(get_db)):
return data
# - 统计每个班级的平均就业时长（仅统计进入就业阶段，即有就业开放时间的学生）
-@router.get("/api/stats/class/avg_employment_duration",
+@router.get("/class/avg_employment_duration",
response_model=List[ClassAvgEmpDurationItem],
summary="班级平均就业时长",
description="统计每个班级的平均就业时长（仅统计进入就业阶段，即有就业开放时间的学生）")
def get_class_avg_employment_duration(db: Session = Depends(get_db)):
data = StatisticsDao.find_class_avg_employment_duration(db)
return data
+"""
\ No newline at end of file
diff --git a/dao/cls_mgmt_dao.py b/dao/cls_mgmt_dao.py
index d9f2777..9b27ff0 100644
--- a/dao/cls_mgmt_dao.py
+++ b/dao/cls_mgmt_dao.py
@@ -3,6 +3,7 @@
from fastapi import HTTPException
from sqlalchemy.orm import Session
+from model import Teacher
from model.cls_mgmt_model import ClsMgmt
from scheme.cls_mgmt_scheme import ClsMgmtResponse, ClsMgmtCreate
from typing import Optional, List
@@ -36,7 +37,9 @@ class ClsMgmtDAO:
根据主键 ID 获取单个班级
:return: 班级对象或 None
"""
- return db.query(ClsMgmt).filter(ClsMgmt.id == class_id,ClsMgmt.is_deleted == 0).first()
+ return (db.query(ClsMgmt)
+ .filter(ClsMgmt.id == class_id,ClsMgmt.is_deleted == 0)
+ .first())
@@ -68,6 +71,18 @@ class ClsMgmtDAO:
id 根据开课时间自动生成，不再自增
"""
class_id = ClsMgmtDAO.generate_class_id(cls_data.cls_start_date)
+ head_tea = db.query(Teacher).filter(
+ Teacher.type == "班主任",
+ Teacher.id == cls_data.head_tea_id
+ ).first()
+ lecturer = db.query(Teacher).filter(
+ Teacher.type == "授课老师",
+ Teacher.id == cls_data.lecturer_id
+ ).first()
+ if not head_tea:
+ raise HTTPException(status_code=404, detail=f"班主任id{cls_data.head_tea_id}不存在")
+ elif not lecturer:
+ raise HTTPException(status_code=404, detail=f"授课老师id{cls_data.lecturer_id}不存在")
new_cls = ClsMgmt(
id=class_id, # 手动赋值班级id
cls_start_date=cls_data.cls_start_date,
@@ -109,13 +124,23 @@ class ClsMgmtDAO:
ClsMgmt.id == class_id,
ClsMgmt.is_deleted == 0 # 只查未删除的
).first()
-
+ head_tea = db.query(Teacher).filter(
+ Teacher.type == "班主任",
+ Teacher.id == cls_data.head_tea_id
+ ).first()
+ lecturer = db.query(Teacher).filter(
+ Teacher.type == "授课老师",
+ Teacher.id == cls_data.lecturer_id
+ ).first()
if update_cls:
- update_cls.head_tea_id=cls_data.head_tea_id
- update_cls.lecturer_id=cls_data.lecturer_id
+ if not head_tea:
+ raise HTTPException(status_code=404, detail=f"班主任id{cls_data.head_tea_id}不存在")
+ elif not lecturer:
+ raise HTTPException(status_code=404, detail=f"授课老师id{cls_data.lecturer_id}不存在")
+ update_cls.head_tea_id = cls_data.head_tea_id
+ update_cls.lecturer_id = cls_data.lecturer_id
db.commit()
return update_cls
-
raise HTTPException(status_code=404, detail=f"班级{class_id}不存在")
diff --git a/dao/statistics_dao.py b/dao/statistics_dao.py
index 974444a..74e6009 100644
--- a/dao/statistics_dao.py
+++ b/dao/statistics_dao.py
@@ -20,7 +20,9 @@ class StatisticsDao:
@staticmethod
def get_class_gender_distribution(db: Session) -> List[ClassGenderDistributionItem]:
-
+ """
+ **多维度班级统计**：统计每个班级的总人数，以及按性别（男、女）细分的人数分布。
+ """
all_cls_gender_stats = (
db.query(
StuInfo.cls_id.label("cls_id"),
@@ -38,12 +40,16 @@ class StatisticsDao:
total_cnt=i.total_cnt,
man_cnt=i.man_cnt,
female_cnt=i.female_cnt,
- ) for i in all_cls_gender_stats
+ )
+ for i in all_cls_gender_stats
]
# 查询每次考试成绩都在输入分数线（如80分）以上的学生的编号、姓名和成绩。
@staticmethod
def get_stu_info_all_exams_above_score(db: Session, score: float) -> List[StuInfoAllExamAboveScore]:
+ """
+ # 查询每次考试成绩都在输入分数线（如80分）以上的学生的编号、姓名和成绩 -- 参数: 分数
+ """
stu_id_list_query = (
db.query(StuScore.stu_id)
.filter(StuScore.is_deleted == 0)
@@ -87,7 +93,11 @@ class StatisticsDao:
@staticmethod
def get_stu_dict_who_fail(db: Session) -> Dict[str, List[StuScore]]:
+ """
+ 获取不及格的学生信息 {stu_id, [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))
@@ -98,7 +108,14 @@ class StatisticsDao:
@staticmethod
def find_students_fail_count(db: Session, fail_cnt: int) -> List[FailingStudentItem]:
- test_max_cnt = len(db.query(StuScore.exam_attempt).group_by(StuScore.exam_attempt).all())
+ """
+ # 2、查询有输入指定次数（如两次）以上不及格的学生的姓名、班级和不及格成绩明细 -- 参数: 不及格次数
+ """
+ test_max_cnt = len(db.query(StuScore.exam_attempt)
+ .filter(StuScore.is_deleted == 0)
+ .group_by(StuScore.exam_attempt)
+ .all()
+ )
if fail_cnt > test_max_cnt:
raise HTTPException(status_code = 400, detail = f'最大次数为{test_max_cnt}')
@@ -131,9 +148,12 @@ class StatisticsDao:
# 统计每次考试每个班级的平均分，并支持按分数从高到低或从低到高动态排序。
@staticmethod
def find_class_exam_averages(db: Session, order_choose: int) -> List[ClassExamAvgScoreItem]:
+ """
+ # 统计每次考试每个班级的平均分，并支持按分数从高到低或从低到高动态排序 -- 参数: 1:低到高, 2:高到低
+ """
avg_score = func.avg(StuScore.exam_score)
order_score = asc(avg_score) if order_choose == 1 else desc(avg_score)
- aaa = (db.query(
+ finally_data = (db.query(
StuInfo.cls_id,
StuScore.exam_attempt,
func.avg(StuScore.exam_score).label("avg_score"),
@@ -146,7 +166,7 @@ class StatisticsDao:
# 班级 - 考试场次 - 平均分
ans = []
- for row in aaa:
+ for row in finally_data:
ans.append(
ClassExamAvgScoreItem(
exam_attempt=row.exam_attempt,
@@ -160,17 +180,20 @@ class StatisticsDao:
@staticmethod
def find_top_salary_students(db: Session, top_n: int) -> List[TopSalaryStudentItem]:
- aaa = (db.query(
+ """
+ # 1、统计就业薪资排名 Top N（动态输入 N）的学生的姓名、班级、就业时间和就业公司。 -- 参数: top_N
+ """
+ finally_data = (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)
- )
+ .filter(StudentEmployManage.is_deleted == 0, StudentEmployManage.is_deleted == 0)
+ .order_by(StudentEmployManage.salary.desc())
+ .limit(top_n)
+ )
ans = [
TopSalaryStudentItem(
@@ -180,26 +203,30 @@ class StatisticsDao:
emp_company_name=row.emp_company,
salary=row.salary
)
- for row in aaa
+ for row in finally_data
]
return ans
@staticmethod
def find_students_employment_duration(db: Session) -> List[StudentEmpDurationItem]:
+ """
+ # 2、统计每个学生的就业时长（计算公式：offer下发时间 - 就业开放时间） -- 参数: 无
+ """
time = func.datediff(StudentEmployManage.send_offer_time, StudentEmployManage.emp_open_time)
- aaa = (db.query(
+
+ finally_data = (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()
- )
+ .filter(StuInfo.is_deleted == 0,
+ StudentEmployManage.is_deleted == 0,
+ StudentEmployManage.send_offer_time.isnot(None),
+ StudentEmployManage.emp_open_time.isnot(None))
+ .order_by(time).all()
+ )
ans = [
StudentEmpDurationItem(
@@ -208,13 +235,16 @@ class StatisticsDao:
cls_id=row.cls_id,
offer_time=row.time
)
- for row in aaa
+ for row in finally_data
]
return ans
@staticmethod
def find_class_avg_employment_duration(db: Session) -> List[ClassAvgEmpDurationItem]:
+ """
+ # 3、统计每个班级的平均就业时长（仅统计进入就业阶段，即有就业开放时间的学生） -- 参数: 无
+ """
duration_time = func.datediff(StudentEmployManage.send_offer_time, StudentEmployManage.emp_open_time)
class_avg_query = (db.query(
StuInfo.cls_id,
diff --git a/main.py b/main.py
index 95b5cfa..7c29a3b 100644
--- a/main.py
+++ b/main.py
@@ -4,7 +4,7 @@
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
-from api import statistics_api, stu_score_api, cls_mgmt_api, employ_api,stu_api, teacher_api # 导入 users 子路由
+from api import statistics_api, stu_score_api, cls_mgmt_api, employ_api,stu_api, teacher_api
# 1. 创建数据库表（如果表不存在）
# Base.metadata.create_all 会扫描所有继承 Base 的模型，生成对应的 CREATE TABLE 语句
@@ -13,9 +13,11 @@ from api import statistics_api, stu_score_api, cls_mgmt_api, employ_api,stu_api,
# 2. 创建 FastAPI 实例
app = FastAPI(
- title="FastAPI + SQLAlchemy 分层架构（MySQL）",
- description="用户管理示例，演示分层架构和 MySQL 集成",
- version="1.0.0"
+ title="码力全开: 学生管理系统",
+ description="本项目旨在开发一个基于 FastAPI 的学生管理系统，"
+ "提供学生基本信息管理、考核成绩管理、就业管理和统计分析四大核心功能模块。"
+ "系统将采用 RESTful API 设计，支持前后端分离架构。
",
+ version="0.141.1"
)
# 3. 添加跨域中间件（允许前端跨域请求）
@@ -29,11 +31,11 @@ app.add_middleware(
# 4. 注册子路由
-app.include_router(statistics_api.router, prefix="/api/stats", tags=["统计分析"])
+app.include_router(statistics_api.router, prefix="/api/stats", tags=["统计分析模块(动态查询与综合能力锻炼)"])
-app.include_router(stu_score_api.router, prefix="/api/score", tags=["学生成绩管理"])
+app.include_router(stu_score_api.router, prefix="/api/score", tags=["学生成绩管理模块"])
-app.include_router(cls_mgmt_api.router, prefix="/api/classes", tags=["班级管理"])
+app.include_router(cls_mgmt_api.router, prefix="/api/classes", tags=["班级管理模块"])
app.include_router(employ_api.router,prefix="/api/employ",tags=["学生就业管理模块"])
@@ -43,7 +45,7 @@ app.include_router(stu_api.router, prefix="/api/student", tags=["学生信息管
# 5. 根路径
@app.get("/")
async def root():
- return {"message": "欢迎访问 FastAPI + SQLAlchemy 分层示例！请访问 /docs 查看 API 文档。"}
+ return {"message": "欢迎访问 码力全开: 学生管理系统 分层示例！请访问 /docs 查看 API 文档。"}
# 6. 如果直接运行此文件，启动 uvicorn 服务器
if __name__ == "__main__":
diff --git a/scheme/employ_scheme.py b/scheme/employ_scheme.py
index 24c16b2..6a4991b 100644
--- a/scheme/employ_scheme.py
+++ b/scheme/employ_scheme.py
@@ -47,10 +47,10 @@ class EmployStatusCreate(BaseModel):
"""
记录学生就业状态接口，需要传入的请求体
"""
- stu_id: str = Field(..., description="学号，非空唯一")
+ stu_id: str = Field(...,max_length=15,description="学号，非空唯一")
emp_open_time: YearMonthDay | None = Field(None, description="就业开放时间,可更改学生就业状态")
send_offer_time: YearMonthDay | None = Field(None, description="offer下发时间，可更改学生就业状态")
- emp_company: str | None = Field(None, description="就业公司名称")
+ emp_company: str | None = Field(None,max_length=50,description="就业公司名称")
salary: Decimal | None = Field(None, decimal_places=2,description="就业薪资，默认为空") # 薪资精确到2小数