diff --git a/api1/__init__.py b/api/__init__.py similarity index 100% rename from api1/__init__.py rename to api/__init__.py diff --git a/api/cls_mgmt_api.py b/api/cls_mgmt_api.py new file mode 100644 index 0000000..44389be --- /dev/null +++ b/api/cls_mgmt_api.py @@ -0,0 +1,144 @@ + +from pathlib import Path + +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy.orm import Session +from typing import List + +from dao.cls_mgmt_dao import ClsMgmtDAO +from database import get_db +from scheme.cls_mgmt_scheme import ClsMgmtResponse, ClsMgmtCreate +from scheme.users import UserCreate, UserUpdate, UserResponse +router = APIRouter() + +@router.get("/", response_model=List[ClsMgmtResponse]) +# @router.get("/",response_model=ClsMgmtResponse) +async def get_classes( + skip: int = Query(0, ge=0, description="跳过的记录数"), + limit: int = Query(100, ge=1, le=200, description="返回的最大记录数"), + db: Session = Depends(get_db) # 依赖注入获得数据库会话 +): + """ + 获取班级列表，支持分页 + """ + classes = ClsMgmtDAO.get_all(db, skip=skip, limit=limit) + return classes # FastAPI 自动根据 response_model 转换为 JSON + +# @router.get("/{class_id}", response_model=ClsMgmtResponse) +# @router.get("/{class_id}") +# async def get_class( +# class_id: str = Path(description="班级id:260814"), +# db: Session = Depends(get_db) +# ): +# """ +# 根据班级 ID 获取详细信息 +# """ +# user = ClsMgmtDAO.get_by_id(db, class_id) +# if not user: +# raise HTTPException(status_code=404, detail="班级不存在") +# return user + +@router.get("/{class_id}",response_model=ClsMgmtResponse) # ← 加这一行 +async def get_class( + class_id: str = Path(description="班级id:260814"), + db: Session = Depends(get_db) +): + """ + 根据班级 ID 获取详细信息 + """ + cls = ClsMgmtDAO.get_by_id(db, class_id) + if not cls: + raise HTTPException(status_code=404, detail="班级不存在") + return cls # 直接返回 ORM 对象，FastAPI 自动转成 ClsMgmtResponse +# @router.get( +# "/{class_id}", +# response_model="ClsMgmtResponse", +# summary="根据班级id获取详细信息" +# ) +# async def get_class( +# class_id: str = Path(description="班级id,例:260814"), +# db: Session = Depends(get_db) +# ): +# cls = ClsMgmtDAO.get_by_id(db, class_id) +# if not cls: +# raise HTTPException(status_code=404, detail="班级不存在") +# return cls + +# @router.post("/{class_id}") # ← 加这一行 +# async def add_class( +# class_id: str = Path(description="班级id:260814"), +# db: Session = Depends(get_db) +# ): +# """ +# 根据班级 ID 获取详细信息 +# """ +# cls = ClsMgmtDAO.get_by_id(db, class_id) +# if not cls: +# raise HTTPException(status_code=404, detail="班级不存在") +# return cls + +@router.post( + "/add_class", + response_model=ClsMgmtResponse + # status_code=201, + # summary="新增班级" +) +async def create_class( + cls_data: ClsMgmtCreate, + db: Session = Depends(get_db) +): + """ + 新增班级 + - 班级id自动生成：根据开课时间取后6位（如 2026-08-14 → 260814） + - 先根据班级id验证是否已存在 + - 已存在 → 返回 400「班级已存在」 + - 不存在 → 新增班级 + """ + # 第一步：根据开课时间生成班级id + class_id = ClsMgmtDAO.generate_class_id(cls_data.cls_start_date) + # 第二步：根据班级id验证是否已存在 + if ClsMgmtDAO.exists(db, class_id): + raise HTTPException( + status_code=400, + detail=f"班级已存在（班级id：{class_id}）" + ) + # 第三步：不存在则新增 + new_cls = ClsMgmtDAO.create(db, cls_data) + return new_cls + +@router.delete( + "/del_class/{class_id}", + # response_model=ClsMgmtResponse + # status_code=201, + # summary="新增班级" +) +async def del_class( + class_id: str = Path(description="班级id:260814"), + db: Session = Depends(get_db) +): + """ + 删除班级 + """ + print(class_id) + del_cls = ClsMgmtDAO.delete_class(db, class_id) + return del_cls + +@router.put( + "/update_class/{class_id}", + response_model=ClsMgmtResponse + # status_code=201, + # summary="新增班级" +) +async def update_class( + cls_data: ClsMgmtCreate, + class_id: str = Path(description="班级id:260814"), + + db: Session = Depends(get_db) +): + """ + 根据班级ID修改班主任和授课老师信息 + """ + update_cls = ClsMgmtDAO.update_cls(db, cls_data,class_id) + return update_cls + + diff --git a/api1/employ_api.py b/api/employ_api.py similarity index 100% rename from api1/employ_api.py rename to api/employ_api.py diff --git a/api/statistics_api.py b/api/statistics_api.py new file mode 100644 index 0000000..af7a624 --- /dev/null +++ b/api/statistics_api.py @@ -0,0 +1,135 @@ +# api/stats +# 本文件定义统计分析相关的所有 API 路由（Controller 层） +from itertools import count +from pydoc import describe + +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy.orm import Session +from typing import List, Optional +from sqlalchemy import func +from starlette import status + +from database import get_db +from dao.users_dao import UserDAO +from dao.statistics_dao import StatisticsDao +from model.stu_model import StuInfo +from scheme.statistics_scheme import StatsResponse, StuInfoAllExamAboveScore, FailingStudentItem, ClassExamAvgScoreItem, \ + TopSalaryStudentItem, StudentEmpDurationItem, ClassAvgEmpDurationItem, StudentAgeDetailItem +from scheme.users import UserCreate, UserUpdate, UserResponse + +router = APIRouter() + +#### 2.6.1 基本信息动态统计 +# - **动态年龄范围查询**：支持用户输入年龄阈值及比较条件（如大于、小于、等于、区间等），动态查询符合条件的学员信息。 +# - **多维度班级统计**：统计每个班级的总人数，以及按性别（男、女）细分的人数分布。 +@router.get("/api/stats/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"), + 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用)") + ): + + if operator == "between": + if min_age is None or max_age is None: + raise HTTPException(status_code=400, detail="最大/最小值 都得输入") + if min_age > max_age: + raise HTTPException(status_code=400, detail="最小值 要 小于 最大值！") + else: + if age is None: + raise HTTPException(status_code=400, detail="age 不能不写！") + + data = StatisticsDao.find_students_by_age_condition(db = db, + operator = operator, + age = age, + min_age = min_age, + max_age = max_age) + return data + + +# **多维度班级统计**：统计每个班级的总人数，以及按性别（男、女）细分的人数分布。 +@router.get("/class/stats/gender", + response_model=List[StatsResponse], + summary="班级性别统计", + description="统计每个班级的总人数，以及按性别（男、女）细分的人数分布。") +def get_class_stats(db: Session = Depends(get_db)): + data = StatisticsDao.get_cls_stats(db) + return data + +#### 2.6.2 成绩综合统计 +# 1- 查询每次考试成绩都在输入分数线（如80分）以上的学生的编号、姓名和成绩。 +# 2- 查询有输入指定次数（如两次）以上不及格的学生的姓名、班级和不及格成绩明细。 +# 3- 统计每次考试每个班级的平均分，并支持按分数从高到低或从低到高动态排序。 +@router.get("/api/stats/students/all_exams_above_score", + response_model = List[StuInfoAllExamAboveScore], + summary="查询成绩在xx分以上的学生信息", + description= "查询成绩在xx分以上的学生信息") +def get_students_all_exams_above_score( + db: Session = Depends(get_db), + score: float = Query(..., ge=0.0, le=100.0, description="输入最低成绩(包含)"), +): + data = StatisticsDao.get_stu_info_all_exams_above_score(db, score) + return data + +# 查询有输入指定次数（如两次）以上不及格的学生的姓名、班级和不及格成绩明细。 +@router.get("/api/stats/failing_records", + response_model=List[FailingStudentItem], + summary="查询有输入指定次数不及格学生", + description="查询有输入指定次数（如两次）以上不及格的学生的姓名、班级和不及格成绩明细。") +def get_frequent_failing_students( + db: Session = Depends(get_db), + fail_cnt: int = Query(..., ge = 0, le = 100, description="不及格次数") +): + data = StatisticsDao.find_students_fail_count(db, fail_cnt) + return data + +# 统计每次考试每个班级的平均分，并支持按分数从高到低或从低到高动态排序。 +@router.get("/api/stats/class/exam_avg_scores", + response_model=List[ClassExamAvgScoreItem], + summary="统计每次考试每个班级的平均分", + description="统计每次考试每个班级的平均分，并支持按分数从高到低或从低到高动态排序") +def get_class_exams_avg_scores( + db: Session = Depends(get_db), + order_choose: int = Query(..., ge = 1, le = 2, description="1: 升序， 2: 降序") +): + data = StatisticsDao.find_class_exam_averages(db, order_choose) + return data + + +#### 2.6.3 就业数据统计 + +# - 统计就业薪资排名 Top N（动态输入 N）的学生的姓名、班级、就业时间和就业公司。 +# - 统计每个学生的就业时长（计算公式：offer下发时间 - 就业开放时间）。 +# - 统计每个班级的平均就业时长（仅统计进入就业阶段，即有就业开放时间的学生） +@router.get("/api/stats/employment/top_salaries", + response_model=List[TopSalaryStudentItem], + summary="就业薪资top_N", + description="统计就业薪资排名 Top N（动态输入 N）的学生的姓名、班级、就业时间和就业公司") +def get_top_salary_students( + db: Session = Depends(get_db), + top_n: int = Query(default=5, ge=1, description="获取前N名")): + data = StatisticsDao.find_top_salary_students(db, top_n) + return data + + +# - 统计每个学生的就业时长（计算公式：offer下发时间 - 就业开放时间）。 +@router.get("/api/stats/employment/job_seeking_duration", + response_model=List[StudentEmpDurationItem], + summary="统计就业时长", + description="统计每个学生的就业时长（计算公式：offer下发时间 - 就业开放时间）") +def get_students_job_seeking_duration(db: Session = Depends(get_db)): + data = StatisticsDao.find_students_employment_duration(db) + return data + + +# - 统计每个班级的平均就业时长（仅统计进入就业阶段，即有就业开放时间的学生） +@router.get("/api/stats/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 diff --git a/api1/stu_api.py b/api/stu_api.py similarity index 100% rename from api1/stu_api.py rename to api/stu_api.py diff --git a/api1/stu_score_api.py b/api/stu_score_api.py similarity index 100% rename from api1/stu_score_api.py rename to api/stu_score_api.py diff --git a/api1/teacher_api.py b/api/teacher_api.py similarity index 100% rename from api1/teacher_api.py rename to api/teacher_api.py diff --git a/dao/cls_mgmt_dao.py b/dao/cls_mgmt_dao.py new file mode 100644 index 0000000..bbdabe7 --- /dev/null +++ b/dao/cls_mgmt_dao.py @@ -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}不存在" + + + + diff --git a/dao/statistics_dao.py b/dao/statistics_dao.py new file mode 100644 index 0000000..eab5757 --- /dev/null +++ b/dao/statistics_dao.py @@ -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 diff --git a/main.py b/main.py index 934830a..83a248c 100644 --- a/main.py +++ b/main.py @@ -5,7 +5,7 @@ from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from database import engine, Base -from api1 import users, statistics_api, stu_score_api, cls_mgmt_api, employ_api,stu_api, teacher_api # 导入 users 子路由 +from api import users, statistics_api, stu_score_api, cls_mgmt_api, employ_api,stu_api, teacher_api # 导入 users 子路由 # 1. 创建数据库表（如果表不存在） # Base.metadata.create_all 会扫描所有继承 Base 的模型，生成对应的 CREATE TABLE 语句 diff --git a/scheme/cls_mgmt_scheme.py b/scheme/cls_mgmt_scheme.py new file mode 100644 index 0000000..f522971 --- /dev/null +++ b/scheme/cls_mgmt_scheme.py @@ -0,0 +1,36 @@ +# import data +from h11 import Data +from pydantic import BaseModel, Field, ConfigDict,EmailStr +from datetime import datetime, date +from typing import Optional + +# ---------- 请求模型 ---------- +class ClsMgmtCreate(BaseModel): + # id: str = Field(..., min_length=3, max_length=50) + cls_start_date: date=Field(...,description="开课时间，年-月-日") + head_tea_id: str=Field(...,description="班主任工号,例:t001") + lecturer_id: str = Field(..., description="班主任工号,例:t001") + +class ClsMgmtUpdate(BaseModel): + # id: str = Field(..., min_length=3, max_length=50) + # cls_start_date: Data=Field(...,description="开课时间，年-月-日") + head_tea_id: str=Field(...,description="班主任工号,例:t001") + lecturer_id: str = Field(..., description="班主任工号,例:t001") + +# ---------- 响应模型 ---------- +class ClsMgmtResponse(BaseModel): + id: str + cls_start_date: date + head_tea_id: str + lecturer_id: str + class Config: + from_attributes = True # 支持 ORM 对象转换 + + + + # model_config = ConfigDict(from_attributes=True) + # class Config: + # orm_mode = True + # created_at: datetime + # updated_at: Optional[datetime] + # diff --git a/scheme/statistics_scheme.py b/scheme/statistics_scheme.py new file mode 100644 index 0000000..0409579 --- /dev/null +++ b/scheme/statistics_scheme.py @@ -0,0 +1,69 @@ +# scheme/statistics_scheme.py +from pydantic import BaseModel, Field, EmailStr +from datetime import datetime, date +from typing import Optional, List + + +# ---------- 请求模型 ---------- +class StatsScoreRequest(BaseModel): + choose: int = Field(..., ge=0, le=3) + + email: EmailStr + full_name: Optional[str] = Field(None, max_length=100) + +# ---------- 响应模型 ---------- +# 多维度班级统计 +class StatsResponse(BaseModel): + cls_id: str + total_cnt: int + man_cnt: int + female_cnt: int + +class ExamScoreItem(BaseModel): + exam_attempt: int + exam_score: float + +# 查询每次考试成绩都在输入分数线（如80分）以上的学生的编号、姓名和成绩。 +class StuInfoAllExamAboveScore(BaseModel): + stu_id: str + stu_name: str + stu_score: List[ExamScoreItem] + +# 查询有输入指定次数（如两次）以上不及格的学生的姓名、班级和不及格成绩明细。 +class FailingStudentItem(BaseModel): + stu_name: str + stu_cls_id: str + stu_fail_cnt: int + stu_score: List[ExamScoreItem] + +# 查询有输入指定次数（如两次）以上不及格的学生的姓名、班级和不及格成绩明细。 +class ClassExamAvgScoreItem(BaseModel): + exam_attempt: int + cls_id: str + avg_score: float + stu_cnt: int + +class TopSalaryStudentItem(BaseModel): + stu_name: str + cls_id: str + offer_time: date + emp_company_name: str + salary: float + +class StudentEmpDurationItem(BaseModel): + stu_id: str + stu_name: str + cls_id: str + offer_time: int + +class ClassAvgEmpDurationItem(BaseModel): + cls_id: str + avg_duration_days: float + employed_cnt: int + +class StudentAgeDetailItem(BaseModel): + stu_id: str + stu_name: str + cls_id: str + gender: str + age: int \ No newline at end of file