"""统计口径独立对账。 后端自检只校验了 `code == 0` 和「返回非空」,口径算错了它发现不了。 这里把每个统计接口的结果,和**用 Python 直接从原始列重算一遍**的值对比, 其中年龄/标准差/就业率/及格率都在 Python 侧独立实现,不借用被测代码的 SQL 表达式。 用法: python verify/statistics_crosscheck.py [--base http://127.0.0.1:8010] """ from __future__ import annotations import argparse import statistics import sys from datetime import date from pathlib import Path import httpx from sqlalchemy import text sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from app.core.database import engine # noqa: E402 PASS: list[str] = [] FAIL: list[str] = [] BASE = "http://127.0.0.1:8010" def check(title: str, ok: bool, detail: str = "") -> None: if ok: PASS.append(title) print(f" [PASS] {title}" + (f" {detail}" if detail else "")) else: FAIL.append(f"{title} — {detail}") print(f" [FAIL] {title} {detail}") def section(name: str) -> None: print(f"\n== {name} ==") def today() -> date: return date.today() def full_years(birth: date, ref: date) -> int: """完整周岁数:生日还没到就减一。这是「年龄」的正确算法。""" return ref.year - birth.year - ((ref.month, ref.day) < (birth.month, birth.day)) # ============================================================ 原始数据 def load_raw() -> dict: """注意:MySQL 的 DECIMAL 列取回来是 Decimal,跟 float 混算会 TypeError, 这里在入口统一转成 float,后面的比较就干净了。""" with engine.connect() as c: classes = [dict(r._mapping) for r in c.execute(text( "SELECT id, class_no, name, capacity, status FROM clazz WHERE is_del=0"))] students = [dict(r._mapping) for r in c.execute(text( "SELECT id, stu_no, name, gender, birth_date, class_id, status FROM student WHERE is_del=0"))] scores = [dict(r._mapping) for r in c.execute(text( "SELECT id, stu_id, exam_seq, score, flag FROM score WHERE is_del=0"))] emps = [dict(r._mapping) for r in c.execute(text( "SELECT id, stu_id, class_id, open_date, offer_date, salary FROM employment WHERE is_del=0"))] for s in scores: s["score"] = float(s["score"]) for e in emps: e["salary"] = float(e["salary"]) if e["salary"] is not None else None return {"classes": classes, "students": students, "scores": scores, "emps": emps} def main() -> None: global BASE parser = argparse.ArgumentParser() parser.add_argument("--base", default=BASE) args = parser.parse_args() BASE = args.base raw = load_raw() students, scores, emps, classes = raw["students"], raw["scores"], raw["emps"], raw["classes"] stu_by_id = {s["id"]: s for s in students} cls_by_id = {c["id"]: c for c in classes} http = httpx.Client(base_url=BASE, timeout=60) tok = http.post("/auth/login", json={"username": "admin", "password": "admin123"}).json()["data"]["access_token"] H = {"Authorization": f"Bearer {tok}"} def api(path: str): b = http.get(path, headers=H).json() assert b.get("code") == 0, f"{path} 返回 code={b.get('code')} {b.get('msg')}" return b["data"] # ---------------- 年龄 ---------------- section("2.6.1 动态年龄查询(Python 独立算年龄对账)") age_expected = {} for s in students: age_expected[s["id"]] = full_years(s["birth_date"], today()) if s["birth_date"] else None for op, val, val2 in [("gte", 25, None), ("gt", 25, None), ("lt", 22, None), ("lte", 22, None), ("eq", 24, None), ("between", 23, 26)]: q = f"/statistics/students/by-age?operator={op}&value={val}&page_size=1" if val2 is not None: q += f"&value2={val2}" d = api(q) if op == "between": lo, hi = min(val, val2), max(val, val2) exp = sum(1 for a in age_expected.values() if a is not None and lo <= a <= hi) else: fn = {"gte": lambda a: a >= val, "gt": lambda a: a > val, "lt": lambda a: a < val, "lte": lambda a: a <= val, "eq": lambda a: a == val}[op] exp = sum(1 for a in age_expected.values() if a is not None and fn(a)) check(f"age {op} {val}{('/' + str(val2)) if val2 else ''} 命中数一致", d["total"] == exp, f"接口 {d['total']} / 独立算 {exp}") # 逐条核对返回行里的 age 字段 d = api("/statistics/students/by-age?operator=gte&value=18&page_size=200") bad = [r for r in d["items"] if r.get("age") != age_expected.get(r["id"])] check("返回行里的 age 与 Python 周岁算法逐一一致", not bad, f"{len(d['items'])} 行中有 {len(bad)} 行不符" + (f" 例:{bad[0]['name']} 接口={bad[0].get('age')} 应为={age_expected[bad[0]['id']]}" if bad else "")) # ---------------- 班级性别分布 ---------------- section("2.6.1 班级人数与性别分布") ov = api("/statistics/classes/overview") check("班级数与库一致", len(ov) == len(classes), f"接口 {len(ov)} / 库 {len(classes)}") for row in ov: mine = [s for s in students if s["class_id"] == row["class_id"]] exp_total = len(mine) exp_male = sum(1 for s in mine if s["gender"] == 1) exp_female = sum(1 for s in mine if s["gender"] == 2) exp_other = exp_total - exp_male - exp_female ok = (row["total"], row["male"], row["female"], row["other"]) == (exp_total, exp_male, exp_female, exp_other) check(f"「{row['class_name']}」总/男/女/未填 一致", ok, f"接口 {row['total']}/{row['male']}/{row['female']}/{row['other']} vs 独立 {exp_total}/{exp_male}/{exp_female}/{exp_other}") check(f"「{row['class_name']}」男+女+未填 == 总数", row["male"] + row["female"] + row["other"] == row["total"]) check("全库男女人数之和 == 学生总数", sum(r["male"] + r["female"] + r["other"] for r in ov) == len(students), f"分布合计 {sum(r['male'] + r['female'] + r['other'] for r in ov)} / 学生 {len(students)}") # ---------------- 年龄分布 ---------------- section("年龄分布直方图") dist = api("/statistics/students/age-distribution") hist: dict[int, int] = {} for a in age_expected.values(): if a is not None: hist[a] = hist.get(a, 0) + 1 check("直方图各桶计数与独立统计一致", {r["age"]: r["count"] for r in dist} == hist, f"接口 {len(dist)} 桶 / 独立 {len(hist)} 桶") check("直方图人数合计 == 学生总数", sum(r["count"] for r in dist) == len(students), f"{sum(r['count'] for r in dist)} / {len(students)}") # ---------------- 每次考核都达标 ---------------- section("2.6.2 每次考核都在分数线以上") TH = 80.0 above = api(f"/statistics/scores/all-above?threshold={TH}") per_stu: dict[int, list[float]] = {} for sc in scores: per_stu.setdefault(sc["stu_id"], []).append(sc["score"]) exp_ids = {sid for sid, vals in per_stu.items() if vals and min(vals) >= TH} got_ids = {r["stu_id"] for r in above} check(f"{TH:g} 分以上(每次考核都达标)学生集合完全一致", got_ids == exp_ids, f"接口 {len(got_ids)} 人 / 独立 {len(exp_ids)} 人" + (f" 差集 {got_ids ^ exp_ids}" if got_ids != exp_ids else "")) bad = [] for r in above: vals = per_stu[r["stu_id"]] if r["exam_count"] != len(vals) or abs(r["min_score"] - min(vals)) > 1e-6 or abs(r["avg_score"] - sum(vals) / len(vals)) > 0.01: bad.append(r["name"]) check("每人考核次数/最低分/平均分 一致", not bad, f"不符:{bad[:3]}") # ---------------- 多次不及格 ---------------- section("2.6.2 不及格次数达到 N 次") LINE, MIN_T = 60.0, 2 fails = api(f"/statistics/scores/failures?threshold={LINE}&min_times={MIN_T}") exp_fail = {sid: [v for v in vals if v < LINE] for sid, vals in per_stu.items()} exp_fail = {sid: v for sid, v in exp_fail.items() if len(v) >= MIN_T} got_fail = {r["stu_id"]: r for r in fails} check("不及格学生集合完全一致", set(got_fail) == set(exp_fail), f"接口 {len(got_fail)} 人 / 独立 {len(exp_fail)} 人" + (f" 差集 {set(got_fail) ^ set(exp_fail)}" if set(got_fail) != set(exp_fail) else "")) bad = [r["name"] for r in fails if r["fail_times"] != len(exp_fail.get(r["stu_id"], []))] check("每人不及格次数一致", not bad, f"不符:{bad[:3]}") bad = [r["name"] for r in fails if len(r["fail_details"]) != r["fail_times"] or any(x["score"] >= LINE for x in r["fail_details"])] check("不及格明细条数正确且每条都低于红线", not bad, f"不符:{bad[:3]}") # ---------------- 班级平均分 ---------------- section("2.6.2 每场考核每个班级的平均分") ca = api("/statistics/scores/class-average?order=desc") exp_groups: dict[tuple[int, int], list[float]] = {} for sc in scores: stu = stu_by_id.get(sc["stu_id"]) if not stu or stu["class_id"] is None: continue exp_groups.setdefault((sc["exam_seq"], stu["class_id"]), []).append(sc["score"]) check("(场次 × 班级) 组合数一致", len(ca) == len(exp_groups), f"接口 {len(ca)} / 独立 {len(exp_groups)}") bad = [] for row in ca: key = (row["exam_seq"], row["class_id"]) vals = exp_groups.get(key) if vals is None: bad.append(f"多余 {key}") continue exp_avg = round(sum(vals) / len(vals), 2) exp_pass = round(sum(1 for v in vals if v >= 60) / len(vals) * 100, 1) if abs(row["avg_score"] - exp_avg) > 0.011 or row["max_score"] != max(vals) \ or row["min_score"] != min(vals) or row["student_count"] != len(vals) \ or abs(row["pass_rate"] - exp_pass) > 0.11: bad.append(f"{row['class_name']} 第{row['exam_seq']}次 接口({row['avg_score']}/{row['max_score']}" f"/{row['min_score']}/{row['student_count']}/{row['pass_rate']}) 独立({exp_avg}/{max(vals)}" f"/{min(vals)}/{len(vals)}/{exp_pass})") check("每组的 均分/最高/最低/人数/及格率 全部一致", not bad, f"问题 {len(bad)} 组;例:{bad[:1]}") avgs = [r["avg_score"] for r in ca] check("默认按平均分降序排列", avgs == sorted(avgs, reverse=True), f"前 5:{avgs[:5]}") asc = api("/statistics/scores/class-average?order=asc") avgs_asc = [r["avg_score"] for r in asc] check("order=asc 时升序排列", avgs_asc == sorted(avgs_asc), f"前 5:{avgs_asc[:5]}") # 回归守卫:曾经 ORDER BY 把 exam_seq 放在主位,导致「按平均分排序」只在每个场次内部生效, # 选「全部场次 + 从高到低」看到的不是排名而是 5 个分块。这里专门盯住这个坑。 all_avgs = sorted({r["avg_score"] for r in ca}, reverse=True) check("降序时第一名就是全局最高平均分", ca[0]["avg_score"] == all_avgs[0], f"实际首行 {ca[0]['avg_score']}({ca[0]['class_name']} 第{ca[0]['exam_seq']}次) / 全局最高 {all_avgs[0]}") seqs = [r["exam_seq"] for r in ca] check("降序列表里场次是交叉出现的(说明场均分是主导键,不是按场次分块)", len(set(seqs[:4])) > 1, f"前 4 行场次:{seqs[:4]}") # 不做「互为逆序」的严格断言:平均分相同的组之间,SQL 不保证次序, # 这种断言迟早会 flaky。只钉住「集合相同 + 各自单调」。 key_of = lambda rs: sorted((r["exam_seq"], r["class_id"]) for r in rs) # noqa: E731 check("升降序返回的是同一组数据(只是方向不同)", key_of(ca) == key_of(asc) and len(ca) == len(asc), f"desc {len(ca)} 行 / asc {len(asc)} 行") check("降序非递增、升序非递减", all(avgs[i] >= avgs[i + 1] for i in range(len(avgs) - 1)) and all(avgs_asc[i] <= avgs_asc[i + 1] for i in range(len(avgs_asc) - 1))) one = api("/statistics/scores/class-average?exam_seq=1") check("exam_seq=1 只返回第 1 次考试", all(r["exam_seq"] == 1 for r in one) and len(one) == len(classes), f"{len(one)} 行,场次 {sorted({r['exam_seq'] for r in one})}") # ---------------- 薪资 Top N ---------------- section("2.6.3 就业薪资 Top N") with_offer = [e for e in emps if e["offer_date"] and e["salary"] is not None] exp_top = sorted(with_offer, key=lambda e: (-e["salary"], e["id"]))[:5] top = api("/statistics/employment/top-salary?top_n=5") check("Top5 条数正确", len(top) == 5, f"{len(top)} 条") check("Top5 薪资与独立排序完全一致", [r["salary"] for r in top] == [e["salary"] for e in exp_top], f"接口 {[r['salary'] for r in top]} vs 独立 {[e['salary'] for e in exp_top]}") check("Top5 学生与独立排序一致", [r["stu_id"] for r in top] == [e["stu_id"] for e in exp_top]) check("名次从 1 开始连续", [r["rank"] for r in top] == [1, 2, 3, 4, 5]) check("薪资降序", [r["salary"] for r in top] == sorted([r["salary"] for r in top], reverse=True)) bad = [r["name"] for r in top if abs(r["salary_wan"] - round(r["salary"] / 10000, 2)) > 0.011] check("薪资(万) 换算正确", not bad, f"不符:{bad[:3]}") # ---------------- 每人就业时长 ---------------- section("2.6.3 每个学生的就业时长") durs = api("/statistics/employment/durations") check("条数 == 就业记录数", len(durs) == len(emps), f"接口 {len(durs)} / 库 {len(emps)}") emp_by_stu = {e["stu_id"]: e for e in emps} bad, missing = [], [] for r in durs: e = emp_by_stu.get(r["stu_id"]) if e is None: missing.append(r["stu_id"]) continue if e["offer_date"] and e["open_date"]: exp = (e["offer_date"] - e["open_date"]).days # Python 直接算天数 if r["duration_days"] != exp: bad.append(f"{r['name']} 接口 {r['duration_days']} 应 {exp}") elif r["duration_days"] is not None: bad.append(f"{r['name']} 无 offer 却有时长 {r['duration_days']}") check("每人时长 == offer 日 - 开放日(Python 独立算)", not bad and not missing, f"不符 {len(bad)} 条,缺 {len(missing)} 条;例:{bad[:2]}") # ---------------- 班级平均就业时长 ---------------- section("2.6.3 每个班级的平均就业时长") cad = api("/statistics/employment/class-avg-duration") check("班级数一致", len(cad) == len(classes), f"{len(cad)} / {len(classes)}") bad = [] for row in cad: mine = [e for e in emps if e["class_id"] == row["class_id"]] cls_students = [s for s in students if s["class_id"] == row["class_id"]] with_d = [e for e in mine if e["open_date"] and e["offer_date"]] exp_avg = round(sum((e["offer_date"] - e["open_date"]).days for e in with_d) / len(with_d), 1) if with_d else None probs = [] if row["student_count"] != len(cls_students): probs.append(f"人数 {row['student_count']}≠{len(cls_students)}") if row["offer_count"] != len(with_d): probs.append(f"offer数 {row['offer_count']}≠{len(with_d)}") if exp_avg is not None and (row["avg_duration_days"] is None or abs(row["avg_duration_days"] - exp_avg) > 0.06): probs.append(f"均时长 {row['avg_duration_days']}≠{exp_avg}") if probs: bad.append(f"{row['class_name']}: " + ";".join(probs)) check("每班 人数/offer数/平均时长 全部一致", not bad, f"问题 {len(bad)} 个班;例:{bad[:1]}") # ---------------- 成绩波动 ---------------- section("2.7.2 成绩波动 Top N") mult = {sid: vals for sid, vals in per_stu.items() if len(vals) >= 2} exp_std = {sid: statistics.pstdev(vals) for sid, vals in mult.items()} # 总体标准差 exp_rng = {sid: max(vals) - min(vals) for sid, vals in mult.items()} vol = api("/statistics/scores/volatility?top_n=5&metric=stddev") order = sorted(exp_std, key=lambda sid: (-exp_std[sid], sid))[:5] check("标准差 Top5 人数正确", len(vol) == 5, f"{len(vol)} 条") check("标准差 Top5 排序一致", [r["stu_id"] for r in vol] == order, f"接口 {[r['stu_id'] for r in vol]} vs 独立 {order}") bad = [f"{r['name']} {r['stddev']}≠{round(exp_std[r['stu_id']], 2)}" for r in vol if abs(r["stddev"] - exp_std[r["stu_id"]]) > 0.011] check("标准差数值与 Python pstdev 一致(总体标准差)", not bad, f"不符:{bad[:3]}") volr = api("/statistics/scores/volatility?top_n=5&metric=range") order_r = sorted(exp_rng, key=lambda sid: (-exp_rng[sid], sid))[:5] check("最大分差 Top5 排序一致", [r["stu_id"] for r in volr] == order_r, f"接口 {[r['stu_id'] for r in volr]} vs 独立 {order_r}") bad = [f"{r['name']} {r['score_range']}≠{round(exp_rng[r['stu_id']], 2)}" for r in volr if abs(r["score_range"] - exp_rng[r["stu_id"]]) > 0.011] check("最大分差数值一致", not bad, f"不符:{bad[:3]}") # 趋势方向:接口用「对考核序次做最小二乘拟合」的斜率,阈值 ±1 分/场 # 这里用另一条代数路径算斜率(nΣxy − ΣxΣy)/(nΣx² − (Σx)²),避开均值形式, # 这样才构成真正的独立对账,而不是把同一个公式再抄一遍。 def lsq_slope(vals: list[float]) -> float: n = len(vals) if n < 2: return 0.0 sx, sy = n * (n - 1) / 2, sum(vals) sxy = sum(i * v for i, v in enumerate(vals)) sxx = sum(i * i for i in range(n)) den = n * sxx - sx * sx return (n * sxy - sx * sy) / den if den else 0.0 # 关键不变量:出参里的 trend 必须与出参里的 trend_slope 自洽。 # 前端会把「标签 + 斜率」并排显示,出现「基本持平 +1.0/场」这种组合就是自相矛盾。 def label_from(s: float, n: int) -> str: if n < 2: return "数据不足" return "上升" if s >= 1.0 else ("下降" if s <= -1.0 else "基本持平") bad = [] allvol = api("/statistics/scores/volatility?top_n=50&metric=stddev") for r in allvol: vals = [x for _, x in sorted( (s["exam_seq"], s["score"]) for s in scores if s["stu_id"] == r["stu_id"])] exp_slope = round(lsq_slope(vals), 2) if abs(r["trend_slope"] - exp_slope) > 0.011: bad.append(f"{r['name']} 斜率 {r['trend_slope']} ≠ 独立算 {exp_slope}") elif r["trend"] != label_from(r["trend_slope"], len(vals)): bad.append(f"{r['name']} 标签 {r['trend']} 与自身斜率 {r['trend_slope']} 不自洽(应为 " f"{label_from(r['trend_slope'], len(vals))})") check("趋势斜率与独立最小二乘一致,且标签与斜率自洽", not bad, f"不符:{bad[:3]}") check("波动榜里若出现上升/下降,斜率方向必须与标签同号", all((r["trend"] == "上升") == (r["trend_slope"] > 0) or r["trend"] == "基本持平" for r in allvol)) # ---------------- 就业漏斗 ---------------- section("2.7.2 班级就业漏斗") funnel = api("/statistics/employment/funnel") check("班级数一致", len(funnel) == len(classes), f"{len(funnel)} / {len(classes)}") bad = [] for row in funnel: cls_students = [s for s in students if s["class_id"] == row["class_id"]] stu_ids = {s["id"] for s in cls_students} got_offer = [e for e in emps if e["stu_id"] in stu_ids and e["offer_date"]] with_sal = [e["salary"] for e in got_offer if e["salary"] is not None] exp_rate = round(len(got_offer) / len(cls_students) * 100, 1) if cls_students else 0.0 probs = [] if row["total"] != len(cls_students): probs.append(f"总数 {row['total']}≠{len(cls_students)}") if row["employed"] != len(got_offer): probs.append(f"就业 {row['employed']}≠{len(got_offer)}") if abs(row["employment_rate"] - exp_rate) > 0.06: probs.append(f"就业率 {row['employment_rate']}≠{exp_rate}") if with_sal and abs(row["avg_salary"] - round(sum(with_sal) / len(with_sal), 2)) > 0.02: probs.append(f"均薪 {row['avg_salary']}≠{round(sum(with_sal)/len(with_sal),2)}") if probs: bad.append(f"{row['class_name']}: " + ";".join(probs)) check("每班 总数/就业数/就业率/平均薪资 一致", not bad, f"问题 {len(bad)} 个班;例:{bad[:1]}") rates = [r["employment_rate"] for r in funnel] check("按就业率降序", rates == sorted(rates, reverse=True), f"{rates}") check("就业率都在 0~100", all(0 <= r <= 100 for r in rates)) # ---------------- 总览 ---------------- section("仪表盘总览") ovw = api("/statistics/overview") check("学生总数", ovw["student_total"] == len(students), f"{ovw['student_total']} / {len(students)}") check("班级总数", ovw["class_total"] == len(classes), f"{ovw['class_total']} / {len(classes)}") exp_emp_total = len([e for e in emps if e["offer_date"]]) check("就业总数(只算已拿 offer)", ovw["employment_total"] == exp_emp_total, f"{ovw['employment_total']} / {exp_emp_total}") exp_rate = round(exp_emp_total / len(students) * 100, 1) check("整体就业率", abs(ovw["employment_rate"] - exp_rate) < 0.06, f"{ovw['employment_rate']} / {exp_rate}") sal = [e["salary"] for e in emps if e["salary"] is not None] check("平均薪资", abs(ovw["avg_salary"] - round(sum(sal) / len(sal), 2)) < 0.02, f"{ovw['avg_salary']} / {round(sum(sal)/len(sal),2)}") check("成绩记录总数", ovw["score_record_total"] == len(scores), f"{ovw['score_record_total']} / {len(scores)}") exp_avg = round(sum(s["score"] for s in scores) / len(scores), 2) check("全部成绩平均分", abs(ovw["score_avg"] - exp_avg) < 0.011, f"{ovw['score_avg']} / {exp_avg}") exp_warn = len({s["stu_id"] for s in scores if s["score"] < 60}) check("红线预警学生数(至少一次低于红线)", ovw["warning_student_count"] == exp_warn, f"{ovw['warning_student_count']} / {exp_warn}") bad = [r["class_name"] for r in ovw["top_classes"] if r["employed"] > r["total"]] check("top_classes 就业数不超过班级总人数", not bad, f"异常:{bad}") check("student_by_status 合计 == 学生总数", sum(ovw["student_by_status"].values()) == len(students), f"{sum(ovw['student_by_status'].values())} / {len(students)}") # ---------------- 收尾 ---------------- print("\n" + "=" * 66) print(f"统计口径对账:通过 {len(PASS)} 项,失败 {len(FAIL)} 项") if FAIL: print("\n失败明细:") for i, f in enumerate(FAIL, 1): print(f" {i}. {f}") sys.exit(1 if FAIL else 0) if __name__ == "__main__": main()