xiaotutongxue/scripts/e2e-m3-file.py

#!/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())