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 http_get(url: str) -> tuple[int, bytes]:
    """
    下载一个 URL 的内容。

    ⚠️ 七牛的签名链接在**响应体**里也是文件内容，但错误时不是（403 会回一段 JSON）——
    所以调用方要按状态码分支，不要盲目当文件用。
    """
    try:
        with urllib.request.urlopen(url, timeout=60) as x:
            return x.status, x.read()
    except urllib.error.HTTPError as e:
        return e.code, b""
    except Exception:
        return 0, b""


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 _portrait_phone_jpeg(w: int = 3200, h: int = 2400) -> bytes:
    """
    造一张"手机竖拍"的照片：**像素是横的**（3200×2400）+ EXIF Orientation=6。

    这跟真实情况一致：手机传感器是横的，竖着拿时记录的是横图 + 一个"转 90°"的标记。
    浏览器认这个标记所以看着是正的，而服务端如果不主动转正、又把标记丢掉，
    出来的就是一张躺倒的图。
    """
    from PIL import Image
    img = Image.new("RGB", (w, h), (205, 212, 220))
    # 左上角画一个方向标记，便于人工确认
    for y in range(0, 300):
        for x in range(0, 300):
            img.putpixel((x, y), (40, 90, 200))
    exif = Image.Exif()
    exif[0x0112] = 6          # 6 = Rotate 90 CW
    buf = io.BytesIO()
    img.save(buf, "JPEG", quality=92, exif=exif)
    return buf.getvalue()


def make_png(w: int = 900, h: int = 1200) -> bytes:
    """
    造一张 PNG（截图类材料的常见形态）。

    用来验证"格式按**字节**判、不按后缀判"：同一份字节改个后缀名，
    服务端必须仍然认出它是 png。
    """
    from PIL import Image
    img = Image.new("RGB", (w, h), (247, 247, 247))
    px = img.load()
    # 画几行横线，模拟表格（也让它不是纯色，免得压出来只有几百字节）
    for y in range(40, h, 60):
        for x in range(30, w - 30):
            px[x, y] = (30, 30, 30)
    buf = io.BytesIO()
    img.save(buf, "PNG")
    return buf.getvalue()


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] 交付件必须取得到；原始件按策略（默认不保留）")

    stdpath = sql(f"SELECT coalesce(std_path,'') FROM collect_file WHERE id={fid}")
    check("std_path 已写入（交付件就是它）", stdpath != "", f"std_path={stdpath}")
    # ⚠️ **不要断言"文件在磁盘上"** —— 存储是可换的（本地磁盘 / 七牛云）。
    #    写死磁盘路径的断言，在切到对象存储后会整片变红，而那看起来像代码坏了。
    #    正确做法是**通过产品接口把文件取回来** —— 对两种存储都成立，
    #    而且顺带验证了签名链接、私有空间访问、内容完整性。
    ustd = call("GET", f"/api/v1/mp/files/{fid}/url", tok)
    url_std = ((ustd.get("data") or {}).get("url")) or ""
    check("能取到交付件的访问链接", url_std != "", f"{ustd}")
    status_std, body_std = http_get(url_std)
    check("交付件取得回来（HTTP 200）", status_std == 200, f"HTTP {status_std}")
    if status_std == 200:
        check("交付件是张真图片（JPEG 开头）", body_std[:2] == b"\xff\xd8", f"{body_std[:2]!r}")

    # 原始件：**默认不保留**（客户决定：交付物就是规范件）。
    # 这里断言的是"策略被正确执行"，而不是"原始件在不在" ——
    # 因为两种策略都是对的，取决于配置。
    rawpath = sql(f"SELECT coalesce(raw_path,'') FROM collect_file WHERE id={fid}")
    keep_raw = os.environ.get("XT_STORAGE_KEEP_RAW", "false").lower() == "true"
    if keep_raw:
        check("开了 keep-raw：原始件保留在 raw/", rawpath != "", f"raw_path={rawpath}")
    else:
        check("默认策略：**不保留原始件**（raw_path 为空）", rawpath == "", f"raw_path={rawpath}")
        # 关键：即使没有原始件，"看原图"也不该报错 —— 要优雅退回到交付件。
        # 学生点"看原图"只是想看自己传的那份，报"文件不存在"是纯粹的困惑。
        uraw = call("GET", f"/api/v1/mp/files/{fid}/url?original=true", tok)
        d = uraw.get("data") or {}
        check("没有原始件时 original=true **不报错**（优雅退到交付件）",
              uraw.get("code") == 0, f"{uraw}")
        check("并且如实标注取到的是哪一份", d.get("original") is False, f"original={d.get('original')}")
        st_raw, body_raw = http_get(d.get("url") or "")
        check("退回的那份也能下载到", st_raw == 200, f"HTTP {st_raw}")

    # ── 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. EXIF 朝向（真实学生照片的头号坑）───────────────
    print("\n[7] EXIF 朝向：竖拍照片不能变成躺倒的")
    # 手机竖着拍时，**文件里的像素其实是横的** + 一个 Orientation=6（显示时转 90°）。
    # ImageIO 不读 EXIF、重编码又丢掉它，所以服务端必须**主动按 EXIF 转正** ——
    # 否则机构看到的是一张转了 90° 的身份证。
    # ⚠️ 这个用例以前没有，因为测试图都是程序生成的、不带 EXIF，而**真实手机照片几乎都带**。
    exif_img = _portrait_phone_jpeg()
    # 前面几节已经把 IDCARD 的 2 份额度用完了，先清掉再传 ——
    # 直接改这一项的额度（而不是换个材料）是为了让用例验的仍是"身份证 + P2 档"
    # 这条真实组合：P1 会裁成固定尺寸、P4 会转 PNG，都验不出"有没有转正"。
    sql(f"UPDATE collect_file SET deleted = 1 WHERE submission_id={sid} AND material_code='IDCARD'")
    up3 = upload("/api/v1/mp/files/upload", tok, sid, "IDCARD", "竖拍身份证.jpg", exif_img, "image/jpeg")
    check("带 EXIF 的竖拍照片能上传", up3.get("code") == 0, f"{up3}")
    if up3.get("code") == 0:
        fid3 = up3["data"]["id"]
        row = sql(f"SELECT width || 'x' || height FROM collect_file WHERE id={fid3}")
        w, _, h = row.partition("x")
        print(f"         输入：像素 3200×2400（横）+ EXIF Orientation=6 → 应是竖的")
        print(f"         输出：{row}")
        check("**输出是竖的**（按 EXIF 转正了，不是躺倒的）",
              int(h or 0) > int(w or 0), f"输出 {row} —— 高应大于宽")
        check("长边仍受 P2 档约束（≤1600）",
              max(int(w or 0), int(h or 0)) <= 1600, f"长边 {max(int(w or 0), int(h or 0))}")
        # 取回来验字节里确实没有残留的 EXIF 标记（否则浏览器可能再转一次 → 反倒躺倒）
        u3 = call("GET", f"/api/v1/mp/files/{fid3}/url", tok)
        st3, body3 = http_get((u3.get("data") or {}).get("url") or "")
        check("转正后的文件里**不再带 EXIF 朝向**（否则会被转两次）",
              st3 == 200 and b"Exif\x00\x00" not in body3[:4096],
              f"HTTP {st3}，前 4096 字节里{'有' if b'Exif' in body3[:4096] else '没有'} EXIF")

    # ── 8. 客户端压缩后的上传（文件名没有后缀 / 格式按字节判）─────────
    print("\n[8] 客户端压缩后上传：文件名没后缀也要收（真 bug 的回归）")
    # 客户端压缩完交给 uni.uploadFile 的是 blob 地址，multipart 里的名字是一串 UUID、
    # **没有后缀** —— 而修复前服务端只看后缀，会把一张好端端的照片判成 .bin 拒掉。
    # 也就是"前端压缩一上线，学生就传不了材料"。
    sql(f"UPDATE collect_file SET deleted = 1 WHERE submission_id={sid} AND material_code='IDCARD'")
    r = upload("/api/v1/mp/files/upload", tok, sid, "IDCARD",
               "4c1e2f3a-0000-1111-2222-333344445555", raw[:60000], "image/jpeg")
    check("**没有后缀**的真 JPEG 被接受（按字节判格式，不看后缀）", r.get("code") == 0, f"{r}")

    # 逆过来：字节是真 PNG、但后缀写着 .txt —— 按字节认作 png（在白名单里）就该收。
    # 守的是"学生把截图改名成 .txt 交上来"，别等导出给机构才发现格式不对。
    png = make_png(900, 1200)
    r2 = upload("/api/v1/mp/files/upload", tok, sid, "IDCARD", "身份证.txt", png, "text/plain")
    check("字节是 PNG、后缀写着 .txt → 按字节认作 png 并接受", r2.get("code") == 0, f"{r2}")

    # ── 9. 清理 ──────────────────────────────────────
    print("\n[9] 清理")
    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())