317 lines
14 KiB
Python
317 lines
14 KiB
Python
"""统计分析 DAO（需求 2.6 与 2.7.2 的 SQL 部分）。
|
||
|
||
这里只放聚合查询；需要在 Python 里做的二次计算（标准差、排名、趋势判断）
|
||
放在 ``app.service.statistics_service``。
|
||
|
||
原则：
|
||
能用一次 SQL 出结果的，绝不拉全表到 Python 里 for 循环。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from sqlalchemy import Select, case, distinct, func, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.utils import age_expression, date_diff_days_expr
|
||
from app.model import Clazz, Employment, Score, Student, class_teachers
|
||
|
||
|
||
class StatisticsDao:
|
||
# ================================================================ 2.6.1
|
||
@classmethod
|
||
def class_overview(cls, db: Session) -> list[dict]:
|
||
"""每个班的总人数 + 男女分布。LEFT JOIN 保证"挂零的班"也在结果里。"""
|
||
stmt = (
|
||
select(
|
||
Clazz.id,
|
||
Clazz.class_no,
|
||
Clazz.name,
|
||
Clazz.status,
|
||
Clazz.head_teacher_id,
|
||
func.count(Student.id).label("total"),
|
||
func.sum(case((Student.gender == 1, 1), else_=0)).label("male"),
|
||
func.sum(case((Student.gender == 2, 1), else_=0)).label("female"),
|
||
func.sum(case((Student.gender.notin_((1, 2)), 1), else_=0)).label("other"),
|
||
)
|
||
.select_from(Clazz)
|
||
.outerjoin(Student, (Student.class_id == Clazz.id) & (Student.is_del == 0))
|
||
.where(Clazz.is_del == 0)
|
||
.group_by(Clazz.id, Clazz.class_no, Clazz.name, Clazz.status, Clazz.head_teacher_id)
|
||
.order_by(Clazz.id.asc())
|
||
)
|
||
return [dict(row._mapping) for row in db.execute(stmt).all()]
|
||
|
||
@classmethod
|
||
def age_distribution(cls, db: Session) -> list[dict]:
|
||
"""年龄段分布（给饼图用）。"""
|
||
age_col = age_expression(Student.birth_date).label("age")
|
||
stmt = (
|
||
select(age_col, func.count(Student.id))
|
||
.where(Student.alive(), Student.birth_date.is_not(None))
|
||
.group_by(age_col)
|
||
.order_by(age_col)
|
||
)
|
||
return [{"age": int(a), "count": c} for a, c in db.execute(stmt).all()]
|
||
|
||
# ================================================================ 2.6.2
|
||
@classmethod
|
||
def all_above(cls, db: Session, threshold: float, class_id: int | None = None) -> list[dict]:
|
||
"""每场考核都在分数线以上的学生：GROUP BY + HAVING MIN(score) >= 阈值。"""
|
||
agg = (
|
||
select(
|
||
Score.stu_id.label("stu_id"),
|
||
func.count(Score.id).label("exam_count"),
|
||
func.min(Score.score).label("min_score"),
|
||
func.avg(Score.score).label("avg_score"),
|
||
func.max(Score.score).label("max_score"),
|
||
)
|
||
.where(Score.alive())
|
||
.group_by(Score.stu_id)
|
||
.having(func.min(Score.score) >= threshold)
|
||
.subquery()
|
||
)
|
||
stmt = (
|
||
select(
|
||
Student.id.label("stu_id"),
|
||
Student.stu_no,
|
||
Student.name,
|
||
Clazz.name.label("class_name"),
|
||
agg.c.exam_count,
|
||
agg.c.min_score,
|
||
agg.c.avg_score,
|
||
agg.c.max_score,
|
||
)
|
||
.join(agg, agg.c.stu_id == Student.id)
|
||
.outerjoin(Clazz, Clazz.id == Student.class_id)
|
||
.where(Student.alive())
|
||
.order_by(agg.c.avg_score.desc())
|
||
)
|
||
if class_id:
|
||
stmt = stmt.where(Student.class_id == class_id)
|
||
return [dict(row._mapping) for row in db.execute(stmt).all()]
|
||
|
||
@classmethod
|
||
def fail_students(cls, db: Session, threshold: float, min_times: int) -> list[dict]:
|
||
"""不及格次数 >= N 的学生：SUM(CASE WHEN score < 阈值 THEN 1 ELSE 0 END) >= N。"""
|
||
fail_flag = case((Score.score < threshold, 1), else_=0)
|
||
agg = (
|
||
select(
|
||
Score.stu_id.label("stu_id"),
|
||
func.sum(fail_flag).label("fail_times"),
|
||
func.count(Score.id).label("exam_count"),
|
||
func.avg(Score.score).label("avg_score"),
|
||
)
|
||
.where(Score.alive())
|
||
.group_by(Score.stu_id)
|
||
.having(func.sum(fail_flag) >= min_times)
|
||
.subquery()
|
||
)
|
||
stmt = (
|
||
select(
|
||
Student.id.label("stu_id"),
|
||
Student.stu_no,
|
||
Student.name,
|
||
Clazz.name.label("class_name"),
|
||
Clazz.class_no,
|
||
agg.c.fail_times,
|
||
agg.c.exam_count,
|
||
agg.c.avg_score,
|
||
)
|
||
.join(agg, agg.c.stu_id == Student.id)
|
||
.outerjoin(Clazz, Clazz.id == Student.class_id)
|
||
.where(Student.alive())
|
||
.order_by(agg.c.fail_times.desc())
|
||
)
|
||
return [dict(row._mapping) for row in db.execute(stmt).all()]
|
||
|
||
@classmethod
|
||
def class_exam_avg(
|
||
cls,
|
||
db: Session,
|
||
exam_seq: int | None = None,
|
||
class_id: int | None = None,
|
||
order: str = "desc",
|
||
pass_line: float = 60.0,
|
||
) -> list[dict]:
|
||
"""每场考核每个班级的平均分，支持按平均分动态升降序。
|
||
|
||
排序：`order` 是**主导**键 —— 需求 2.6.2 要的是「按分数从高到低/从低到高动态排序」，
|
||
如果让 exam_seq 排在前面，整张表就只是「每个场次内部有序」，
|
||
选「全部场次 + 从高到低」时看到的不是排名而是 5 个分块。
|
||
场次和班级 ID 只作次级稳定键。
|
||
"""
|
||
pass_flag = case((Score.score >= pass_line, 1), else_=0)
|
||
avg_col = func.avg(Score.score).label("avg_score")
|
||
stmt = (
|
||
select(
|
||
Score.exam_seq,
|
||
Clazz.id.label("class_id"),
|
||
Clazz.name.label("class_name"),
|
||
Clazz.class_no,
|
||
func.count(Score.id).label("student_count"),
|
||
avg_col,
|
||
func.max(Score.score).label("max_score"),
|
||
func.min(Score.score).label("min_score"),
|
||
(func.sum(pass_flag) * 100.0 / func.count(Score.id)).label("pass_rate"),
|
||
)
|
||
.join(Student, Student.id == Score.stu_id)
|
||
.join(Clazz, Clazz.id == Student.class_id)
|
||
.where(Score.alive(), Student.alive(), Clazz.alive())
|
||
)
|
||
if exam_seq is not None:
|
||
stmt = stmt.where(Score.exam_seq == exam_seq)
|
||
if class_id is not None:
|
||
stmt = stmt.where(Clazz.id == class_id)
|
||
stmt = stmt.group_by(Score.exam_seq, Clazz.id, Clazz.name, Clazz.class_no)
|
||
stmt = stmt.order_by(
|
||
avg_col.desc() if order.lower() == "desc" else avg_col.asc(),
|
||
Score.exam_seq.asc(),
|
||
Clazz.id.asc(),
|
||
)
|
||
return [dict(row._mapping) for row in db.execute(stmt).all()]
|
||
|
||
# ================================================================ 2.6.3
|
||
@classmethod
|
||
def salary_top(cls, db: Session, top_n: int) -> list[dict]:
|
||
"""就业薪资 Top N：ORDER BY salary DESC LIMIT N。"""
|
||
stmt = (
|
||
select(
|
||
Student.id.label("stu_id"),
|
||
Student.stu_no,
|
||
Student.name,
|
||
Clazz.name.label("class_name"),
|
||
Employment.company,
|
||
Employment.salary,
|
||
Employment.open_date,
|
||
Employment.offer_date,
|
||
Employment.city,
|
||
)
|
||
.join(Employment, Employment.stu_id == Student.id)
|
||
.outerjoin(Clazz, Clazz.id == Student.class_id)
|
||
.where(Student.alive(), Employment.alive(), Employment.salary.is_not(None))
|
||
.order_by(Employment.salary.desc())
|
||
.limit(top_n)
|
||
)
|
||
return [dict(row._mapping) for row in db.execute(stmt).all()]
|
||
|
||
@classmethod
|
||
def student_durations(cls, db: Session, class_id: int | None = None) -> list[dict]:
|
||
"""每个学生的就业时长（SQL 侧算天数）。"""
|
||
duration = date_diff_days_expr(Employment.offer_date, Employment.open_date).label("duration_days")
|
||
stmt = (
|
||
select(
|
||
Student.id.label("stu_id"),
|
||
Student.stu_no,
|
||
Student.name,
|
||
Clazz.name.label("class_name"),
|
||
Employment.company,
|
||
Employment.open_date,
|
||
Employment.offer_date,
|
||
duration,
|
||
)
|
||
.join(Employment, Employment.stu_id == Student.id)
|
||
.outerjoin(Clazz, Clazz.id == Student.class_id)
|
||
.where(Student.alive(), Employment.alive(), Employment.open_date.is_not(None))
|
||
.order_by(duration.desc())
|
||
)
|
||
if class_id:
|
||
stmt = stmt.where(Student.class_id == class_id)
|
||
return [dict(row._mapping) for row in db.execute(stmt).all()]
|
||
|
||
@classmethod
|
||
def class_avg_duration(cls, db: Session) -> list[dict]:
|
||
"""每个班级的平均就业时长（只统计已开放就业的学生）。"""
|
||
duration = date_diff_days_expr(Employment.offer_date, Employment.open_date)
|
||
stmt = (
|
||
select(
|
||
Clazz.id.label("class_id"),
|
||
Clazz.class_no,
|
||
Clazz.name.label("class_name"),
|
||
func.count(distinct(Student.id)).label("student_count"),
|
||
func.count(distinct(case((Employment.open_date.is_not(None), Student.id)))).label("open_count"),
|
||
func.count(distinct(case((Employment.offer_date.is_not(None), Student.id)))).label("offer_count"),
|
||
func.avg(case((Employment.offer_date.is_not(None), duration))).label("avg_duration_days"),
|
||
func.min(case((Employment.offer_date.is_not(None), duration))).label("min_duration_days"),
|
||
func.max(case((Employment.offer_date.is_not(None), duration))).label("max_duration_days"),
|
||
)
|
||
.select_from(Clazz)
|
||
.outerjoin(Student, (Student.class_id == Clazz.id) & (Student.is_del == 0))
|
||
.outerjoin(Employment, (Employment.stu_id == Student.id) & (Employment.is_del == 0))
|
||
.where(Clazz.is_del == 0)
|
||
.group_by(Clazz.id, Clazz.class_no, Clazz.name)
|
||
.order_by(func.avg(case((Employment.offer_date.is_not(None), duration))))
|
||
)
|
||
return [dict(row._mapping) for row in db.execute(stmt).all()]
|
||
|
||
# ================================================================ 2.7.2
|
||
@classmethod
|
||
def class_funnel(cls, db: Session, high_line: float) -> list[dict]:
|
||
"""班级就业漏斗：总人数 -> 已就业 -> 高薪，就业率在 Python 里算完排序。"""
|
||
stmt = (
|
||
select(
|
||
Clazz.id.label("class_id"),
|
||
Clazz.class_no,
|
||
Clazz.name.label("class_name"),
|
||
func.count(distinct(Student.id)).label("total"),
|
||
func.count(distinct(case((Employment.offer_date.is_not(None), Student.id)))).label("employed"),
|
||
func.count(distinct(case((Employment.salary >= high_line, Student.id)))).label("high_salary"),
|
||
func.avg(case((Employment.offer_date.is_not(None), Employment.salary))).label("avg_salary"),
|
||
func.avg(
|
||
case(
|
||
(Employment.offer_date.is_not(None),
|
||
date_diff_days_expr(Employment.offer_date, Employment.open_date)),
|
||
)
|
||
).label("avg_duration_days"),
|
||
)
|
||
.select_from(Clazz)
|
||
.outerjoin(Student, (Student.class_id == Clazz.id) & (Student.is_del == 0))
|
||
.outerjoin(Employment, (Employment.stu_id == Student.id) & (Employment.is_del == 0))
|
||
.where(Clazz.is_del == 0)
|
||
.group_by(Clazz.id, Clazz.class_no, Clazz.name)
|
||
)
|
||
return [dict(row._mapping) for row in db.execute(stmt).all()]
|
||
|
||
@classmethod
|
||
def score_stats_per_student(cls, db: Session) -> list[dict]:
|
||
"""每个学生的成绩次数 / 平均 / 最高 / 最低 / 最大分差（SQL 聚合）。"""
|
||
stmt = (
|
||
select(
|
||
Score.stu_id,
|
||
func.count(Score.id).label("exam_count"),
|
||
func.avg(Score.score).label("avg_score"),
|
||
func.max(Score.score).label("max_score"),
|
||
func.min(Score.score).label("min_score"),
|
||
(func.max(Score.score) - func.min(Score.score)).label("score_range"),
|
||
)
|
||
.where(Score.alive())
|
||
.group_by(Score.stu_id)
|
||
.having(func.count(Score.id) >= 2) # 只有一次成绩谈不上波动
|
||
)
|
||
return [dict(row._mapping) for row in db.execute(stmt).all()]
|
||
|
||
@classmethod
|
||
def score_samples(cls, db: Session, stu_ids: list[int] | None = None) -> dict[int, list[dict]]:
|
||
"""按学生取成绩明细（按序次升序），一次查完。"""
|
||
stmt = select(Score.stu_id, Score.exam_seq, Score.score, Score.flag).where(Score.alive())
|
||
if stu_ids:
|
||
stmt = stmt.where(Score.stu_id.in_(stu_ids))
|
||
stmt = stmt.order_by(Score.stu_id, Score.exam_seq)
|
||
result: dict[int, list[dict]] = {}
|
||
for stu_id, seq, score, flag in db.execute(stmt).all():
|
||
result.setdefault(stu_id, []).append(
|
||
{"exam_seq": int(seq), "score": float(score), "flag": int(flag)}
|
||
)
|
||
return result
|
||
|
||
# ================================================================ 总览
|
||
@classmethod
|
||
def teacher_class_load(cls, db: Session) -> list[dict]:
|
||
"""老师带班数量（给概览用）。"""
|
||
stmt = (
|
||
select(class_teachers.c.teacher_id, func.count(class_teachers.c.class_id))
|
||
.select_from(class_teachers.join(Clazz, Clazz.id == class_teachers.c.class_id))
|
||
.where(Clazz.alive())
|
||
.group_by(class_teachers.c.teacher_id)
|
||
)
|
||
return [{"teacher_id": t, "class_count": c} for t, c in db.execute(stmt).all()]
|