368 lines
15 KiB
Python
368 lines
15 KiB
Python
|
"""统计分析业务层（需求 2.6 / 2.7.2）。
|
|||
|
|
|
|||
|
|
DAO 出聚合结果，这里做二次计算：排名、比率、标准差、趋势、四舍五入。
|
|||
|
|
这样 SQL 保持"只取数"，业务口径全部集中在 Python 里，改口径不用改 SQL。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import statistics
|
|||
|
|
from statistics import StatisticsError
|
|||
|
|
|
|||
|
|
from sqlalchemy import func, or_, select
|
|||
|
|
from sqlalchemy.orm import Session
|
|||
|
|
|
|||
|
|
from app.core.config import settings
|
|||
|
|
from app.core.exceptions import BusinessError
|
|||
|
|
from app.core.utils import age_expression
|
|||
|
|
from app.dao.advisor_dao import AdvisorDao
|
|||
|
|
from app.dao.clazz_dao import ClazzDao
|
|||
|
|
from app.dao.employment_dao import EmploymentDao
|
|||
|
|
from app.dao.score_dao import ScoreDao
|
|||
|
|
from app.dao.statistics_dao import StatisticsDao
|
|||
|
|
from app.dao.student_dao import StudentDao
|
|||
|
|
from app.dao.teacher_dao import TeacherDao
|
|||
|
|
from app.model import Score, Student
|
|||
|
|
from app.model.constants import CLASS_STATUS_TEXT, STUDENT_STATUS_TEXT
|
|||
|
|
|
|||
|
|
ROUND = 2
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _r(value, digits: int = ROUND):
|
|||
|
|
return round(value, digits) if value is not None else None
|
|||
|
|
|
|||
|
|
|
|||
|
|
class StatisticsService:
|
|||
|
|
# ============================================================ 2.6.1
|
|||
|
|
@classmethod
|
|||
|
|
def age_query(
|
|||
|
|
cls,
|
|||
|
|
db: Session,
|
|||
|
|
operator: str,
|
|||
|
|
value: int | None = None,
|
|||
|
|
value2: int | None = None,
|
|||
|
|
**filters,
|
|||
|
|
):
|
|||
|
|
"""把"大于/小于/等于/区间"翻译成 age_min / age_max 两个参数。"""
|
|||
|
|
op = (operator or "gte").strip().lower()
|
|||
|
|
age_min = age_max = None
|
|||
|
|
|
|||
|
|
if op in ("gt", ">"):
|
|||
|
|
age_min = (value or 0) + 1
|
|||
|
|
elif op in ("gt_eq", "gte", ">="):
|
|||
|
|
age_min = value
|
|||
|
|
elif op in ("lt", "<"):
|
|||
|
|
age_max = (value or 0) - 1
|
|||
|
|
elif op in ("lt_eq", "lte", "<="):
|
|||
|
|
age_max = value
|
|||
|
|
elif op in ("eq", "=", "=="):
|
|||
|
|
age_min = age_max = value
|
|||
|
|
elif op in ("ne", "!="):
|
|||
|
|
# "不等于"用区间实现：<= v-1 或 >= v+1
|
|||
|
|
pass
|
|||
|
|
elif op in ("between", "range"):
|
|||
|
|
if value is None or value2 is None:
|
|||
|
|
raise BusinessError("区间查询需要同时提供 value 和 value2")
|
|||
|
|
age_min, age_max = min(value, value2), max(value, value2)
|
|||
|
|
else:
|
|||
|
|
raise BusinessError(
|
|||
|
|
"operator 只支持 gt/gte/lt/lte/eq/between（也接受 > >= < <= = 写法）"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
if age_min is None and age_max is None and op not in ("ne",):
|
|||
|
|
raise BusinessError("请提供年龄阈值 value")
|
|||
|
|
|
|||
|
|
stmt = StudentDao.build_stmt(age_min=age_min, age_max=age_max, **filters)
|
|||
|
|
if op == "ne":
|
|||
|
|
age_col = age_expression(Student.birth_date)
|
|||
|
|
stmt = stmt.where(or_(age_col < value, age_col > value))
|
|||
|
|
return stmt
|
|||
|
|
|
|||
|
|
@classmethod
|
|||
|
|
def class_overview(cls, db: Session) -> list[dict]:
|
|||
|
|
rows = StatisticsDao.class_overview(db)
|
|||
|
|
teacher_map = {t.id: t.name for t in TeacherDao.all(db, TeacherDao.build_stmt())}
|
|||
|
|
result = []
|
|||
|
|
for row in rows:
|
|||
|
|
total = int(row["total"] or 0)
|
|||
|
|
male = int(row["male"] or 0)
|
|||
|
|
female = int(row["female"] or 0)
|
|||
|
|
result.append(
|
|||
|
|
{
|
|||
|
|
"class_id": row["id"],
|
|||
|
|
"class_no": row["class_no"],
|
|||
|
|
"class_name": row["name"],
|
|||
|
|
"status_text": {1: "在读", 2: "已结课", 3: "已解散"}.get(row["status"]),
|
|||
|
|
"head_teacher_name": None,
|
|||
|
|
"total": total,
|
|||
|
|
"male": male,
|
|||
|
|
"female": female,
|
|||
|
|
"other": total - male - female,
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
# 补班主任名字（一次查全表映射，避免 N+1）
|
|||
|
|
for item, row in zip(result, rows):
|
|||
|
|
item["head_teacher_name"] = teacher_map.get(row.get("head_teacher_id")) if "head_teacher_id" in row else None
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
# ============================================================ 2.6.2
|
|||
|
|
@classmethod
|
|||
|
|
def all_above(cls, db: Session, threshold: float, class_id: int | None = None) -> list[dict]:
|
|||
|
|
rows = StatisticsDao.all_above(db, threshold, class_id)
|
|||
|
|
samples = StatisticsDao.score_samples(db, [r["stu_id"] for r in rows])
|
|||
|
|
for row in rows:
|
|||
|
|
row["min_score"] = _r(row["min_score"])
|
|||
|
|
row["avg_score"] = _r(row["avg_score"])
|
|||
|
|
row["scores"] = samples.get(row["stu_id"], [])
|
|||
|
|
return rows
|
|||
|
|
|
|||
|
|
@classmethod
|
|||
|
|
def fail_students(cls, db: Session, threshold: float, min_times: int) -> list[dict]:
|
|||
|
|
rows = StatisticsDao.fail_students(db, threshold, min_times)
|
|||
|
|
samples = StatisticsDao.score_samples(db, [r["stu_id"] for r in rows])
|
|||
|
|
for row in rows:
|
|||
|
|
row["avg_score"] = _r(row["avg_score"])
|
|||
|
|
row["fail_details"] = [
|
|||
|
|
s for s in samples.get(row["stu_id"], []) if s["score"] < threshold
|
|||
|
|
]
|
|||
|
|
return rows
|
|||
|
|
|
|||
|
|
@classmethod
|
|||
|
|
def class_exam_avg(
|
|||
|
|
cls, db: Session, exam_seq: int | None, class_id: int | None, order: str, pass_line: float
|
|||
|
|
) -> list[dict]:
|
|||
|
|
rows = StatisticsDao.class_exam_avg(db, exam_seq, class_id, order, pass_line)
|
|||
|
|
for row in rows:
|
|||
|
|
row["avg_score"] = _r(row["avg_score"])
|
|||
|
|
row["max_score"] = _r(row["max_score"])
|
|||
|
|
row["min_score"] = _r(row["min_score"])
|
|||
|
|
row["pass_rate"] = _r(row["pass_rate"], 1)
|
|||
|
|
return rows
|
|||
|
|
|
|||
|
|
# ============================================================ 2.6.3
|
|||
|
|
@classmethod
|
|||
|
|
def salary_top(cls, db: Session, top_n: int) -> list[dict]:
|
|||
|
|
if top_n <= 0:
|
|||
|
|
raise BusinessError("top_n 必须为正整数")
|
|||
|
|
rows = StatisticsDao.salary_top(db, top_n)
|
|||
|
|
result = []
|
|||
|
|
for index, row in enumerate(rows, start=1):
|
|||
|
|
salary = float(row["salary"]) if row["salary"] is not None else 0.0
|
|||
|
|
duration = None
|
|||
|
|
if row["open_date"] and row["offer_date"]:
|
|||
|
|
duration = (row["offer_date"] - row["open_date"]).days
|
|||
|
|
result.append(
|
|||
|
|
{
|
|||
|
|
"rank": index,
|
|||
|
|
"stu_id": row["stu_id"],
|
|||
|
|
"stu_no": row["stu_no"],
|
|||
|
|
"name": row["name"],
|
|||
|
|
"class_name": row["class_name"],
|
|||
|
|
"company": row["company"],
|
|||
|
|
"salary": salary,
|
|||
|
|
"salary_wan": _r(salary / 10000),
|
|||
|
|
"offer_date": row["offer_date"],
|
|||
|
|
"duration_days": duration,
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
@classmethod
|
|||
|
|
def student_durations(cls, db: Session, class_id: int | None = None) -> list[dict]:
|
|||
|
|
rows = StatisticsDao.student_durations(db, class_id)
|
|||
|
|
for row in rows:
|
|||
|
|
row["duration_days"] = int(row["duration_days"]) if row["duration_days"] is not None else None
|
|||
|
|
row["state"] = "已拿到 offer" if row["offer_date"] else "就业中（未收到 offer）"
|
|||
|
|
return rows
|
|||
|
|
|
|||
|
|
@classmethod
|
|||
|
|
def class_avg_duration(cls, db: Session) -> list[dict]:
|
|||
|
|
rows = StatisticsDao.class_avg_duration(db)
|
|||
|
|
result = []
|
|||
|
|
for row in rows:
|
|||
|
|
open_count = int(row["open_count"] or 0)
|
|||
|
|
result.append(
|
|||
|
|
{
|
|||
|
|
"class_id": row["class_id"],
|
|||
|
|
"class_name": row["class_name"],
|
|||
|
|
"student_count": int(row["student_count"] or 0),
|
|||
|
|
"open_count": open_count,
|
|||
|
|
"offer_count": int(row["offer_count"] or 0),
|
|||
|
|
"avg_duration_days": _r(row["avg_duration_days"], 1),
|
|||
|
|
"min_duration_days": int(row["min_duration_days"]) if row["min_duration_days"] is not None else None,
|
|||
|
|
"max_duration_days": int(row["max_duration_days"]) if row["max_duration_days"] is not None else None,
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
# 平均时长按"有数据"的排前面，没数据的沉底
|
|||
|
|
result.sort(key=lambda x: (x["avg_duration_days"] is None, -(x["avg_duration_days"] or 0)))
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
# ============================================================ 2.7.2
|
|||
|
|
@classmethod
|
|||
|
|
def score_volatility(cls, db: Session, top_n: int = 5, metric: str = "stddev") -> list[dict]:
|
|||
|
|
"""成绩波动 Top N。
|
|||
|
|
|
|||
|
|
标准差在 Python 里算（statistics.pstdev），最大分差在 SQL 里算，
|
|||
|
|
两种方式都在这一份代码里做了，答辩时可以直接对比口径差异。
|
|||
|
|
"""
|
|||
|
|
stats_rows = StatisticsDao.score_stats_per_student(db)
|
|||
|
|
if not stats_rows:
|
|||
|
|
return []
|
|||
|
|
samples = StatisticsDao.score_samples(db, [r["stu_id"] for r in stats_rows])
|
|||
|
|
|
|||
|
|
student_map = {
|
|||
|
|
s.id: s
|
|||
|
|
for s in StudentDao.all(db, StudentDao.build_stmt(order_by="id", order="asc"))
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
items = []
|
|||
|
|
for row in stats_rows:
|
|||
|
|
values = [s["score"] for s in samples.get(row["stu_id"], [])]
|
|||
|
|
if len(values) < 2:
|
|||
|
|
continue
|
|||
|
|
try:
|
|||
|
|
stddev = statistics.pstdev(values)
|
|||
|
|
except StatisticsError:
|
|||
|
|
stddev = 0.0
|
|||
|
|
score_range = float(row["score_range"] or 0)
|
|||
|
|
faculty = {
|
|||
|
|
"stddev": stddev,
|
|||
|
|
"range": score_range,
|
|||
|
|
"max_minus_min": score_range,
|
|||
|
|
}
|
|||
|
|
if metric not in faculty:
|
|||
|
|
raise BusinessError("metric 只支持 stddev（标准差）或 range（最大分差）")
|
|||
|
|
trend_slope = _r(cls._trend(values)) or 0.0 # 先定稿到出参精度，再据它判方向
|
|||
|
|
student = student_map.get(row["stu_id"])
|
|||
|
|
items.append(
|
|||
|
|
{
|
|||
|
|
"stu_id": row["stu_id"],
|
|||
|
|
"stu_no": student.stu_no if student else None,
|
|||
|
|
"name": student.name if student else None,
|
|||
|
|
"class_name": student.class_name if student else None,
|
|||
|
|
"exam_count": int(row["exam_count"]),
|
|||
|
|
"avg_score": _r(row["avg_score"]),
|
|||
|
|
"stddev": _r(stddev),
|
|||
|
|
"score_range": _r(score_range),
|
|||
|
|
"trend": cls._trend_text(trend_slope, len(values)),
|
|||
|
|
"trend_slope": trend_slope,
|
|||
|
|
"scores": values,
|
|||
|
|
"_key": faculty[metric],
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
items.sort(key=lambda x: x["_key"], reverse=True)
|
|||
|
|
for index, item in enumerate(items[:top_n], start=1):
|
|||
|
|
item["rank"] = index
|
|||
|
|
item.pop("_key", None)
|
|||
|
|
return items[:top_n]
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def _trend(values: list[float]) -> float:
|
|||
|
|
"""成绩趋势：对「考核序次」做最小二乘拟合，返回斜率（分/场）。
|
|||
|
|
|
|||
|
|
为什么不用「首次 vs 末次」：单次考试成绩噪声很大，首末两点一比，
|
|||
|
|
89 -> 88 这种正常抖动会被说成"下降"，而 32 -> 82 这种真实回升又可能因为
|
|||
|
|
末次恰好回落而被说成"下降"。
|
|||
|
|
为什么不用「前半段 vs 后半段均值的差」：那是阶跃比较，看不出变化幅度，
|
|||
|
|
也没法排序。而且它对 5 次成绩的分段是不均的（前 2 后 3），
|
|||
|
|
会出现"首考 87.8、末考 82.0 却判为上升"这种让人不信服的结果。
|
|||
|
|
|
|||
|
|
最小二乘斜率是趋势的标准定义：用上全部数据点、抗噪声、可正可负、单位可解释。
|
|||
|
|
"""
|
|||
|
|
n = len(values)
|
|||
|
|
if n < 2:
|
|||
|
|
return 0.0
|
|||
|
|
x_mean = (n - 1) / 2
|
|||
|
|
y_mean = sum(values) / n
|
|||
|
|
numerator = sum((i - x_mean) * (v - y_mean) for i, v in enumerate(values))
|
|||
|
|
denominator = sum((i - x_mean) ** 2 for i in range(n))
|
|||
|
|
if denominator == 0:
|
|||
|
|
return 0.0
|
|||
|
|
return numerator / denominator
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def _trend_text(slope: float, n: int) -> str:
|
|||
|
|
"""把斜率翻成中文。阈值 1 分/场：一学期 5 次考核看下来涨跌不到 4 分，算原地踏步。
|
|||
|
|
|
|||
|
|
注意：调用方必须先把斜率四舍五入到出参精度（2 位小数）再传进来，
|
|||
|
|
否则会出现「标签写 基本持平、旁边数字写 +1.0/场」这种自相矛盾的显示。
|
|||
|
|
"""
|
|||
|
|
if n < 2:
|
|||
|
|
return "数据不足"
|
|||
|
|
if slope >= 1.0:
|
|||
|
|
return "上升"
|
|||
|
|
if slope <= -1.0:
|
|||
|
|
return "下降"
|
|||
|
|
return "基本持平"
|
|||
|
|
|
|||
|
|
@classmethod
|
|||
|
|
def employment_funnel(cls, db: Session, high_line: float | None = None) -> list[dict]:
|
|||
|
|
high_line = high_line or settings.HIGH_SALARY_LINE
|
|||
|
|
rows = StatisticsDao.class_funnel(db, high_line)
|
|||
|
|
result = []
|
|||
|
|
for row in rows:
|
|||
|
|
total = int(row["total"] or 0)
|
|||
|
|
employed = int(row["employed"] or 0)
|
|||
|
|
high = int(row["high_salary"] or 0)
|
|||
|
|
result.append(
|
|||
|
|
{
|
|||
|
|
"class_id": row["class_id"],
|
|||
|
|
"class_no": row["class_no"],
|
|||
|
|
"class_name": row["class_name"],
|
|||
|
|
"total": total,
|
|||
|
|
"employed": employed,
|
|||
|
|
"high_salary": high,
|
|||
|
|
"employment_rate": _r(employed / total * 100, 1) if total else 0.0,
|
|||
|
|
"high_salary_rate": _r(high / employed * 100, 1) if employed else 0.0,
|
|||
|
|
"avg_salary": _r(row["avg_salary"]),
|
|||
|
|
"avg_duration_days": _r(row["avg_duration_days"], 1),
|
|||
|
|
"high_line": high_line,
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
result.sort(key=lambda x: x["employment_rate"], reverse=True)
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
# ============================================================ 总览
|
|||
|
|
@classmethod
|
|||
|
|
def overview(cls, db: Session) -> dict:
|
|||
|
|
status_map = StudentDao.count_by_status(db)
|
|||
|
|
student_total = sum(status_map.values())
|
|||
|
|
|
|||
|
|
emp_stats = EmploymentDao.salary_stats(db, settings.HIGH_SALARY_LINE)
|
|||
|
|
|
|||
|
|
score_total = int(db.scalar(select(func.count(Score.id)).where(Score.alive())) or 0)
|
|||
|
|
score_avg = db.scalar(select(func.avg(Score.score)).where(Score.alive()))
|
|||
|
|
warning_ids = StudentDao.warning_student_ids(db)
|
|||
|
|
|
|||
|
|
funnel = cls.employment_funnel(db)
|
|||
|
|
|
|||
|
|
return {
|
|||
|
|
"student_total": student_total,
|
|||
|
|
"student_by_status": {
|
|||
|
|
STUDENT_STATUS_TEXT.get(code, str(code)): count
|
|||
|
|
for code, count in sorted(status_map.items())
|
|||
|
|
},
|
|||
|
|
"class_total": ClazzDao.count(db, ClazzDao.build_stmt()),
|
|||
|
|
"teacher_total": TeacherDao.count(db, TeacherDao.build_stmt()),
|
|||
|
|
"advisor_total": AdvisorDao.count(db, AdvisorDao.build_stmt()),
|
|||
|
|
"employment_total": emp_stats["count"],
|
|||
|
|
"employment_rate": _r(emp_stats["count"] / student_total * 100, 1) if student_total else 0.0,
|
|||
|
|
"avg_salary": _r(emp_stats["avg"]),
|
|||
|
|
"high_salary_count": emp_stats["high"],
|
|||
|
|
"score_record_total": score_total,
|
|||
|
|
"score_avg": _r(float(score_avg)) if score_avg is not None else None,
|
|||
|
|
"warning_student_count": len(warning_ids),
|
|||
|
|
"top_classes": [
|
|||
|
|
{
|
|||
|
|
"class_name": item["class_name"],
|
|||
|
|
"class_no": item["class_no"],
|
|||
|
|
"total": item["total"],
|
|||
|
|
"employed": item["employed"],
|
|||
|
|
"employment_rate": item["employment_rate"],
|
|||
|
|
}
|
|||
|
|
for item in funnel[:5]
|
|||
|
|
],
|
|||
|
|
"volatility_top": cls.score_volatility(db, 5)[:5],
|
|||
|
|
}
|