From e8be87eca031c6e36d9d61f89ac4675c006cc12d Mon Sep 17 00:00:00 2001 From: jianqi Date: Mon, 14 Sep 2026 16:39:48 +0800 Subject: [PATCH] =?UTF-8?q?=E7=AC=AC=E4=BA=94=E7=89=88=5F=E5=AE=8C?= =?UTF-8?q?=E6=95=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- api/statistics_api.py | 75 ++++++++++++++++++++++++++++++----------- dao/cls_mgmt_dao.py | 35 ++++++++++++++++--- dao/statistics_dao.py | 68 ++++++++++++++++++++++++++----------- main.py | 18 +++++----- scheme/employ_scheme.py | 4 +-- 5 files changed, 147 insertions(+), 53 deletions(-) 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小数