feat(submit): 提交 + 红黄绿校验（PRD §5.7 的 VAL 系列）

一次给全，不"改一条再拒一次"：学生一次可能同时缺 3 项材料、2 个字段没填，
逐条拒绝会让他改 5 次提交 5 次。所以 validate 收全部问题后一次性返回
{ruleCode, severity, title, detail}，ruleCode 用 PRD 的 VAL-xx 编号。

两段式提交：只有黄色警告时先返回 submitted=false 让学生确认，
再带 confirmWarnings=true 才真的提交。为什么不前端自己判断 ——
只有服务端的校验结果算数，两次请求之间学生可能又改坏了什么。

实现的规则：VAL-01 证件号 / VAL-05 材料完整性 / VAL-06 格式 /
VAL-07 体积 / VAL-08 清晰度 / VAL-09 重复文件 / VAL-13 报考三项 /
VAL-15 补考科目数按等级 / VAL-16 在校生三项 / VAL-17 工作年限 /
VAL-18 未规范化 / VAL-20 体积下限 / VAL-22 表单必填（**新增**：
PRD 的 VAL-01~21 是具体业务规则，没覆盖动态表单的必填字段，按系列顺延补的）。

═══ 本轮抓到的四个真问题（都记在代码注释里了）═══

① **校验器不能用列名当字段 key**。字段 key 是 snake_case
   （work_unit / job_level / enroll_date），不是列名；而且报考三项
   存在 ext_json 里、**不是独立列**（契约 §2.4）。写 camelCase 的表现是
   "明明填了却说没填"。更糟的是 VAL-16 读列、VAL-22 读表单装配值 ——
   **两条规则对同一份数据给出相反结论**，学生不知道该信哪一条。
   现在统一用 FillService.getForm 装配出来的那一份作为唯一取值口径。

② **VAL-22 没判可见性**。不判可见就会要求一个在校生填「补考理论」——
   而那个字段在他页面上根本不存在。求值口径已与前端 fill.vue 的 visible()
   逐字对齐（norm 归一 true/false↔1/0、未知 op 一律可见）。

③ **VAL-01 读错了列**：collect_roster.id_card_no 是 **AES 密文**
   （列注释写明的），明文只在导入与门禁时短暂存在。一开始拿它算校验位，
   于是每一份填报单都被判"身份证号未通过校验" —— 100% 阻断。
   这种"规则很对但读错列"的错，反而容易被当成"校验太严"而不是 bug。
   现在只确认名册里有证件号（用脱敏列），校验位交给导入与门禁。

④ **静态资源没有 Cache-Control/ETag**（新增 StaticCacheHeaderFilter）。
   没有它浏览器会走启发式缓存：旧 index.html → 旧 chunk →
   **学生发版后看到的是上个月的页面**，而开发机硬刷新一次就没了，
   极容易被当成"偶发"。现在 /assets/** 一年 immutable（文件名带内容哈希）、
   *.html no-cache（每次校验）、/api/** no-store。

排查过程里还确认了一件事：材料上传那轮的"浏览器加载旧 chunk"
不是构建问题 —— 从 8080 下载的 chunk 与磁盘文件 SHA-256 一致、里面确实有新代码，
是缓存头缺失导致的浏览器行为。

验收：e2e-m3-submit.py **30/30**（含"该拦的确实拦了"和"修好后确实放过"）。
回归：草稿 27 / 门禁 29 / 材料上传 29 / 材料目录 68 全绿。

scripts/e2e-m3-submit.py

@@ -0,0 +1,318 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+提交与校验 验收测试（PRD §5.7 的 VAL 系列 + §9.8 提交）。
+
+    [x] 缺材料 / 缺必填字段 → **阻断**，且**一次给全**（不是改一条再拒一次）
+    [x] VAL-16 在校生三项：缺一项就拦
+    [x] VAL-15 补考科目数按等级限制（四级最多 1 科）
+    [x] 只有黄色警告时：先返回 submitted=false 让学生确认，confirmWarnings=true 才真的提交
+    [x] 提交成功后状态 → 1（待初审），再提交被拒
+    [x] 红黄计数落库（班主任端要按它排序）
+
+⚠️ 这个 E2E 的重点不是"能不能提交"，而是**拦得对不对**。
+   一个校验器最危险的状态是"永远通过" —— 它看起来一切正常，
+   直到机构把整批材料退回来。所以下面每条规则都断言了"该拦的确实拦了"和"修好后确实放过"。
+
+跑法：python scripts/e2e-m3-submit.py
+"""
+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
+
+sys.stdout.reconfigure(encoding="utf-8")
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+SRC = os.path.join(ROOT, "backend", "xt-server", "src", "main", "resources")
+PG = os.environ.get("XT_PG_CONTAINER", "xt-pg")
+UID = 992002
+
+PASSED: list[str] = []
+FAILED: list[tuple[str, str]] = []
+
+
+def ok(n: str) -> None:
+    PASSED.append(n)
+    print(f"  [OK]   {n}")
+
+
+def fail(n: str, d: str) -> None:
+    FAILED.append((n, d))
+    print(f"  [FAIL] {n}\n         {d}")
+
+
+def check(n: str, cond: bool, d: str = "") -> bool:
+    if cond:
+        ok(n)
+    else:
+        fail(n, d or "断言不成立")
+    return cond
+
+
+def sql(q: str) -> str:
+    r = subprocess.run(
+        ["docker", "exec", "-e", "PGPASSWORD=fill_dev_pwd", PG, "psql", "-U", "fill",
+         "-d", "filldb", "-t", "-A", "-c", q], capture_output=True, text=True, timeout=40)
+    return (r.stdout or r.stderr).strip()
+
+
+def read_cfg() -> dict:
+    text = "\n".join(open(os.path.join(SRC, f), encoding="utf-8").read()
+                     for f in ("application-dev.yml", "application.yml")
+                     if os.path.exists(os.path.join(SRC, f)))
+    out = {}
+    for k, p in [("jwt", r"^\s*secret:\s*(\S+)\s*$"), ("issuer", r"^\s*issuer:\s*(\S+)\s*$")]:
+        m = re.search(p, text, re.M)
+        if not m:
+            raise SystemExit(f"找不到 {k}")
+        out[k] = m.group(1)
+    return out
+
+
+def b64u(b: bytes) -> str:
+    return base64.urlsafe_b64encode(b).rstrip(b"=").decode()
+
+
+def mint(secret: str, issuer: str, uid: int, tenant: int) -> str:
+    key = base64.b64decode(secret)
+    now = int(time.time())
+    signing = (b64u(json.dumps({"alg": "HS512"}, separators=(",", ":")).encode()) + "."
+               + b64u(json.dumps({"iss": issuer, "sub": str(uid), "tenantId": tenant,
+                                  "orgIds": [], "roles": ["STUDENT"], "userType": 4,
+                                  "iat": now, "exp": now + 3600}, separators=(",", ":")).encode()))
+    return signing + "." + b64u(hmac.new(key, signing.encode(), hashlib.sha512).digest())
+
+
+def call(method: str, path: str, token: str | None = None, body: dict | None = None) -> dict:
+    h = {"Content-Type": "application/json; charset=utf-8"}
+    if token:
+        h["Authorization"] = f"Bearer {token}"
+    data = json.dumps(body, ensure_ascii=False).encode() if body is not None else None
+    req = urllib.request.Request("http://localhost:8080" + path, data=data, headers=h, method=method)
+    try:
+        with urllib.request.urlopen(req, timeout=60) as x:
+            return json.loads(x.read().decode("utf-8"))
+    except urllib.error.HTTPError as e:
+        try:
+            return json.loads(e.read().decode("utf-8"))
+        except Exception:
+            return {"_http": e.code}
+
+
+def upload(token: str, sid: int, code: str, filename: str, data: bytes) -> dict:
+    boundary = "----xte2e" + uuid.uuid4().hex[:12]
+    parts = []
+    for k, v in [("submissionId", str(sid)), ("materialCode", code)]:
+        parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'.encode())
+    parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{filename}"\r\n'
+                 f"Content-Type: image/jpeg\r\n\r\n".encode())
+    parts.append(data)
+    parts.append(f"\r\n--{boundary}--\r\n".encode())
+    req = urllib.request.Request(
+        "http://localhost:8080/api/v1/mp/files/upload", 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=90) as x:
+            return json.loads(x.read().decode("utf-8"))
+    except urllib.error.HTTPError as e:
+        try:
+            return json.loads(e.read().decode("utf-8"))
+        except Exception:
+            return {"_http": e.code}
+
+
+def small_jpeg() -> bytes:
+    from PIL import Image
+    buf = io.BytesIO()
+    Image.new("RGB", (900, 1200), (200, 210, 220)).save(buf, "JPEG", quality=90)
+    return buf.getvalue()
+
+
+def rules(r: dict) -> set[str]:
+    return {i.get("ruleCode") for i in ((r.get("data") or {}).get("issues") or [])}
+
+
+def sub(token: str, sid: int, confirm: bool = False) -> dict:
+    return call("POST", f"/api/v1/mp/submissions/{sid}/submit", token, {"confirmWarnings": confirm})
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--task", type=int, default=17)
+    ap.add_argument("--tenant", type=int, default=1)
+    args = ap.parse_args()
+    task_id, tenant = args.task, args.tenant
+    base = "/api/v1/mp"
+
+    print("=" * 72)
+    print(f"提交与校验 E2E　task={task_id}")
+    print("=" * 72)
+
+    cfg = read_cfg()
+    tok = mint(cfg["jwt"], cfg["issuer"], UID, tenant)
+
+    rid = sql(f"SELECT id FROM collect_roster WHERE task_id={task_id} LIMIT 1")
+    if not rid.isdigit():
+        fail("取名册行", "任务下没有名册")
+        return 1
+    sql(f"UPDATE collect_roster SET bind_user_id={UID}, bind_type=1 WHERE id={rid}")
+
+    r = call("POST", f"{base}/submissions", tok, {"taskId": task_id})
+    if r.get("code") != 0:
+        fail("建草稿", f"{r}")
+        return 1
+    sid = r["data"]["id"]
+    print(f"  submission={sid}  roster={rid}")
+
+    try:
+        # ── 1. 什么都没填就提交：必须被拦，且一次给全 ──
+        print("\n[1] 空着提交 → 必须拦住")
+        r = sub(tok, sid)
+        check("HTTP 200 + code=0（校验不通过是业务结果，不是异常）",
+              r.get("code") == 0, f"code={r.get('code')} msg={r.get('msg')}")
+        d = r.get("data") or {}
+        check("submitted=false", d.get("submitted") is False, f"{d.get('submitted')}")
+        check("redCount > 0", (d.get("redCount") or 0) > 0, f"red={d.get('redCount')}")
+        check("**一次给全**：问题数 ≥ 3（不是只报第一条）",
+              len(d.get("issues") or []) >= 3, f"{len(d.get('issues') or [])} 条")
+        rs = rules(r)
+        check("VAL-05 缺必传材料（身份证）", "VAL-05" in rs, f"实际规则: {sorted(rs)}")
+        check("VAL-22 必填字段缺失", "VAL-22" in rs, f"实际规则: {sorted(rs)}")
+        check("VAL-01 **不该误报**（id_card_no 存的是 AES 密文，不能当明文算校验位 —— 实测错过一次，表现是 100% 阻断）",
+              "VAL-01" not in rs, f"实际规则: {sorted(rs)}")
+        check("状态没被改动（仍是草稿）", d.get("status") == 0, f"status={d.get('status')}")
+        db = sql(f"SELECT status || '|' || coalesce(submit_time::text,'-') FROM collect_submission WHERE id={sid}")
+        check("库里也没被改成已提交", db.split("|")[0] == "0", f"{db}")
+
+        # ── 2. VAL-16 在校生三项 ──
+        print("\n[2] 在校生三项（VAL-16）")
+        fill = {"work_unit": "无", "is_retake": False, "is_student": True,
+                "work_years": "3", "cert_occupation": "", "cert_level": "", "cert_no": ""}
+        call("PUT", f"{base}/submissions/{sid}", tok, {"formData": fill})
+        r = sub(tok, sid)
+        rs = rules(r)
+        check("isStudent=是 但三项都空 → VAL-16", "VAL-16" in rs, f"实际: {sorted(rs)}")
+        # 只补一项，还应该拦
+        call("PUT", f"{base}/submissions/{sid}", tok, {"formData": {"school": "示范职业学校"}})
+        r = sub(tok, sid)
+        check("只补了 1 项仍然拦（缺项要列全）", "VAL-16" in rules(r), f"实际: {sorted(rules(r))}")
+        call("PUT", f"{base}/submissions/{sid}", tok, {"formData": {
+            "school": "示范职业学校", "major": "数控技术", "enroll_date": "2023-09-01"}})
+        r = sub(tok, sid)
+        check("三项补齐 → VAL-16 消失", "VAL-16" not in rules(r), f"实际: {sorted(rules(r))}")
+
+        # ── 3. VAL-15 补考科目数 ──
+        print("\n[3] 补考科目数按等级（VAL-15）")
+        call("PUT", f"{base}/submissions/{sid}", tok, {"formData": {"is_retake": True}})
+        r = sub(tok, sid)
+        check("勾了补考但一科没选 → VAL-15（警告，可提交）",
+              "VAL-15" in rules(r), f"实际: {sorted(rules(r))}")
+        check("这一条是警告不是错误（severity=2）",
+              any(i.get("ruleCode") == "VAL-15" and i.get("severity") == 2
+                  for i in (r.get("data") or {}).get("issues") or []),
+              "severity 不对")
+        # 四级最多 1 科：勾 2 科必须拦
+        call("PUT", f"{base}/submissions/{sid}", tok, {"formData": {
+            "retake_theory": "理论", "retake_practice": "实操"}})
+        r = sub(tok, sid)
+        check("四级勾 2 科 → VAL-15 阻断", "VAL-15" in rules(r), f"实际: {sorted(rules(r))}")
+        check("升级为错误（severity=3）",
+              any(i.get("ruleCode") == "VAL-15" and i.get("severity") == 3
+                  for i in (r.get("data") or {}).get("issues") or []),
+              "severity 不对")
+        # ⚠️ 补考三科都是**必填且可见**（is_retake=1），所以不能留空 ——
+        #    学生必须每项都明确选，「无」是合法值（PRD §2.6 的三条铁律）。
+        #    用「无」而不是 "" 也正是为了把这条铁律跑一遍
+        call("PUT", f"{base}/submissions/{sid}", tok, {"formData": {
+            "retake_practice": "无", "retake_overall": "无"}})
+        r = sub(tok, sid)
+        check("只留 1 科 → 不再有 VAL-15 错误",
+              "VAL-15" not in rules(r), f"实际: {sorted(rules(r))}")
+
+        # ── 4. 传材料 → VAL-05 消失 ──
+        print("\n[4] 传上必传材料")
+        img = small_jpeg()
+        up = upload(tok, sid, "IDCARD", "身份证正面.jpg", img)
+        if not check("上传成功", up.get("code") == 0, f"{up}"):
+            return 1
+        call("POST", f"/api/v1/mp/files/{up['data']['id']}/process", tok)
+        # 同一张图再传一份（IDCARD 允许 2 份）→ 制造一条**可预期的黄色警告**，
+        # 否则"无红无黄 → 直接提交成功"，下一步那条"先确认再提交"的路径就成了没被测过的代码
+        up2 = upload(tok, sid, "IDCARD", "身份证反面.jpg", img)
+        if up2.get("code") == 0:
+            call("POST", f"/api/v1/mp/files/{up2['data']['id']}/process", tok)
+        r = sub(tok, sid)
+        rs = rules(r)
+        check("VAL-05 消失（材料齐了）", "VAL-05" not in rs, f"实际: {sorted(rs)}")
+        check("VAL-09 同一张图传两份 → 黄色警告", "VAL-09" in rs, f"实际: {sorted(rs)}")
+        check("有警告但无红色 → 未确认时**不提交**",
+              (r.get("data") or {}).get("submitted") is False, f"{r.get('data')}")
+
+        # ── 5. 只剩警告 → 两次提交 ──
+        print("\n[5] 只有警告时：先确认，再提交")
+        r = sub(tok, sid, confirm=False)
+        d = r.get("data") or {}
+        if not d:
+            print(f"         ⚠ 响应没有 data：{json.dumps(r, ensure_ascii=False)[:300]}")
+        rs = rules(r)
+        red = [i for i in (d.get("issues") or []) if i.get("severity") == 3]
+        check("已无红色项", not red, f"仍有红: {[i.get('ruleCode') for i in red]}")
+        if (d.get("yellowCount") or 0) > 0:
+            check("有警告且未确认 → submitted=false（等学生确认）",
+                  d.get("submitted") is False, f"{d.get('submitted')}")
+            r2 = sub(tok, sid, confirm=True)
+            check("带 confirmWarnings=true → 提交成功",
+                  (r2.get("data") or {}).get("submitted") is True, f"{r2.get('data')}")
+        else:
+            check("没有警告 → 直接提交成功", d.get("submitted") is True, f"{d}")
+
+        st = sql(f"SELECT status || '|' || (submit_time IS NOT NULL)::text || '|' || "
+                 f"red_count || '|' || yellow_count FROM collect_submission WHERE id={sid}")
+        parts = st.split("|")
+        check("库里状态 → 1（待初审）", parts[0] == "1", f"status={parts[0]}")
+        check("submit_time 已写入", parts[1] == "true", f"{parts[1]}")
+        check("红黄计数落库（班主任端要用它排序）", parts[2] == "0", f"red={parts[2]} yellow={parts[3]}")
+
+        # ── 6. 防重复提交 ──
+        print("\n[6] 防重复提交")
+        r = sub(tok, sid)
+        check("再提交被拒", r.get("code") not in (0, None), f"code={r.get('code')}")
+        check("提示里说明了当前状态（学生最怕不知道提交成功没有）",
+              "已提交" in (r.get("msg") or "") or "初审" in (r.get("msg") or ""),
+              f"msg={r.get('msg')}")
+    finally:
+        print("\n[7] 清理")
+        sql(f"DELETE FROM collect_file WHERE submission_id={sid}")
+        sql(f"DELETE FROM collect_submission WHERE id={sid}")
+        sql(f"UPDATE collect_roster SET bind_user_id=NULL, bind_time=NULL, bind_type=NULL WHERE id={rid}")
+        import shutil
+        for sub2 in ("raw", "std"):
+            d2 = os.path.join(ROOT, "backend", "data", "uploads", sub2, "1", str(task_id))
+            if os.path.exists(d2):
+                shutil.rmtree(d2, ignore_errors=True)
+        ok("已清理")
+
+    print("\n" + "=" * 72)
+    print(f"通过 {len(PASSED)}　失败 {len(FAILED)}　共 {len(PASSED) + len(FAILED)}")
+    for n, dd in FAILED:
+        print(f"  · {n}\n      {dd}")
+    print("=" * 72)
+    return 1 if FAILED else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())