sh_stu0814/max_code_test_gitea_2
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

第五版_完整

Browse Source
This commit is contained in:
jianqi
2026-09-14 16:39:48 +08:00
parent c2909191d2
commit e8be87eca0
5 changed files with 147 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/statistics_api.py
+56 -19
View File
@@ -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 <br> 或者:>/>=/</<=/==/between <br>"""),
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<br>
2、统计每个学生的就业时长（计算公式：offer下发时间 - 就业开放时间） -- 参数: 无<br>
3、统计每个班级的平均就业时长（仅统计进入就业阶段，即有就业开放时间的学生） -- 参数: 无<br>
""")
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、统计每个班级的平均就业时长（仅统计进入就业阶段，即有就业开放时间的学生） -- 参数: 无<br>
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
"""
dao/cls_mgmt_dao.py
+30 -5
View File
@@ -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}不存在")
dao/statistics_dao.py
+49 -19
View File
@@ -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、统计每个班级的平均就业时长（仅统计进入就业阶段，即有就业开放时间的学生） -- 参数: 无<br>
"""
duration_time = func.datediff(StudentEmployManage.send_offer_time, StudentEmployManage.emp_open_time)
class_avg_query = (db.query(
StuInfo.cls_id,
main.py
+10 -8
View File
@@ -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 设计，支持前后端分离架构。<br>",
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__":
scheme/employ_scheme.py
+2 -2
View File
@@ -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小数