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],
|
||
}
|