143 lines
5.5 KiB
Python
143 lines
5.5 KiB
Python
# dao/statistics_dao.py
|
|
# 统计分析数据访问层：
|
|
# 使用 SQLAlchemy 的聚合函数(func.count/avg/max/min)、分组(group_by)、关联(join)
|
|
# 以及动态查询参数，实现多维度统计分析。返回纯数据(dict/list)，不含任何 HTTP 概念。
|
|
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from model.advisor_model import AdvisorInfo
|
|
from model.cls_mgmt_model import ClsMgmt
|
|
from model.employ_model import StudentEmployManage
|
|
from model.stu_model import StuInfo
|
|
from model.stu_score_model import StuScore
|
|
from model.teacher_model import Teacher
|
|
|
|
|
|
def _count_alive(db: Session, model) -> int:
|
|
"""统计某张表的有效（未删除）记录数。"""
|
|
return db.query(func.count(model.id)).filter(model.is_deleted == 0).scalar() or 0
|
|
|
|
|
|
def _employed_query(db: Session):
|
|
"""已就业 = 未删除且已填公司名。"""
|
|
return db.query(StudentEmployManage).filter(
|
|
StudentEmployManage.is_deleted == 0,
|
|
StudentEmployManage.emp_company.isnot(None),
|
|
)
|
|
|
|
|
|
def overview(db: Session) -> Dict[str, Any]:
|
|
"""统计学生、班级、教师、顾问、已就业人数及就业率。"""
|
|
student_count = _count_alive(db, StuInfo)
|
|
class_count = _count_alive(db, ClsMgmt)
|
|
teacher_count = _count_alive(db, Teacher)
|
|
advisor_count = _count_alive(db, AdvisorInfo)
|
|
employed_count = _employed_query(db).count()
|
|
|
|
return {
|
|
"student_count": student_count,
|
|
"class_count": class_count,
|
|
"teacher_count": teacher_count,
|
|
"advisor_count": advisor_count,
|
|
"employed_count": employed_count,
|
|
"employment_rate": round(employed_count / student_count, 4) if student_count else 0,
|
|
}
|
|
|
|
|
|
def _group_count(db: Session, column, *filters) -> List[tuple]:
|
|
"""按某一列分组计数，返回 [(分组值, 数量), ...]。"""
|
|
return (
|
|
db.query(column, func.count(StuInfo.id))
|
|
.filter(StuInfo.is_deleted == 0, *filters)
|
|
.group_by(column)
|
|
.all()
|
|
)
|
|
|
|
|
|
def student_gender_distribution(db: Session) -> List[Dict[str, Any]]:
|
|
"""学生性别分布。"""
|
|
return [{"gender": g or "未知", "count": c} for g, c in _group_count(db, StuInfo.gender)]
|
|
|
|
|
|
def student_education_distribution(db: Session) -> List[Dict[str, Any]]:
|
|
"""学生学历分布。"""
|
|
return [{"education": e or "未知", "count": c} for e, c in _group_count(db, StuInfo.education)]
|
|
|
|
|
|
def student_state_distribution(db: Session) -> List[Dict[str, Any]]:
|
|
"""学生就读状态分布。"""
|
|
return [{"state": s or "未知", "count": c} for s, c in _group_count(db, StuInfo.state)]
|
|
|
|
|
|
def student_hometown_distribution(db: Session, limit: int = 10) -> List[Dict[str, Any]]:
|
|
"""学生生源地分布(Top N)。"""
|
|
rows = (
|
|
db.query(StuInfo.hometown, func.count(StuInfo.id))
|
|
.filter(StuInfo.is_deleted == 0, StuInfo.hometown.isnot(None))
|
|
.group_by(StuInfo.hometown)
|
|
.order_by(func.count(StuInfo.id).desc())
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
return [{"hometown": h, "count": c} for h, c in rows]
|
|
|
|
|
|
def score_summary(db: Session, cls_id: Optional[str] = None) -> Dict[str, Any]:
|
|
"""统计平均分、最高分、最低分、及格人数与及格率，可动态按班级过滤(关联 stu_info)。"""
|
|
base = (
|
|
db.query(StuScore)
|
|
.join(StuInfo, StuScore.stu_id == StuInfo.id)
|
|
.filter(StuScore.is_deleted == False, StuInfo.is_deleted == 0) # noqa: E712 布尔列
|
|
)
|
|
if cls_id:
|
|
base = base.filter(StuInfo.cls_id == cls_id)
|
|
|
|
total = base.count()
|
|
avg = base.with_entities(func.avg(StuScore.exam_score)).scalar()
|
|
max_score = base.with_entities(func.max(StuScore.exam_score)).scalar()
|
|
min_score = base.with_entities(func.min(StuScore.exam_score)).scalar()
|
|
pass_count = base.filter(StuScore.exam_score >= 60).count()
|
|
|
|
return {
|
|
"cls_id": cls_id,
|
|
"total": total,
|
|
"avg_score": round(float(avg), 2) if avg is not None else None,
|
|
"max_score": float(max_score) if max_score is not None else None,
|
|
"min_score": float(min_score) if min_score is not None else None,
|
|
"pass_count": pass_count,
|
|
"pass_rate": round(pass_count / total, 4) if total else 0,
|
|
}
|
|
|
|
|
|
def score_level_distribution(db: Session) -> List[Dict[str, Any]]:
|
|
"""成绩等级分布。"""
|
|
rows = (
|
|
db.query(StuScore.score_level, func.count(StuScore.id))
|
|
.filter(StuScore.is_deleted == False) # noqa: E712 布尔列
|
|
.group_by(StuScore.score_level)
|
|
.all()
|
|
)
|
|
return [{"score_level": s or "未知", "count": c} for s, c in rows]
|
|
|
|
|
|
def employ_summary(db: Session) -> Dict[str, Any]:
|
|
"""就业综合统计：就业率与薪资（平均 / 最高 / 最低）。"""
|
|
student_count = _count_alive(db, StuInfo)
|
|
|
|
employed = _employed_query(db)
|
|
employed_count = employed.count()
|
|
avg_salary = employed.with_entities(func.avg(StudentEmployManage.salary)).scalar()
|
|
max_salary = employed.with_entities(func.max(StudentEmployManage.salary)).scalar()
|
|
min_salary = employed.with_entities(func.min(StudentEmployManage.salary)).scalar()
|
|
|
|
return {
|
|
"student_count": student_count,
|
|
"employed_count": employed_count,
|
|
"employment_rate": round(employed_count / student_count, 4) if student_count else 0,
|
|
"avg_salary": round(float(avg_salary), 2) if avg_salary is not None else None,
|
|
"max_salary": float(max_salary) if max_salary is not None else None,
|
|
"min_salary": float(min_salary) if min_salary is not None else None,
|
|
} |