first commit

app/excel_parse.py

@@ -0,0 +1,253 @@
+from __future__ import annotations
+
+import re
+from io import BytesIO
+from pathlib import Path
+
+from openpyxl import load_workbook
+
+
+META_ALIASES = {
+    "sid": ("学号", "考号", "id", "学生id"),
+    "name": ("姓名", "学生姓名", "名字"),
+    "class_name": ("班级", "班", "行政班"),
+    "track": ("方向", "科类", "首选", "首选科目"),
+    "subjects": ("选科", "再选", "科目组合"),
+    "status": ("学籍", "学籍类型", "身份"),
+    "move": ("异动", "备注", "说明"),
+}
+
+EXAM_HINT = re.compile(
+    r"(总分|期末|联考|月考|模考|开学|学考|诊断|调研|高考)",
+    re.I,
+)
+SKIP_SHEETS = ("说明", "readme", "目录")
+ART_RE = re.compile(r"艺术|体育|艺体")
+LEAVE_RE = re.compile(r"请假|出国|休学|国际|不在校|港澳|港籍.*假")
+
+
+def _norm(v) -> str:
+    if v is None:
+        return ""
+    if isinstance(v, float) and v.is_integer():
+        return str(int(v))
+    return str(v).strip()
+
+
+def _is_exam_header(text: str) -> bool:
+    t = text.strip()
+    if not t or t in {"考试次数"}:
+        return False
+    if t.endswith("-总分") or t.endswith("总分"):
+        return True
+    return bool(EXAM_HINT.search(t)) and "次数" not in t
+
+
+def _map_meta(header: str) -> str | None:
+    h = header.strip().lower()
+    for key, aliases in META_ALIASES.items():
+        for a in aliases:
+            if h == a.lower() or h.replace(" ", "") == a.lower():
+                return key
+    return None
+
+
+def _sheet_kind(name: str) -> str:
+    n = name.lower()
+    if any(x in n for x in ("艺体", "艺术", "体育")):
+        return "art"
+    if "异动" in n:
+        return "leave"
+    if "历史" in n:
+        return "history"
+    if "物理" in n:
+        return "physics"
+    if n in ("全体", "全部", "汇总"):
+        return "all"
+    return "other"
+
+
+def parse_workbook(path: Path | BytesIO) -> dict:
+    wb = load_workbook(path, data_only=True, read_only=True)
+    sheets = []
+    students: list[dict] = []
+    exam_order: list[str] = []
+    warnings: list[str] = []
+
+    for ws in wb.worksheets:
+        kind = _sheet_kind(ws.title)
+        rows = ws.iter_rows(values_only=True)
+        try:
+            header_row = next(rows)
+        except StopIteration:
+            continue
+        headers = [_norm(h) for h in header_row]
+        if not any(headers):
+            continue
+        meta_idx = {}
+        exam_idx = {}
+        for i, h in enumerate(headers):
+            if not h:
+                continue
+            mk = _map_meta(h)
+            if mk:
+                meta_idx[mk] = i
+                continue
+            if _is_exam_header(h):
+                exam_idx[h] = i
+                if h not in exam_order:
+                    exam_order.append(h)
+        if "sid" not in meta_idx and "name" not in meta_idx:
+            warnings.append(f"工作表「{ws.title}」未识别到学号/姓名，已跳过")
+            continue
+
+        count = 0
+        for raw in rows:
+            if raw is None or all(v is None or str(v).strip() == "" for v in raw):
+                continue
+            rec = {
+                "sid": _norm(raw[meta_idx["sid"]]) if "sid" in meta_idx else "",
+                "name": _norm(raw[meta_idx["name"]]) if "name" in meta_idx else "",
+                "class_name": _norm(raw[meta_idx["class_name"]]) if "class_name" in meta_idx else "未分班",
+                "track": _norm(raw[meta_idx["track"]]) if "track" in meta_idx else "",
+                "subjects": _norm(raw[meta_idx["subjects"]]) if "subjects" in meta_idx else "",
+                "status": _norm(raw[meta_idx["status"]]) if "status" in meta_idx else "",
+                "move": _norm(raw[meta_idx["move"]]) if "move" in meta_idx else "",
+                "sheet": ws.title,
+                "sheet_kind": kind,
+                "scores": {},
+            }
+            if not rec["sid"] and not rec["name"]:
+                continue
+            if not rec["sid"]:
+                rec["sid"] = f"{ws.title}-{rec['name']}-{count}"
+            if kind == "physics" and not rec["track"]:
+                rec["track"] = "物理方向"
+            if kind == "history" and not rec["track"]:
+                rec["track"] = "历史方向"
+            if not rec["track"]:
+                if "物理" in rec["subjects"]:
+                    rec["track"] = "物理方向"
+                elif "历史" in rec["subjects"]:
+                    rec["track"] = "历史方向"
+            for exam, idx in exam_idx.items():
+                if idx >= len(raw):
+                    continue
+                val = raw[idx]
+                if val is None or str(val).strip() in ("", "缺考", "—", "-", "/"):
+                    rec["scores"][exam] = None
+                    continue
+                try:
+                    rec["scores"][exam] = float(val)
+                except (TypeError, ValueError):
+                    rec["scores"][exam] = None
+            rec["art"] = bool(ART_RE.search(rec["move"]) or ART_RE.search(ws.title))
+            rec["leave"] = bool(LEAVE_RE.search(rec["move"]))
+            students.append(rec)
+            count += 1
+        sheets.append(
+            {
+                "name": ws.title,
+                "kind": kind,
+                "rows": count,
+                "exams": list(exam_idx.keys()),
+            }
+        )
+
+    wb.close()
+
+    # Prefer specialized sheets over 全体 to avoid duplicates
+    has_split = any(s["kind"] in ("physics", "history") for s in sheets)
+    if has_split:
+        kept = []
+        skipped_all = 0
+        for s in students:
+            if s["sheet_kind"] == "all":
+                skipped_all += 1
+                continue
+            kept.append(s)
+        students = kept
+        if skipped_all:
+            warnings.append(f"已忽略「全体」表中的 {skipped_all} 行，避免与物理/历史表重复")
+
+    # de-duplicate by sid, keep first non-all
+    seen = {}
+    uniq = []
+    dups = 0
+    for s in students:
+        if s["sid"] in seen:
+            dups += 1
+            continue
+        seen[s["sid"]] = True
+        uniq.append(s)
+    students = uniq
+    if dups:
+        warnings.append(f"按学号去重 {dups} 条")
+
+    if not exam_order:
+        warnings.append("未识别到考试成绩列，请确认表头含「总分」或「期末/联考/月考」")
+
+    return {
+        "sheets": sheets,
+        "students": students,
+        "exams": exam_order,
+        "warnings": warnings,
+    }
+
+
+def summarize_parse(parsed: dict) -> dict:
+    students = parsed["students"]
+    exams = parsed["exams"]
+    tracks = {}
+    classes = {}
+    status = {}
+    art = leave = 0
+    exam_present = {e: 0 for e in exams}
+    for s in students:
+        tracks[s["track"] or "未分方向"] = tracks.get(s["track"] or "未分方向", 0) + 1
+        classes[s["class_name"]] = classes.get(s["class_name"], 0) + 1
+        status[s["status"] or "未填"] = status.get(s["status"] or "未填", 0) + 1
+        if s["art"]:
+            art += 1
+        if s["leave"]:
+            leave += 1
+        for e in exams:
+            if s["scores"].get(e) is not None:
+                exam_present[e] += 1
+    score_scale = []
+    for e in exams:
+        vals = [s["scores"][e] for s in students if s["scores"].get(e) is not None]
+        if not vals:
+            continue
+        score_scale.append(
+            {
+                "exam": e,
+                "n": len(vals),
+                "min": round(min(vals), 1),
+                "max": round(max(vals), 1),
+                "mean": round(sum(vals) / len(vals), 1),
+            }
+        )
+    return {
+        "n": len(students),
+        "n_class": len(classes),
+        "tracks": tracks,
+        "classes": dict(sorted(classes.items(), key=lambda kv: _class_key(kv[0]))),
+        "status": status,
+        "art": art,
+        "leave": leave,
+        "exam_present": exam_present,
+        "score_scale": score_scale,
+        "sheets": parsed["sheets"],
+        "exams": exams,
+        "warnings": parsed["warnings"],
+    }
+
+
+def _class_key(name: str):
+    m = re.search(r"\d+", str(name))
+    return (int(m.group()) if m else 999, str(name))
+
+
+def class_sort_key(name: str):
+    return _class_key(name)