360 lines
12 KiB
Python
360 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import re
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
|
|
from openpyxl import load_workbook
|
|
from openpyxl.utils import column_index_from_string
|
|
|
|
|
|
META_ALIASES = {
|
|
"sid": ("学号", "考号", "id", "学生id"),
|
|
"name": ("姓名", "学生姓名", "名字"),
|
|
"class_name": ("班级", "班", "行政班"),
|
|
"track": ("方向", "科类", "首选", "首选科目"),
|
|
"subjects": ("选科", "再选", "科目组合"),
|
|
"status": ("学籍", "学籍类型", "身份"),
|
|
"move": ("异动", "备注", "说明"),
|
|
}
|
|
|
|
EXAM_HINT = re.compile(
|
|
r"(总分|期中|期末|联考|月考|模考|开学|学考|诊断|调研|高考|分班合成)",
|
|
re.I,
|
|
)
|
|
SUBJECT_SUFFIX = re.compile(
|
|
r"-(语文|数学|英语|外语|物理|化学|生物|历史|地理|政治|思想政治)分$"
|
|
)
|
|
SKIP_SHEETS = ("说明", "readme", "目录")
|
|
|
|
|
|
def _is_skip_sheet(name: str) -> bool:
|
|
n = name.lower()
|
|
if any(x in n for x in SKIP_SHEETS):
|
|
return True
|
|
return "概率分析" in n or "指标数" in n
|
|
|
|
ART_RE = re.compile(r"艺术|体育|艺体")
|
|
LEAVE_RE = re.compile(r"请假|出国|休学|国际|不在校|港澳|港籍.*假")
|
|
CELL_REF_RE = re.compile(r"\$?([A-Za-z]+)\$?(\d+)")
|
|
MISSING_TOKENS = {"", "缺考", "—", "-", "/", "none", "nan"}
|
|
|
|
|
|
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 _safe_arith(expr: str) -> float:
|
|
tree = ast.parse(expr, mode="eval")
|
|
allowed = (
|
|
ast.Expression,
|
|
ast.BinOp,
|
|
ast.UnaryOp,
|
|
ast.Add,
|
|
ast.Sub,
|
|
ast.Mult,
|
|
ast.Div,
|
|
ast.USub,
|
|
ast.UAdd,
|
|
ast.Constant,
|
|
ast.Load,
|
|
ast.Mod,
|
|
)
|
|
for node in ast.walk(tree):
|
|
if not isinstance(node, allowed):
|
|
raise ValueError("unsupported formula")
|
|
return float(eval(compile(tree, "<score-formula>", "eval"), {"__builtins__": {}}, {}))
|
|
|
|
|
|
def _eval_simple_formula(formula: str, raw, row_number: int) -> float | None:
|
|
if not isinstance(formula, str) or not formula.startswith("="):
|
|
return None
|
|
expr = formula[1:].strip()
|
|
if re.search(r"[^0-9A-Za-z+\-*/().$,\s]", expr):
|
|
return None
|
|
|
|
def repl(match: re.Match) -> str:
|
|
col = column_index_from_string(match.group(1).upper()) - 1
|
|
src_row = int(match.group(2))
|
|
if src_row != row_number or col < 0 or col >= len(raw):
|
|
raise ValueError("formula refers outside the current row")
|
|
val = raw[col]
|
|
if isinstance(val, str) and val.startswith("="):
|
|
raise ValueError("nested formula")
|
|
if val is None or str(val).strip().lower() in MISSING_TOKENS:
|
|
raise ValueError("missing ref")
|
|
return str(float(val))
|
|
|
|
try:
|
|
filled = CELL_REF_RE.sub(repl, expr)
|
|
return _safe_arith(filled)
|
|
except (TypeError, ValueError, SyntaxError, ZeroDivisionError):
|
|
return None
|
|
|
|
|
|
def _to_score(val, raw, row_number: int):
|
|
if val is None:
|
|
return None
|
|
if isinstance(val, str):
|
|
text = val.strip()
|
|
if text.lower() in MISSING_TOKENS:
|
|
return None
|
|
if text.startswith("="):
|
|
return _eval_simple_formula(text, raw, row_number)
|
|
val = text
|
|
try:
|
|
return float(val)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _is_exam_header(text: str) -> bool:
|
|
t = text.strip()
|
|
if not t or t in {"考试次数"} or "次数" in t:
|
|
return False
|
|
# 总分、数学分、语文分、分班合成-数学分 等都视为成绩列。
|
|
if t.endswith("分"):
|
|
return True
|
|
return bool(EXAM_HINT.search(t))
|
|
|
|
|
|
def infer_score_kind(exam_names: list[str], source_name: str = "") -> str:
|
|
found: list[str] = []
|
|
for name in exam_names:
|
|
m = SUBJECT_SUFFIX.search(str(name).strip())
|
|
if m:
|
|
found.append(m.group(1))
|
|
elif str(name).endswith("总分"):
|
|
found.append("总分")
|
|
uniq = list(dict.fromkeys(found))
|
|
if len(uniq) == 1:
|
|
return uniq[0]
|
|
if uniq and "总分" not in uniq:
|
|
return uniq[0]
|
|
for subj in ("语文", "数学", "英语", "外语", "化学", "生物", "地理", "政治"):
|
|
if subj in source_name:
|
|
return subj
|
|
return "总分"
|
|
|
|
|
|
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=False, read_only=True)
|
|
sheets = []
|
|
students: list[dict] = []
|
|
exam_order: list[str] = []
|
|
warnings: list[str] = []
|
|
source_name = path.name if isinstance(path, Path) else ""
|
|
|
|
for ws in wb.worksheets:
|
|
kind = _sheet_kind(ws.title)
|
|
if _is_skip_sheet(ws.title):
|
|
continue
|
|
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 row_number, raw in enumerate(rows, start=2):
|
|
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
|
|
rec["scores"][exam] = _to_score(raw[idx], raw, row_number)
|
|
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("未识别到考试成绩列，请确认表头含「总分」或单科分（如「高一下期末-数学分」）")
|
|
else:
|
|
for exam in exam_order:
|
|
present = sum(1 for s in students if s["scores"].get(exam) is not None)
|
|
if present == 0:
|
|
warnings.append(f"考试列「{exam}」没有读到数值，若是未计算的公式请先在 Excel 中打开保存后再上传")
|
|
|
|
score_kind = infer_score_kind(exam_order, source_name)
|
|
return {
|
|
"sheets": sheets,
|
|
"students": students,
|
|
"exams": exam_order,
|
|
"score_kind": score_kind,
|
|
"warnings": warnings,
|
|
}
|
|
|
|
|
|
def summarize_parse(parsed: dict) -> dict:
|
|
students = parsed["students"]
|
|
exams = parsed["exams"]
|
|
tracks = {}
|
|
classes = {}
|
|
status = {}
|
|
art = leave = moves = 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
|
|
if (s.get("move") or "").strip() or s.get("sheet_kind") in ("art", "leave"):
|
|
moves += 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,
|
|
"moves": moves,
|
|
"exam_present": exam_present,
|
|
"score_scale": score_scale,
|
|
"sheets": parsed["sheets"],
|
|
"exams": exams,
|
|
"score_kind": parsed.get("score_kind") or infer_score_kind(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)
|