111 lines
4.2 KiB
Python
111 lines
4.2 KiB
Python
from typing import List
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Path, Query
|
|
from sqlalchemy.orm import Session
|
|
|
|
from dao.wl_statistics_dao import StatisticDao
|
|
from database import get_db
|
|
from dao import wl_statistics_dao
|
|
from scheme.wl_statistics_scheme import (AgeCompareOp, ClassAvgEmpTimeOut,
|
|
ClassAvgScoreOut, ClassGenderStatOut,
|
|
EmpTopOut, FailCountModel,
|
|
ScoreOutLine, SortOrder, EmpTimeOut)
|
|
from scheme.wl_student_scheme import StudentOut
|
|
|
|
router = APIRouter(prefix='/statistics',tags=["统计分析模块"])
|
|
|
|
|
|
# ---- 2.6.1.1 动态年龄范围查询 ----
|
|
@router.get(
|
|
"/students/age",
|
|
response_model=list[StudentOut],
|
|
summary="动态年龄范围查询",
|
|
responses={400: {"description": "查询参数不合法，例如区间查询缺少上界 value2"}},
|
|
)
|
|
def query_age_range(
|
|
op: AgeCompareOp = Query(AgeCompareOp.gt, description="比较条件：gt/ge/lt/le/eq/between"),
|
|
value: int = Query(..., description="年龄阈值"),
|
|
value2: int | None = Query(None, description="年龄上界，仅 op=between 时必填"),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
return StatisticDao.search_students_by_age(db, op, value, value2)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
# ---- 2.6.1.2 多维度班级统计 ----
|
|
@router.get(
|
|
'/classes/count',
|
|
summary="多维度班级统计",
|
|
response_model=list[ClassGenderStatOut]
|
|
)
|
|
def query_classes_count(db: Session = Depends(get_db)):
|
|
return StatisticDao.statistic_classes_count(db)
|
|
|
|
# 查询每次考试成绩都在输入分数线（如80分）以上的学生的编号、姓名和成绩
|
|
@router.get(
|
|
'/scores/out_score_line/{score_line}',
|
|
summary='统计每次超分数线的学生',
|
|
response_model=list[ScoreOutLine]
|
|
)
|
|
def query_out_score_line(score_line:float,db:Session=Depends(get_db)):
|
|
return StatisticDao.query_score_by_line(db, score_line)
|
|
|
|
# 查询有输入指定次数（如两次）以上不及格的学生的姓名、班级和不及格成绩明细
|
|
@router.get(
|
|
'/scores/fail_score/{fail_count}',
|
|
summary='统计不及格次数超过指定次数的学生信息',
|
|
response_model=list[FailCountModel]
|
|
)
|
|
def query_info_by_fail_count(fail_count:int,db: Session = Depends(get_db)):
|
|
return StatisticDao.query_by_fail_count(db,fail_count)
|
|
|
|
# 统计每次考试每个班级的平均分，并支持按分数从高到低或从低到高动态排序
|
|
@router.get(
|
|
'/scores/get_class_avg_score',
|
|
summary='统计每次考试每个班级的平均分，并支持按分数从高到低或从低到高动态排序',
|
|
response_model=list[ClassAvgScoreOut]
|
|
)
|
|
def get_class_avg_score(
|
|
sort_order: SortOrder | None = Query(
|
|
None, description="排序方向：asc 从低到高，desc 从高到低；不传则按班级+考试序次"),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
return StatisticDao.get_class_avg_score(db, sort_order)
|
|
|
|
# 统计就业薪资排名Top N（动态输入N）的学生的姓名、班级、就业时间和就业公司
|
|
@router.get(
|
|
'/emp/top_rank/{rank}',
|
|
summary='统计就业薪资排名Top N',
|
|
response_model=list[EmpTopOut]
|
|
)
|
|
def get_top_rank(
|
|
rank: int = Path(..., gt=0, description="取薪资排名前 N 名"),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
return StatisticDao.get_info_by_rank(db, rank)
|
|
|
|
# 统计每个学生的就业时长（计算公式：offer下发时间-就业开放时间）
|
|
@router.get(
|
|
'/emp/emp_time',
|
|
summary='统计每个学生的就业时长',
|
|
response_model=list[EmpTimeOut]
|
|
)
|
|
def get_emp_time(db: Session = Depends(get_db)):
|
|
return StatisticDao.get_emp_time(db)
|
|
|
|
# 统计每个班级的平均就业时长（仅统计已进入就业阶段且已拿到 offer 的学生）
|
|
@router.get(
|
|
'/emp/class_avg_emp_time',
|
|
summary='统计每个班级的平均就业时长',
|
|
response_model=list[ClassAvgEmpTimeOut]
|
|
)
|
|
def get_class_avg_emp_time(
|
|
sort_order: SortOrder | None = Query(
|
|
None, description="排序方向：asc 从低到高，desc 从高到低；不传则按班级编号"),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
return StatisticDao.get_class_avg_emp_time(db, sort_order)
|
|
|
|
|