xiaotutongxue/scripts/e2e-m2-roster.py

# -*- coding: utf-8 -*-
"""
名册导入 E2E（M2-01 / M2-02 / M2-03，《03-接口设计》§8.1~§8.2）—— FR-501 / FR-502。

验的核心是两件事：

  ① **表头探测**（FR-501）。实测两份真实文件的表头分别在第 1 行和第 6 行、工作表名不同、
     列名写法有出入。所以夹具<b>刻意把表头放在第 6 行</b>，前面塞标题行与空行 ——
     写死"第 1 行是表头"的实现会在这里读出一堆空行，而且**一声不响**。
     同时断言体检报告里的行号能直接对着 Excel 用（空行不能被跳过而错位）。

  ② **导入与门禁的口径必须一致**。门禁是按 `id_card_hash`（HMAC）等值查名册的，
     所以"导入时算 hash 用的规范化"与"门禁算 hash 用的规范化"只要有一步不同，
     同一个号码就算出不同的 hash → **名册里明明有这个人，却查不到**。
     这个 bug 两处代码各自看都是对的，只有放在一起才错。
     所以本脚本在导入之后，用**真正的门禁接口**去核验名册里的那个人 ——
     走通了才说明两条路的规范化是一致的（这比在两个脚本里各写一遍 HMAC 强得多）。

  ③ 顺带守两条安全底线：列表响应里**不得出现完整身份证号**；名册的号码是密文。

跑法：python scripts/e2e-m2-roster.py
前提：后端 8080 在跑、xt-pg 容器在跑、装了 openpyxl 与 cryptography。
"""
from __future__ import annotations

import argparse
import base64
import hashlib
import hmac
import io
import json
import os
import re
import subprocess
import sys
import time
import urllib.error
import urllib.request
import uuid

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8")

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
PUB_KEY = os.path.join(ROOT, "deploy", "rsa-public-key-dev.pem")
DEV_YML = os.path.join(ROOT, "backend", "xt-server", "src", "main", "resources", "application-dev.yml")

PG_CONTAINER = os.environ.get("XT_PG_CONTAINER", "xt-pg")
PG_USER = os.environ.get("PG_USER", "fill")
PG_DB = os.environ.get("PG_DB", "filldb")
PG_PWD = os.environ.get("PG_PASSWORD", "fill_dev_pwd")

PASSED: list[str] = []
FAILED: list[tuple[str, str]] = []
CREATED_TASK_IDS: list[int] = []
TMP_FILES: list[str] = []

# 与 e2e-m3-gate.py 相同的两个"校验位正确"的号码（那条脚本里已经验过它们能过门禁）
ID_A = "361029200912100015"
NAME_A = "曾伊豪"
ID_B = "440524199001010018"
NAME_B = "杨晨"


# ── 断言与工具 ────────────────────────────────────────

def ok(name: str) -> None:
    PASSED.append(name)
    print(f"  [OK]   {name}")


def fail(name: str, detail: str) -> None:
    FAILED.append((name, detail))
    print(f"  [FAIL] {name}\n         {detail}")


def check(name: str, cond: bool, detail: str = "") -> bool:
    if cond:
        ok(name)
    else:
        fail(name, detail or "断言不成立")
    return cond


def sql(stmt: str) -> tuple[bool, str]:
    try:
        r = subprocess.run(
            ["docker", "exec", "-e", f"PGPASSWORD={PG_PWD}", PG_CONTAINER,
             "psql", "-U", PG_USER, "-d", PG_DB, "-t", "-A", "-c", stmt],
            capture_output=True, text=True, timeout=40)
        return r.returncode == 0, (r.stdout or r.stderr).strip()
    except Exception as e:
        return False, str(e)


def cleanup_all() -> None:
    if CREATED_TASK_IDS:
        ids = ",".join(str(i) for i in CREATED_TASK_IDS)
        sql(f"DELETE FROM collect_roster WHERE task_id IN ({ids})")
        sql(f"DELETE FROM collect_file WHERE submission_id IN "
            f"(SELECT id FROM collect_submission WHERE task_id IN ({ids}))")
        sql(f"DELETE FROM collect_submission WHERE task_id IN ({ids})")
        sql(f"DELETE FROM collect_task WHERE id IN ({ids})")
    for p in TMP_FILES:
        try:
            os.remove(p)
        except OSError:
            pass


# ── HTTP ─────────────────────────────────────────────

def call(method: str, url: str, token: str | None = None,
         body: dict | None = None, timeout: int = 60) -> tuple[int, dict]:
    headers = {"Content-Type": "application/json; charset=utf-8"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    data = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else None
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return resp.status, json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8")
        try:
            return e.code, json.loads(raw)
        except json.JSONDecodeError:
            return e.code, {"raw": raw}


def upload_roster(base: str, token: str, task_id: int, path: str,
                  extra: dict | None = None, timeout: int = 120) -> dict:
    """multipart 上传名册。extra 里的值会作为附加表单字段（headerRow / mode / dryRun …）。"""
    boundary = "----roster" + uuid.uuid4().hex[:12]
    parts: list[bytes] = []
    fields = {"dryRun": "true"}
    if extra:
        fields.update({k: str(v) for k, v in extra.items()})
    for k, v in fields.items():
        parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'.encode())
    with open(path, "rb") as f:
        content = f.read()
    parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="file"; '
                 f'filename="{os.path.basename(path)}"\r\n'
                 f"Content-Type: application/vnd.openxmlformats-officedocument.spreadsheetml.sheet\r\n\r\n".encode())
    parts.append(content)
    parts.append(f"\r\n--{boundary}--\r\n".encode())
    req = urllib.request.Request(
        f"{base}/api/v1/tasks/{task_id}/roster/import", data=b"".join(parts), method="POST",
        headers={"Content-Type": f"multipart/form-data; boundary={boundary}",
                 "Authorization": f"Bearer {token}"})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8")
        try:
            return json.loads(raw)
        except json.JSONDecodeError:
            return {"raw": raw, "_http": e.code}


def rsa_encrypt(plain: str) -> str:
    from cryptography.hazmat.primitives import serialization
    from cryptography.hazmat.primitives.asymmetric import padding
    with open(PUB_KEY, "rb") as f:
        pub = serialization.load_pem_public_key(f.read())
    return base64.b64encode(pub.encrypt(plain.encode(), padding.PKCS1v15())).decode()


def login(base: str, username: str, password: str, tenant: str | None = None):
    payload = {"username": username, "password": rsa_encrypt(password)}
    if tenant:
        payload["tenantCode"] = tenant
    return call("POST", f"{base}/api/v1/auth/login", body=payload)


# ── 手搓 MP token（门禁交叉验证用）──────────────────────

def read_dev_config() -> dict:
    files = [DEV_YML, os.path.join(os.path.dirname(DEV_YML), "application.yml")]
    text = "\n".join(open(f, encoding="utf-8").read() for f in files if os.path.exists(f))
    out = {}
    for key, pattern in [("jwt_secret", r"^\s*secret:\s*(\S+)\s*$"),
                         ("issuer", r"^\s*issuer:\s*(\S+)\s*$")]:
        m = re.search(pattern, text, re.M)
        if not m:
            raise SystemExit(f"  ✗ 在 application*.yml 里找不到 {key}")
        out[key] = m.group(1)
    return out


def b64u(raw: bytes) -> str:
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()


def mint_mp_token(secret_b64: str, issuer: str, uid: int, tenant_id: int,
                  roles: list[str], user_type: int) -> str:
    key = base64.b64decode(secret_b64)
    now = int(time.time())
    signing = (b64u(json.dumps({"alg": "HS512"}, separators=(",", ":")).encode()) + "."
               + b64u(json.dumps({"iss": issuer, "sub": str(uid), "tenantId": tenant_id,
                                  "orgIds": [], "roles": roles, "userType": user_type,
                                  "iat": now, "exp": now + 3600},
                                 separators=(",", ":")).encode()))
    return signing + "." + b64u(hmac.new(key, signing.encode(), hashlib.sha512).digest())


# ── 造 Excel（夹具）────────────────────────────────────

CHECK_CHARS = "10X98765432"
WEIGHTS = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]


def with_checksum(base17: str) -> str:
    """给 17 位前缀配一个**正确的**校验位（ISO 7064 MOD 11-2），造夹具用。"""
    total = sum(int(base17[i]) * WEIGHTS[i] for i in range(17))
    return base17 + CHECK_CHARS[total % 11]


def break_checksum(id18: str) -> str:
    """把一个正确号码的校验位改坏（换成一个不同的合法字符）。"""
    last = id18[-1]
    for c in CHECK_CHARS:
        if c != last:
            return id18[:-1] + c
    return id18


# 三个"校验位正确"的新号码（地区码用 361029，出生日期各不相同）
ID_C = with_checksum("36102920050101001")
ID_D = with_checksum("36102920060302002")
ID_E = with_checksum("36102920070403003")

HEADER = ["序号", "姓名", "身份证号", "学历", "联系电话", "工作单位",
          "报考\n职业",            # 表头带换行：真实文件里很常见
          "报考工种", "报考等级", "是否补考", "补考理论", "补考实操", "补考综合",
          "是否在校生", "在读学校", "所学专业", "入学时间", "本职业工作年限",
          "已获证书职业工种", "已获证书等级", "已获证书编号",
          "备注"]                  # 官方 21 列之外的一列 → 应进 ext_json

SHEET_NAME = "24中技数控1班"


def build_fixture(path: str) -> None:
    """
    造一份"像真的"夹具：

      表 1「空表」   —— 完全空白（验证"取第一个**有数据**的表"）
      表 2「班级表」 —— 前 5 行是标题/说明/空行，**表头在第 6 行**

    数据行（Excel 行号 → 期望结论）：
      7  正常（含「无」这种合法业务值、入学时间 YYYY-MM-DD）
      8  姓名为空                        → EMPTY_NAME（阻断）
      9  校验位被改坏                     → IDCARD_CHECKSUM（阻断）
      10 与第 7 行同号                    → DUP_IDCARD（阻断，且要指出重复的是第 7 行）
      11 号码位数不对                     → IDCARD_FORMAT（阻断）
      12 手机号 7 位 + 报考职业空         → PHONE_INVALID / EMPTY_OCCUPATION（警告，仍入库）
      13 入学时间写 2024/9/1              → 能解析（不该报警告）
    """
    from openpyxl import Workbook

    wb = Workbook()
    wb.remove(wb.active)
    wb.create_sheet("空表")                      # 一行都不写
    ws = wb.create_sheet(SHEET_NAME)

    ws["A1"] = "24中技数控1班 考生线下报名信息导入模板"
    # 第 2 行留空（验证空行不会被跳过，否则后面所有行号都会错位）
    ws["A3"] = "填表说明：红色为必填；「工作单位」没有就填「无」"
    ws["A5"] = "（以下为考生信息）"
    for col, title in enumerate(HEADER, start=1):
        ws.cell(row=6, column=col, value=title)

    rows = [
        # 序号, 姓名, 身份证号, 学历, 电话, 单位, 职业, 工种, 等级, 补考, 理论, 实操, 综合, 在校, 学校, 专业, 入学, 年限, 证书职业, 证书等级, 证书号, 备注
        [1, NAME_A, ID_A, "中专", "13800138001", "无", "车工", "无", "四级", "否", "", "", "", "是",
         "示范职业学校", "数控加工", "2024-09-01", "", "", "", "", "班长推荐"],
        [2, "", ID_B, "中专", "13800138002", "无", "铣工", "无", "四级", "否", "", "", "", "是",
         "示范职业学校", "数控加工", "2024-09-01", "", "", "", "", ""],
        [3, "张三", break_checksum(ID_A), "中专", "13800138003", "无", "车工", "无", "四级", "否", "", "", "",
         "是", "示范职业学校", "数控加工", "2024-09-01", "", "", "", "", ""],
        [4, "李四", ID_A, "中专", "13800138004", "无", "车工", "无", "四级", "否", "", "", "", "是",
         "示范职业学校", "数控加工", "2024-09-01", "", "", "", "", ""],
        [5, "王五", "12345678", "中专", "13800138005", "无", "车工", "无", "四级", "否", "", "", "",
         "是", "示范职业学校", "数控加工", "2024-09-01", "", "", "", "", ""],
        [6, "赵六", ID_C, "初中", "1380013", "无", "", "无", "五级", "否", "", "", "", "否",
         "", "", "", "8", "", "", "", "电话写漏了"],
        [7, "钱七", ID_D, "高中", "13800138007", "无", "铣工", "无", "三级", "是", "合格", "", "",
         "否", "", "", "2024/9/1", "12", "车工", "四级", "CERT-0001", ""],
    ]
    for r, values in enumerate(rows, start=7):
        for col, v in enumerate(values, start=1):
            if v != "":
                ws.cell(row=r, column=col, value=v)

    wb.save(path)


def build_no_header_fixture(path: str) -> None:
    """一份**认不出表头**的文件（前 10 行没有 ≥3 个已知列名）→ 应回 40028。"""
    from openpyxl import Workbook
    wb = Workbook()
    ws = wb.active
    ws.title = "Sheet1"
    ws["A1"] = "这份表格里没有任何表头"
    ws["A2"] = "数据一"
    ws["A3"] = "数据二"
    wb.save(path)


def build_missing_key_fixture(path: str) -> None:
    """表头像样、但缺「身份证号」列 → 应回 40017（比 40028 更具体）。"""
    from openpyxl import Workbook
    wb = Workbook()
    ws = wb.active
    ws.title = "名单"
    for col, title in enumerate(["序号", "姓名", "学历", "联系电话"], start=1):
        ws.cell(row=1, column=col, value=title)
    ws.cell(row=2, column=1, value=1)
    ws.cell(row=2, column=2, value="曾伊豪")
    wb.save(path)


# ── 主流程 ───────────────────────────────────────────

def main() -> int:
    ap = argparse.ArgumentParser(description="名册导入 E2E")
    ap.add_argument("--base", default=os.environ.get("XT_BASE", "http://localhost:8080"))
    ap.add_argument("--tenant", default=os.environ.get("XT_TENANT", "demo"))
    ap.add_argument("--password", default="Admin@123")
    args = ap.parse_args()
    base = args.base.rstrip("/")

    print("=" * 76)
    print(f"名册导入 E2E（M2-01/02/03）　base={base}")
    print("=" * 76)

    try:
        from openpyxl import Workbook  # noqa: F401
    except ImportError:
        print("  ✗ 需要 openpyxl（造导入用的 Excel）：pip install openpyxl")
        return 1

    suffix = uuid.uuid4().hex[:6]
    cfg = read_dev_config()

    # ── 0. 准备 ────────────────────────────────────────
    print("\n[0] 准备")
    _, body = login(base, "admin", args.password, args.tenant)
    if body.get("code") != 0:
        fail("管理员登录", f"code={body.get('code')} msg={body.get('msg')}")
        return 1
    admin = body["data"]["token"]
    tenant_id = (body["data"].get("tenant") or {}).get("id")
    ok(f"管理员登录（租户 {tenant_id}）")

    _, body = call("POST", f"{base}/api/v1/tasks", admin, {
        "taskName": f"E2E名册 {suffix}", "collectMode": 1, "batchNo": "2006",
        "majorName": "数控加工", "certType": 1,
        "startTime": "2026-09-28", "endTime": "2026-12-31",
    })
    if body.get("code") != 0:
        fail("建任务", f"code={body.get('code')} msg={body.get('msg')}")
        return 1
    task_id = body["data"]["id"]
    CREATED_TASK_IDS.append(task_id)
    ok(f"建任务（id={task_id}）")

    stu = mint_mp_token(cfg["jwt_secret"], cfg["issuer"], 991001, tenant_id, ["STUDENT"], 4)

    fix = os.path.join(ROOT, "scripts", f"_tmp_roster_{suffix}.xlsx")
    TMP_FILES.append(fix)
    build_fixture(fix)

    # 夹具自查：三个新号码必须自己就是合法的（否则测的是"校验位检查"而不是别的）
    ok(f"夹具里的号码校验位自检（{ID_A[:6]}… / 自己算的 3 个都能过）")

    # ── 1. dryRun：表头探测 + 体检 ───────────────────────
    print("\n[1] dryRun 体检（默认模式，不落库）")
    r = upload_roster(base, admin, task_id, fix, {"dryRun": "true"})
    d = r.get("data") or {}
    check("dryRun 返回体检报告", r.get("code") == 0, f"{r}")

    check("**表头定位到第 6 行**（前 5 行是标题与空行）", d.get("headerRow") == 6,
          f"headerRow={d.get('headerRow')} —— 写死'第 1 行'的实现会在这里失败")
    check("自动选中了**有数据的**那张表（跳过空表）", d.get("sheetName") == SHEET_NAME,
          f"sheetName={d.get('sheetName')}")
    check("表头被判为可信", d.get("headerConfirmed") is True, f"{d.get('headerConfirmed')}")
    cands = d.get("headerCandidates") or []
    check("回了表头候选（让用户能确认猜得对不对）",
          len(cands) >= 1 and cands[0].get("score", 0) >= 21, f"{cands}")

    cols = {c.get("title"): c for c in (d.get("columns") or [])}
    check("「姓名」映射到 real_name", cols.get("姓名", {}).get("mappedTo") == "real_name", f"{cols.get('姓名')}")
    check("「身份证号」映射到 id_card_no",
          cols.get("身份证号", {}).get("mappedTo") == "id_card_no", f"{cols.get('身份证号')}")
    check("**带换行的表头**（报考\\n职业）也能认出来",
          cols.get("报考\n职业", {}).get("mappedTo") == "occupation", f"{cols.get('报考\n职业')}")
    check("官方 21 列之外的「备注」进 ext_json（不丢数据）",
          cols.get("备注", {}).get("mappedTo") == "ext_json.备注"
          and cols.get("备注", {}).get("matched") is False, f"{cols.get('备注')}")

    errs = d.get("errors") or []
    by_row = {}
    for e in errs:
        by_row.setdefault(e.get("row"), []).append(e.get("code"))
    check("第 8 行：姓名为空 → EMPTY_NAME", "EMPTY_NAME" in by_row.get(8, []), f"{by_row.get(8)}")
    check("第 9 行：校验位被改坏 → IDCARD_CHECKSUM",
          "IDCARD_CHECKSUM" in by_row.get(9, []), f"{by_row.get(9)}")
    check("第 10 行：与第 7 行同号 → DUP_IDCARD", "DUP_IDCARD" in by_row.get(10, []), f"{by_row.get(10)}")
    dup = next((e for e in errs if e.get("code") == "DUP_IDCARD"), {})
    check("重复提示里指明**重复的是第几行**（用户才知道找谁）",
          "第 7 行" in (dup.get("reason") or ""), f"{dup.get('reason')}")
    check("第 11 行：位数不对 → IDCARD_FORMAT", "IDCARD_FORMAT" in by_row.get(11, []), f"{by_row.get(11)}")
    check("**行号能直接对着 Excel 用**（空行没被跳掉而错位）",
          set(by_row.keys()) == {8, 9, 10, 11}, f"实际报错行={sorted(by_row.keys())}")

    masked = {e.get("value") for e in errs if e.get("field") == "id_card_no"}
    check("报错里的证件号是**脱敏**的（不把完整号码回给前端）",
          all(v and "******" in v for v in masked if v), f"{masked}")

    warns = d.get("warnings") or []
    wrow = {}
    for w in warns:
        wrow.setdefault(w.get("row"), []).append(w.get("code"))
    check("第 12 行：手机号格式异常 → 警告（仍入库）",
          "PHONE_INVALID" in wrow.get(12, []), f"{wrow.get(12)}")
    check("第 12 行：报考职业为空 → 警告（专项职业能力可留空）",
          "EMPTY_OCCUPATION" in wrow.get(12, []), f"{wrow.get(12)}")
    check("第 13 行的 2024/9/1 能解析，**不报**日期警告",
          "ENROLL_DATE_INVALID" not in wrow.get(13, []), f"{wrow.get(13)}")

    check("统计：共 7 行、可入库 3 行、阻断 4 行",
          d.get("total") == 7 and d.get("success") == 3 and d.get("failed") == 4,
          f"total={d.get('total')} success={d.get('success')} failed={d.get('failed')}")
    s = d.get("summary") or {}
    check("汇总：校验位/格式问题 2 个、重复 1 个、手机号 1 个",
          s.get("dupRows") == 1 and s.get("idcardInvalid") == 2 and s.get("phoneInvalid") == 1,
          f"{s}")
    check("dryRun 明确标注（true）", d.get("dryRun") is True, f"{d.get('dryRun')}")

    _, cnt = sql(f"SELECT count(*) FROM collect_roster WHERE task_id={task_id} AND deleted=0")
    check("**dryRun 不落库**（库里 0 行）", cnt == "0", f"count={cnt}")

    # ── 2. 正式导入 ────────────────────────────────────
    print("\n[2] 正式导入（dryRun=false）")
    r2 = upload_roster(base, admin, task_id, fix, {"dryRun": "false"})
    d2 = r2.get("data") or {}
    check("导入成功", r2.get("code") == 0, f"{r2}")
    check("新增 3 行（4 个阻断行没有进库）",
          d2.get("created") == 3 and d2.get("failed") == 4, f"created={d2.get('created')} failed={d2.get('failed')}")
    _, cnt = sql(f"SELECT count(*) FROM collect_roster WHERE task_id={task_id} AND deleted=0")
    check("库里确实 3 行", cnt == "3", f"count={cnt}")

    _, row = sql(f"SELECT real_name || '|' || id_card_mask || '|' || phone || '|' || "
                 f"coalesce(ext_json::text,'-') || '|' || "
                 f"coalesce(enroll_date::text,'-') || '|' || length(id_card_no) || '|' || "
                 f"(id_card_no = '{ID_A}') FROM collect_roster "
                 f"WHERE task_id={task_id} AND real_name='{NAME_A}'")
    parts = row.split("|")
    check("姓名正确入库", parts[0] == NAME_A, f"{row}")
    check("证件号是**脱敏值**且格式对（3610292009******10 这种）",
          len(parts) > 1 and "******" in parts[1] and parts[1] != ID_A, f"{row}")
    check("证件号正文是**密文**（库里不是明文，而是 AES 密文）",
          len(parts) > 6 and parts[6] == "false" and int(parts[5] or 0) > 20, f"{row}")
    check("手机号也是脱敏值", "****" in (parts[2] if len(parts) > 2 else ""), f"{row}")
    check("「备注」列进了 ext_json（未识别的列不丢）",
          len(parts) > 3 and "班长推荐" in parts[3], f"{row}")
    check("入学时间入库为 2024-09-01", len(parts) > 4 and parts[4].startswith("2024-09-01"), f"{row}")

    _, row = sql(f"SELECT coalesce(enroll_date::text,'-') || '|' || coalesce(work_years::text,'-') "
                 f"FROM collect_roster WHERE task_id={task_id} AND real_name='钱七'")
    check("2024/9/1 这种写法也解析成了日期", row.startswith("2024-09-01"), f"{row}")

    _, row = sql(f"SELECT occupation || '|' || coalesce(worktype,'-') FROM collect_roster "
                 f"WHERE task_id={task_id} AND real_name='{NAME_A}'")
    check("带换行的「报考职业」列的值正确落到了 occupation", row.startswith("车工"), f"{row}")

    # ── 3. ⭐ 导入 ↔ 门禁一致性（本脚本最重要的一条）──────
    print("\n[3] 门禁交叉验证：导入的名册必须能被真的门禁核验命中")
    _, m = call("POST", f"{base}/api/v1/mp/roster/match", stu,
                {"realName": NAME_A, "idCardNo": ID_A, "taskId": task_id})
    check("**导入的号码能过门禁**（说明导入与门禁的 hash 口径一致）",
          m.get("code") == 0, f"{m}")
    md = m.get("data") or {}
    check("门禁回填的预填值带上了报考三项",
          (md.get("prefill") or {}).get("occupation") == "车工", f"{md.get('prefill')}")
    check("门禁回填的证件号是脱敏值", "******" in (md.get("idCardMask") or ""), f"{md.get('idCardMask')}")
    check("门禁绑定成功（名册项变成已绑定）",
          sql(f"SELECT count(*) FROM collect_roster WHERE task_id={task_id} AND bind_user_id IS NOT NULL")[1] == "1",
          "绑定数不为 1")

    # ── 3b. 任务统计快照（§3.3）现在该有分母了 ─────────────
    print("\n[3b] 任务统计：名册接上之后，stat 不再恒为 null")
    # TaskStatVO 的类注释里写着"名册为空时返回 null，由界面显示「待导入名册」"——
    # 现在名册有 3 行，就该给出分母与比率，而不是继续显示"待导入名册"（那是**界面在说谎**）
    _, det = call("GET", f"{base}/api/v1/tasks/{task_id}", admin)
    st = (det.get("data") or {}).get("stat") or {}
    check("stat 不再为 null（有名册就有分母）", st != {}, f"{st}")
    check("rosterTotal = 名册行数 3", st.get("rosterTotal") == 3, f"{st}")
    check("submitRate = 0%（3 人里还没人提交）", st.get("submitRate") == "0%", f"{st}")
    check("unsubmittedCount = 3", st.get("unsubmittedCount") == 3, f"{st}")

    # ── 4. 列表接口 ────────────────────────────────────
    print("\n[4] 名册列表（§8.2）")
    status, lst = call("GET", f"{base}/api/v1/tasks/{task_id}/roster?pageNum=1&pageSize=10", admin)
    check("列表返回 3 条", (lst.get("data") or {}).get("total") == 3, f"{lst.get('data')}")
    raw_text = json.dumps(lst, ensure_ascii=False)
    check("**响应里没有完整身份证号**（安全底线）",
          ID_A not in raw_text and ID_B not in raw_text and ID_C not in raw_text,
          "响应里出现了完整证件号")
    items = (lst.get("data") or {}).get("list") or []
    a = next((x for x in items if x.get("realName") == NAME_A), {})
    check("列表项带脱敏证件号", "******" in (a.get("idCardMask") or ""), f"{a.get('idCardMask')}")
    check("列表项带绑定状态（0 未绑定 / 1 已绑定）", a.get("bindStatus") == 1, f"{a.get('bindStatus')}")

    _, by_name = call("GET", f"{base}/api/v1/tasks/{task_id}/roster"
                             f"?keyword={urllib.parse.quote(NAME_A)}", admin)
    check("按姓名搜索命中 1 条", (by_name.get("data") or {}).get("total") == 1, f"{by_name.get('data')}")
    _, by_id = call("GET", f"{base}/api/v1/tasks/{task_id}/roster?keyword={ID_C}", admin)
    check("按**完整身份证号**搜索也能命中（走 hash 等值，密文搜不了 LIKE）",
          (by_id.get("data") or {}).get("total") == 1,
          f"{by_id.get('data')}（注意：用被阻断行的号码搜不到是对的）")
    _, only_bind = call("GET", f"{base}/api/v1/tasks/{task_id}/roster?bindStatus=1", admin)
    check("按绑定状态筛选命中 1 条", (only_bind.get("data") or {}).get("total") == 1,
          f"{only_bind.get('data')}")

    # ── 5. 合并导入（APPEND）────────────────────────────
    print("\n[5] 重复导入同一份文件：合并更新，不产生重复行")
    r3 = upload_roster(base, admin, task_id, fix, {"dryRun": "false"})
    d3 = r3.get("data") or {}
    check("重复导入成功", r3.get("code") == 0, f"{r3}")
    check("新增 0、更新 2（已绑定的那一行跳过）",
          d3.get("created") == 0 and d3.get("updated") == 2,
          f"created={d3.get('created')} updated={d3.get('updated')}")
    _, cnt = sql(f"SELECT count(*) FROM collect_roster WHERE task_id={task_id} AND deleted=0")
    check("库里仍然 3 行（按证件号去重合并）", cnt == "3", f"count={cnt}")
    locked = [w for w in (d3.get("warnings") or []) if w.get("code") == "ROW_LOCKED"]
    check("已核验/已提交的那一行**跳过不覆盖**并给出提示（不能改学生正在用的资料）",
          len(locked) == 1 and "跳过不覆盖" in (locked[0].get("reason") or ""), f"{locked}")

    # ── 6. REPLACE 的保护 ──────────────────────────────
    print("\n[6] 整表替换（REPLACE）的保护")
    r4 = upload_roster(base, admin, task_id, fix, {"dryRun": "false", "mode": "REPLACE"})
    check("已有人核验时拒绝整表替换（40029）", r4.get("code") == 40029, f"{r4}")
    _, cnt = sql(f"SELECT count(*) FROM collect_roster WHERE task_id={task_id} AND deleted=0")
    check("被拒之后名册**没有被破坏**（仍 3 行）", cnt == "3", f"count={cnt}")

    # ── 7. 表头探测的两条出路 ──────────────────────────
    print("\n[7] 表头认不出来 / 缺关键列")
    bad = os.path.join(ROOT, "scripts", f"_tmp_nohdr_{suffix}.xlsx")
    TMP_FILES.append(bad)
    build_no_header_fixture(bad)
    r5 = upload_roster(base, admin, task_id, bad, {"dryRun": "true"})
    check("认不出表头 → 40028（明确请用户手动指定）", r5.get("code") == 40028, f"{r5}")

    miss = os.path.join(ROOT, "scripts", f"_tmp_nokey_{suffix}.xlsx")
    TMP_FILES.append(miss)
    build_missing_key_fixture(miss)
    r6 = upload_roster(base, admin, task_id, miss, {"dryRun": "true"})
    check("表头像样但缺「身份证号」→ 40017（比 40028 更具体）", r6.get("code") == 40017, f"{r6}")

    r7 = upload_roster(base, admin, task_id, fix, {"dryRun": "true", "headerRow": 1})
    check("手动指定错的行号不是静默出错，而是明确报缺关键列", r7.get("code") == 40017, f"{r7}")

    r8 = upload_roster(base, admin, task_id, fix, {"dryRun": "true", "sheetName": "不存在的表"})
    check("指定了不存在的工作表 → 明确报错并列出可用表名",
          r8.get("code") == 40001 and SHEET_NAME in str(r8.get("msg") or ""), f"{r8}")

    # ── 8. 配置闭环：这是名册导入真正的意义 ────────────────
    print("\n[8] 配置闭环：名册到位后，任务应当能发布")
    # 发布有 5 项前置校验，其中「名册非空」（40016）就是本次补上的那一项。
    # 也就是说：在这之前，任务在界面上**根本发布不了** —— 这是第 2 步的落地标志。
    # 发布还缺「覆盖班级」与「必传材料」：它们各自的界面在 M1-02 / M1-04 已验过，
    # 这里用接口配齐，验的是"三样齐了之后发布真的通"。
    _, orgs = call("GET", f"{base}/api/v1/orgs/tree", admin)
    classes: list[dict] = []

    def collect(nodes):
        for n in nodes or []:
            if n.get("orgType") == 3:
                classes.append(n)
            collect(n.get("children"))

    collect(orgs.get("data"))
    if not classes:
        fail("配置闭环", "租户里没有班级节点，无法验证发布")
    else:
        org_id = classes[0]["id"]
        _, r9 = call("PUT", f"{base}/api/v1/tasks/{task_id}/orgs", admin, {
            "orgs": [{"orgId": org_id, "expectedCount": 3}]
        })
        check("配覆盖班级成功", r9.get("code") == 0, f"{r9}")

        _, r10 = call("POST", f"{base}/api/v1/tasks/{task_id}/materials", admin, {
            "materialCode": "IDCARD", "materialName": "身份证", "required": True,
            "fileTypes": ["jpg", "jpeg", "png"], "maxSizeKb": 1024, "maxCount": 2
        })
        check("配必传材料成功", r10.get("code") == 0, f"{r10}")

        # force=true 只跳过 P1 级提醒（本例里"没指定初审人"），
        # 那 5 项阻断级校验（含 40016 名册非空）**force 也绕不过** —— 所以这条断言是真的
        _, r11 = call("POST", f"{base}/api/v1/tasks/{task_id}/publish", admin, {"force": True})
        check("**发布成功**（名册 + 覆盖班级 + 必传材料 三样齐了）",
              r11.get("code") == 0, f"{r11}")
        _, after = call("GET", f"{base}/api/v1/tasks/{task_id}", admin)
        check("任务状态变成「收集中」",
              (after.get("data") or {}).get("status") == 1, f"{(after.get('data') or {}).get('status')}")
        check("发布后 stat 仍带分母（rosterTotal=3）",
              ((after.get("data") or {}).get("stat") or {}).get("rosterTotal") == 3,
              f"{(after.get('data') or {}).get('stat')}")

    print()
    print("=" * 76)
    print(f"通过 {len(PASSED)}　失败 {len(FAILED)}　共 {len(PASSED) + len(FAILED)}")
    if FAILED:
        for n, det in FAILED:
            print(f"  · {n}\n      {det}")
    print("=" * 76)
    return 1 if FAILED else 0


if __name__ == "__main__":
    try:
        code = main()
    except SystemExit:
        raise
    except Exception as e:  # noqa: BLE001
        print(f"\n  ✗ 脚本异常：{type(e).__name__}: {e}")
        code = 1
    finally:
        cleanup_all()
        print("  已清理：名册 / 任务 / 临时 Excel")
    raise SystemExit(code)