feat(file): 材料上传与规范化（压缩到 1M 内 + 按规则改名）

这是产品的核心资产。实测那批真实材料里 **52.7% 的原始文件名是违规的**
（`曾伊豪-...110身份证.jpg.jpg` 这种），而认定机构只看两件事：
**体积有没有超 1M**、**名字对不对得上人**。这两件事做不对，
材料会被整批退回 —— 而退回时往往已经截止了。

两步走，不能合成一步：
 ① upload：先原样存下来。这一步必须快、必须成功 —— 学生的手机可能是慢网络，
    上传中断是最常见的失败。先落盘，学生就不必重传。
 ② process：压缩 + 改名。慢（要解码、试多档质量），但在服务端跑，学生不用等网络。

压缩策略：**先保清晰度，再压质量，最后才缩尺寸**。顺序不能反 ——
证件类的可读性是底线（机构要能看清号码），缩过的证件翻拍件数字就糊了。
实测：11.6MB 的 2400x3200 原图 → **712KB**，1020x1360（P2 长边限 1600），q70。

命名策略：模板 + 白名单 + 非法字符清理。
  ⚠️ 非法字符清理不能省：学生的原始文件名里 `/` 很常见（"学籍表(1/2)"），
     而带 `/` 的名字**根本写不进磁盘** —— 表现是"保存失败"，
     而错误信息不会告诉你是因为文件名里有个斜杠。

原始文件**绝不覆盖**（另存 std/）：认定机构退件时我们要能拿出原始件，
覆盖掉原图就再也说不清了（"是不是你们压坏的"）。E2E 断言了原图一字不差。

⚠️ 与契约的一处偏差（已知、有意）：契约写的是 POST /mp/files/upload-token
   （OSS 直传凭证），当前**后端代理接收**（multipart）。原因：后端还没接对象存储，
   直传凭证要先有 bucket 与签名机制。硬做一个"返回假凭证"的接口只会让人以为能直传。
   接入对象存储后可以并存：小文件走代理、大文件走直传。

⚠️ 当前把文件存在**本地磁盘**（fill.upload.dir）。多实例部署时各存各的会丢文件，
   所以接入对象存储前**不能上多副本**。这一点必须记住，别当成"已经解决了"。

E2E 里修了一个夹具陷阱（值得记）：用**裸 SQL** 插材料项会绕过 MaterialCatalog
   的默认值（DDL 列默认是 AUTO / 1 份），于是"档位没生效""份数不对" ——
   而这两条都会被误当成代码 bug。实测就踩了：IDCARD 期望 P2 + 2 份，
   裸 SQL 插出来是 AUTO + 1 份，AUTO 的行为（不缩放、先降质量）被当成了"P2 没生效"。

细节修正：文件名里的脱敏号用 X 而不是下划线。`*` 在 Windows 文件名里非法，
   被 sanitize 换成下划线后 `361029******0015` 变成 `361029_0015` ——
   没人看得出那是脱敏号。现在是 `361029XXXXXX0015`。

验收：e2e-m3-file.py 29/29（含体积、档位、命名、原图保留、幂等、白名单、份数）。

scripts/e2e-m3-file.py

@@ -0,0 +1,327 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+材料上传与规范化 验收测试（《03-接口设计》§9.5~§9.7 的核心）。
+
+    [x] 上传原图（后端代理接收）
+    [x] 处理：**压到体积上限内**（这是认定机构的硬要求，无例外）
+    [x] 处理：**按命名模板改名**（规范名要能唯一归属到人）
+    [x] 档位生效：P2 证件翻拍 → 长边被限到 1600
+    [x] 原始件保留（退回时要能拿出原图核对）
+    [x] 格式白名单 / 份数上限
+    [x] 幂等：重复处理不产生第二份规范文件
+
+⚠️ 为什么这个 E2E 特别有价值：
+   压缩与改名是**产品的核心资产**（实测 52.7% 的原始文件名违规），
+   而它同时是"最容易被以为做对了"的地方 —— 尺寸小了、名字变了，
+   但**体积没降到 1M 以内**，或者名字里少了证件号。
+   只有拿真实图片跑一遍、逐项断言，才知道到底行不行。
+
+跑法：python scripts/e2e-m3-file.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.parse
+import urllib.request
+import uuid
+
+sys.stdout.reconfigure(encoding="utf-8")
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+DEV_YML = os.path.join(ROOT, "backend", "xt-server", "src", "main", "resources", "application-dev.yml")
+DATA_DIR = os.path.join(ROOT, "backend", "data", "uploads")
+PG = os.environ.get("XT_PG_CONTAINER", "xt-pg")
+OPENID = os.environ.get("XT_TEST_OPENID", "e2e_file_openid")
+
+PASSED: list[str] = []
+FAILED: list[tuple[str, str]] = []
+CREATED_FILES: list[int] = []
+CREATED_SUBS: list[int] = []
+
+
+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:
+    # issuer 在主配置 application.yml 里，secret 在 application-dev.yml 里 ——
+    # 只读一个会漏（而症状是 401，完全指不到"配置没读到"）
+    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 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()))
+    sig = hmac.new(key, signing.encode(), hashlib.sha512).digest()
+    return signing + "." + b64u(sig)
+
+
+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(path: str, token: str, submission_id: int, material_code: str,
+           filename: str, data: bytes, ctype: str) -> dict:
+    boundary = "----xte2e" + uuid.uuid4().hex[:12]
+    parts = []
+    for k, v in [("submissionId", str(submission_id)), ("materialCode", material_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: {ctype}\r\n\r\n".encode())
+    parts.append(data)
+    parts.append(f"\r\n--{boundary}--\r\n".encode())
+    body = b"".join(parts)
+    req = urllib.request.Request(
+        "http://localhost:8080" + path, data=body, 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 make_big_photo(w: int = 2400, h: int = 3200) -> bytes:
+    """造一张"大且难压"的图：随机噪点。纯色图会被压到几十 KB，测不出压缩能力。"""
+    from PIL import Image
+    import random
+    random.seed(42)
+    img = Image.new("RGB", (w, h))
+    px = img.load()
+    # 每 4 个像素填一次，四两拨千斤地造出高频噪点（全填太慢）
+    for y in range(0, h, 2):
+        for x in range(0, w, 2):
+            c = (random.randint(0, 255), random.randint(0, 255), random.randint(0, 255))
+            px[x, y] = c
+            if x + 1 < w:
+                px[x + 1, y] = c
+            if y + 1 < h:
+                px[x, y + 1] = c
+                if x + 1 < w:
+                    px[x + 1, y + 1] = c
+    buf = io.BytesIO()
+    img.save(buf, "JPEG", quality=98)
+    return buf.getvalue()
+
+
+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
+
+    print("=" * 72)
+    print(f"材料上传与规范化 E2E　task={task_id}")
+    print("=" * 72)
+
+    cfg = read_cfg()
+    # 用一个固定的测试账号（不必是真实微信账号）：先过门禁，再传材料
+    uid = 992001
+    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", "/api/v1/mp/submissions", tok, {"taskId": task_id})
+    if r.get("code") != 0:
+        fail("建草稿", f"{r}")
+        return 1
+    sid = r["data"]["id"]
+    CREATED_SUBS.append(sid)
+    ok(f"已过门禁并建草稿（submission={sid}）")
+
+    # ⚠️ 夹具必须**显式**把材料项配成目录规定的那样。
+    #    用裸 SQL 插入会绕过 MaterialCatalog 的默认值（DDL 的列默认是 AUTO / 1 份），
+    #    于是"档位没生效""份数不对"—— 而这两条都会被误当成代码 bug。
+    #    实测就踩了：IDCARD 期望 P2 + 2 份，裸 SQL 插出来是 AUTO + 1 份，
+    #    压缩结果 1252x1669（AUTO 先降质量、不行才缩尺寸）而被当成"P2 档位没生效"。
+    sql(f"""UPDATE collect_material_def SET compress_profile='P2', max_count=2, min_quality=70
+            WHERE task_id={task_id} AND material_code='IDCARD'""")
+    prof = sql(f"SELECT compress_profile || '/' || max_count FROM collect_material_def "
+               f"WHERE task_id={task_id} AND material_code='IDCARD'")
+    check("夹具：IDCARD 已配成目录规定的 P2 + 2 份", prof == "P2/2", f"实际 {prof}")
+
+    # ── 1. 上传一张真实的大图 ─────────────────────────
+    print("\n[1] 上传原始大图")
+    raw = make_big_photo()
+    print(f"         原始图：2400x3200，{len(raw) / 1024:.0f} KB")
+    check("原图确实超过 1M（否则测不出压缩能力）", len(raw) > 1024 * 1024, f"{len(raw) / 1024:.0f} KB")
+
+    r = upload("/api/v1/mp/files/upload", tok, sid, "IDCARD", "身份证正面.jpg", raw, "image/jpeg")
+    check("上传成功", r.get("code") == 0, f"code={r.get('code')} msg={r.get('msg')}")
+    fid = (r.get("data") or {}).get("id")
+    CREATED_FILES.append(fid)
+    check("返回文件 id", isinstance(fid, int), f"{fid}")
+
+    row = sql(f"SELECT raw_size || '|' || coalesce(raw_path,'') FROM collect_file WHERE id={fid}")
+    raw_size_db = row.split("|")[0]
+    check("原始件已落盘记录（raw_size 是原图大小）", raw_size_db == str(len(raw)),
+          f"库里 {raw_size_db} vs 上传 {len(raw)}")
+
+    # ── 2. 处理：压缩 + 改名 ──────────────────────────
+    print("\n[2] 规范化（压缩 + 按规则改名）")
+    r = call("POST", f"/api/v1/mp/files/{fid}/process", tok)
+    if not check("处理成功", r.get("code") == 0, f"code={r.get('code')} msg={r.get('msg')}"):
+        return 1
+    d = r.get("data") or {}
+    print(f"         {d.get('originalName')} → {d.get('standardName')}")
+    print(f"         {d.get('rawSize')} → {d.get('finalSize')} 字节，"
+          f"{d.get('width')}x{d.get('height')} q{d.get('quality')}")
+
+    check("**体积压到 1024KB 以内**（认定机构的硬要求）",
+          (d.get("finalSize") or 0) <= 1024 * 1024, f"finalSize={d.get('finalSize')}")
+    check("体积确实降下来了（不是原样返回）",
+          (d.get("finalSize") or 0) < (d.get("rawSize") or 0), f"{d.get('rawSize')} → {d.get('finalSize')}")
+
+    # IDCARD 的默认档位是 P2（证件翻拍）→ 长边限到 1600
+    long_side = max(d.get("width") or 0, d.get("height") or 0)
+    check("P2 档位生效：长边被限到 1600", long_side <= 1600, f"长边={long_side}")
+    check("短边按比例缩放（不是被压扁）",
+          abs((d.get("width") or 1) / (d.get("height") or 1) - 2400 / 3200) < 0.02,
+          f"宽高比 {d.get('width')}x{d.get('height')}")
+
+    # ── 3. 命名规则 ───────────────────────────────────
+    print("\n[3] 规范文件名")
+    name = d.get("standardName") or ""
+    check("名字里有姓名（能归属到人）", "张同学" in name, f"name={name}")
+    check("名字里有证件号（脱敏，不是完整号）",
+          "361029" in name and "361029200912100015" not in name, f"name={name}")
+    check("脱敏号在文件名里用 X 而不是下划线（* 是 Windows 非法字符，"
+          "直接替换成下划线会让人看不出那是脱敏号）",
+          "XXXXXX" in name.upper() or "X" in name.upper(), f"name={name}")
+    check("名字里有材料名", "身份证" in name, f"name={name}")
+    check("后缀正确（压成 jpg）", name.lower().endswith(".jpg"), f"name={name}")
+    check("名字里没有非法字符（/ : * ? \" < > |）",
+          not re.search(r'[\\/:*?"<>|]', name), f"name={name}")
+    check("长度不超过 90（Windows 路径总长限制）", len(name) <= 90, f"len={len(name)}")
+
+    # ── 4. 原始件保留 ─────────────────────────────────
+    print("\n[4] 原始件必须保留")
+    rawpath = sql(f"SELECT coalesce(raw_path,'') FROM collect_file WHERE id={fid}")
+    check("raw_path 还在（没有被规范文件覆盖）", rawpath != "", f"raw_path={rawpath}")
+    full = os.path.join(DATA_DIR, rawpath.replace("/", os.sep))
+    check("原始文件真的存在磁盘上", os.path.exists(full), f"路径={full}")
+    if os.path.exists(full):
+        check("原始文件大小与上传一致（一字不差）",
+              os.path.getsize(full) == len(raw), f"{os.path.getsize(full)} vs {len(raw)}")
+    stdpath = sql(f"SELECT coalesce(std_path,'') FROM collect_file WHERE id={fid}")
+    full2 = os.path.join(DATA_DIR, stdpath.replace("/", os.sep))
+    check("规范文件也落盘了", os.path.exists(full2), f"路径={full2}")
+
+    # ── 5. 幂等 ───────────────────────────────────────
+    print("\n[5] 幂等：重复处理")
+    r2 = call("POST", f"/api/v1/mp/files/{fid}/process", tok)
+    check("重复处理成功", r2.get("code") == 0, f"code={r2.get('code')}")
+    cnt = sql(f"SELECT count(*) FROM collect_file WHERE submission_id={sid} AND material_code='IDCARD'")
+    check("没有产生第二份记录", cnt == "1", f"{cnt}")
+    check("规范文件名稳定（同一次处理结果一致）",
+          (r2.get("data") or {}).get("standardName") == name,
+          f"{(r2.get('data') or {}).get('standardName')}")
+
+    # ── 6. 白名单与份数 ───────────────────────────────
+    print("\n[6] 格式白名单与份数上限")
+    r = upload("/api/v1/mp/files/upload", tok, sid, "IDCARD", "身份证.exe", b"MZfake", "application/octet-stream")
+    check("白名单外的格式被拒（40002）", r.get("code") == 40002, f"code={r.get('code')} msg={r.get('msg')}")
+    r = upload("/api/v1/mp/files/upload", tok, sid, "IDCARD", "身份证反面.jpg", raw[:50000], "image/jpeg")
+    check("超过份数上限被拒（IDCARD 默认 2 份，现在已有 1 份 → 这次应成功）",
+          r.get("code") == 0, f"code={r.get('code')} msg={r.get('msg')}")
+    if r.get("code") == 0:
+        CREATED_FILES.append((r.get("data") or {}).get("id"))
+        r = upload("/api/v1/mp/files/upload", tok, sid, "IDCARD", "身份证再来一份.jpg", raw[:50000], "image/jpeg")
+        check("第 3 份被拒（且提示先删上一份）",
+              r.get("code") == 40001 and "最多" in (r.get("msg") or ""),
+              f"code={r.get('code')} msg={r.get('msg')}")
+
+    # ── 7. 清理 ───────────────────────────────────────
+    print("\n[7] 清理")
+    sql(f"DELETE FROM collect_file WHERE submission_id IN ({','.join(str(x) for x in CREATED_SUBS)})")
+    sql(f"DELETE FROM collect_submission WHERE id IN ({','.join(str(x) for x in CREATED_SUBS)})")
+    sql(f"UPDATE collect_roster SET bind_user_id=NULL, bind_time=NULL, bind_type=NULL WHERE id={rid}")
+    import shutil
+    for sub in ("raw", "std"):
+        d2 = os.path.join(DATA_DIR, sub, "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())