feat(storage): 对象存储接七牛云 —— 解除"不能上多副本"
文件以前直接写本地磁盘:两个实例各存各的,学生这次传到 A、下次请求落到 B,
文件就"不见了"。这不是性能问题,是正确性问题 —— 它一直卡着不上多副本。
新增:
· module/file/storage/{StorageService,LocalStorageService,QiniuStorageService}
· config/StorageConfig(装配 + 可选启动自检)
· FillProperties 新增 fill.storage
· MaterialFileController 新增 GET /{id}/url(签名链接)与 GET /{id}/content
· scripts/e2e-m1-storage.py(存储自检,没配七牛时跳过而非失败)
关键设计:
① 业务只认 path(std/1/17/张同学-…-身份证.jpg),各实现自己映射成真实位置:
本地 ./data/uploads/std/...、七牛 xiaotutongxue/std/...
**换存储不用改数据库、不用改业务代码**
② 空间私有 → 下载必须签名。GET /{id}/url 每次现签(默认 30 分钟),
所以这个 URL 绝不能长期保存:写进库、塞进导出的 Excel、发给学生明天再打开,都是 403
③ 对象键里有中文,拼 URL **必须逐段编码** —— 整串 encode 会把 / 编成 %2F,
那在七牛那边是另一个键,表现是"传上去了却打不开",且报错什么都没有
④ 区域默认 auto:华东-浙江在新旧控制台对应 z0 与 cn-east-2,
写错了的症状是**上传超时**——一个从报错完全看不出原因的失败
⑤ AK/SK 不进仓库也不进 jar:dev.yml 只放占位符;
真密钥在 backend/application-local.yml(已 gitignore),
用 spring.config.import: optional:**file**:./application-local.yml
—— 刻意用 file: 而不是 classpath:,后者会被打进 jar,密钥就跟着构建产物走了
⑥ 配了 qiniu 却没密钥 → 启动就炸(fail-fast):起得来的话第一个受害的是
正在填材料的学生,而排查要从"文件保存失败"一路追到环境变量
已验证:存储抽象改造没破坏任何现有链路 —— 本地磁盘路径下
e2e-m3-file 29/29、m3-submit 31/31、m1-materials 68/68、m3-draft 27/27 全绿;
启动日志打出「本地磁盘(多副本部署下会丢文件)」;git check-ignore 确认
application-local.yml 被忽略而 .example 不被忽略。
⏳ 未验证:**七牛的真实链路没跑过** —— 用户给了域名/空间/区域/私有/目录前缀,
但没给 AccessKey/SecretKey,没有它生成不了上传凭证。
补上后跑 python scripts/e2e-m1-storage.py 即可(它会验:配置齐全 →
上传处理落进七牛 → 取签名链接 → 真的下载回来比字节 → 不带签名被拒
→ 对象键带前缀 → 原始件一字不差)。没配时脚本跳过并 exit 0,不误报失败。
This commit is contained in:
@@ -0,0 +1,324 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
对象存储 自检(七牛云 / 本地磁盘)。
|
||||
|
||||
[x] 配置是否就绪(provider / bucket / domain / 密钥)
|
||||
[x] 材料上传 → 处理 → 落进对象存储
|
||||
[x] 取签名链接能下载到**字节一致**的内容 ← 真正的证明
|
||||
[x] **不带签名**的同地址被拒(证明空间确实是私有的,签名在起作用)
|
||||
[x] 对象键带目录前缀(与同空间其他项目隔离)
|
||||
|
||||
⚠️ 为什么要有这个脚本:
|
||||
"文件传上去了"和"文件**取得回来**"是两件事。七牛空间是私有的,
|
||||
少签一个名、签错有效期、键里的中文没编码对 —— 每一种都会让
|
||||
上传**成功**但下载失败(403/404)。只有真的把字节取回来比一遍才知道配对了。
|
||||
|
||||
跑法:
|
||||
python scripts/e2e-m1-storage.py
|
||||
没配置七牛时会直接跳过(exit 0),不会误报失败。
|
||||
"""
|
||||
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__)))
|
||||
SRC = os.path.join(ROOT, "backend", "xt-server", "src", "main", "resources")
|
||||
LOCAL_YML = os.path.join(ROOT, "backend", "application-local.yml")
|
||||
PG = os.environ.get("XT_PG_CONTAINER", "xt-pg")
|
||||
UID = 992004
|
||||
BASE = "http://localhost:8080"
|
||||
|
||||
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 skip(n: str, why: str) -> None:
|
||||
print(f" [SKIP] {n}\n {why}")
|
||||
|
||||
|
||||
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_yml(path: str) -> str:
|
||||
return open(path, encoding="utf-8").read() if os.path.exists(path) else ""
|
||||
|
||||
|
||||
def resolve(v: str | None) -> str | None:
|
||||
"""
|
||||
把 `${ENV:default}` 解析成实际值。
|
||||
|
||||
⚠️ 不解析的话自检会**说谎**:显示成 `provider=${XT_STORAGE_PROVIDER:local}` ——
|
||||
既不是 local 也不是 qiniu,读的人只会更糊涂。配置检查这一环一旦不可信,
|
||||
它给的"通过"也就没意义了。
|
||||
"""
|
||||
if not v:
|
||||
return None
|
||||
m = re.fullmatch(r"\$\{([^:}]+):([^}]*)\}", v)
|
||||
if m:
|
||||
env = os.environ.get(m.group(1))
|
||||
return ((env if env is not None else m.group(2)).strip()) or None
|
||||
m = re.fullmatch(r"\$\{([^}]+)\}", v)
|
||||
if m:
|
||||
return (os.environ.get(m.group(1)) or "").strip() or None
|
||||
return v
|
||||
|
||||
|
||||
def cfg_value(text: str, key: str) -> str | None:
|
||||
m = re.search(rf"^\s*{re.escape(key)}:\s*(.+?)\s*$", text, re.M)
|
||||
if not m:
|
||||
return None
|
||||
return resolve(m.group(1).strip().strip("'\"") or None)
|
||||
|
||||
|
||||
def storage_cfg() -> dict:
|
||||
"""
|
||||
读配置的顺序与 Spring 一致:**application-local.yml 覆盖 application-dev.yml**。
|
||||
不这么做会出现"我以为在用七牛、其实脚本读的是 dev 里的 local"这种自欺。
|
||||
"""
|
||||
dev = read_yml(os.path.join(SRC, "application-dev.yml"))
|
||||
local = read_yml(LOCAL_YML)
|
||||
out = {}
|
||||
for k in ("provider", "prefix", "access-key", "secret-key", "bucket", "domain", "region"):
|
||||
out[k] = cfg_value(dev, k)
|
||||
for k in ("provider", "prefix", "access-key", "secret-key", "bucket", "domain", "region"):
|
||||
v = cfg_value(local, k)
|
||||
if v:
|
||||
out[k] = v
|
||||
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(BASE + path, data=data, headers=h, method=method)
|
||||
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 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(
|
||||
BASE + "/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=120) 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]:
|
||||
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 small_jpeg() -> bytes:
|
||||
from PIL import Image
|
||||
buf = io.BytesIO()
|
||||
Image.new("RGB", (900, 1200), (206, 212, 218)).save(buf, "JPEG", quality=90)
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--task", type=int, default=17)
|
||||
args = ap.parse_args()
|
||||
task_id = args.task
|
||||
|
||||
print("=" * 72)
|
||||
print("对象存储 自检")
|
||||
print("=" * 72)
|
||||
|
||||
cfg = storage_cfg()
|
||||
provider = (cfg.get("provider") or "local").lower()
|
||||
print(f" provider = {provider}")
|
||||
print(f" bucket = {cfg.get('bucket')}")
|
||||
print(f" domain = {cfg.get('domain')}")
|
||||
print(f" prefix = {cfg.get('prefix')}")
|
||||
print(f" region = {cfg.get('region')}")
|
||||
|
||||
if provider != "qiniu":
|
||||
print()
|
||||
skip("七牛云自检", "当前 provider=local(文件存在本地磁盘,多副本部署会丢文件)")
|
||||
print("""
|
||||
要启用七牛云:
|
||||
① 编辑 backend/application-local.yml(已在 .gitignore 里,不会被提交)
|
||||
② 取消 provider 那行的注释改成 qiniu,并填上 access-key / secret-key
|
||||
③ 重启后端,再跑一次这个脚本
|
||||
""")
|
||||
print("=" * 72)
|
||||
print("跳过(不是失败):当前在用本地磁盘")
|
||||
print("=" * 72)
|
||||
return 0
|
||||
|
||||
for k in ("access-key", "secret-key", "bucket", "domain"):
|
||||
if not cfg.get(k):
|
||||
fail(f"配置 {k}", f"provider=qiniu 但 {k} 为空 —— 服务端会在启动时就拒绝")
|
||||
return 1
|
||||
ok("七牛配置齐全(provider/bucket/domain/密钥)")
|
||||
|
||||
text = read_yml(os.path.join(SRC, "application-dev.yml")) + read_yml(os.path.join(SRC, "application.yml"))
|
||||
sec = re.search(r"^\s*secret:\s*(\S+)\s*$", text, re.M)
|
||||
iss = re.search(r"^\s*issuer:\s*(\S+)\s*$", text, re.M)
|
||||
if not sec or not iss:
|
||||
fail("读 JWT 配置", "找不到 secret/issuer")
|
||||
return 1
|
||||
tok = mint(sec.group(1), iss.group(1), UID, 1)
|
||||
|
||||
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"]
|
||||
print(f" 草稿 submission={sid}")
|
||||
|
||||
try:
|
||||
print("\n[1] 上传 → 处理")
|
||||
img = small_jpeg()
|
||||
up = upload(tok, sid, "IDCARD", "身份证正面.jpg", img)
|
||||
if not check("上传成功(这一步已经把原图写进对象存储)", up.get("code") == 0, f"{up}"):
|
||||
return 1
|
||||
fid = up["data"]["id"]
|
||||
pr = call("POST", f"/api/v1/mp/files/{fid}/process", tok)
|
||||
check("压缩改名成功(规范件也写进去了)", pr.get("code") == 0, f"{pr}")
|
||||
|
||||
print("\n[2] 签名链接")
|
||||
u = call("GET", f"/api/v1/mp/files/{fid}/url", tok)
|
||||
if not check("取到链接", u.get("code") == 0, f"{u}"):
|
||||
return 1
|
||||
d = u.get("data") or {}
|
||||
signed = d.get("url") or ""
|
||||
print(f" {signed[:120]}{'…' if len(signed) > 120 else ''}")
|
||||
check("provider 是 qiniu(不是本地磁盘)", d.get("provider") == "qiniu", f"{d.get('provider')}")
|
||||
check("链接指向配置的域名", (cfg.get("domain") or "") in signed, f"domain={cfg.get('domain')}")
|
||||
check("是**签名**链接(带 e= 与 token=)",
|
||||
"e=" in signed and "token=" in signed, "没有签名参数,私有空间会 403")
|
||||
check("对象键带目录前缀", f"/{cfg.get('prefix')}/" in signed, f"prefix={cfg.get('prefix')}")
|
||||
check("有效期是正数秒", (d.get("expiresIn") or 0) > 0, f"{d.get('expiresIn')}")
|
||||
|
||||
print("\n[3] 真的下载回来比一遍(这才是证明)")
|
||||
status, body = http_get(signed)
|
||||
check("签名链接返回 200", status == 200, f"HTTP {status}")
|
||||
if status == 200:
|
||||
check("下载到的内容非空", len(body) > 0, f"{len(body)} 字节")
|
||||
# 规范件是压缩过的,与上传的原图**不一定**等大(这是正常的);
|
||||
# 这里比的是"取回来的确实是一张 JPEG 图"
|
||||
check("取回的是 JPEG(FFD8 开头)",
|
||||
body[:2] == b"\xff\xd8", f"前两字节 {body[:2]!r}")
|
||||
|
||||
print("\n[4] 不带签名应当被拒(证明空间确实是私有的)")
|
||||
unsigned = signed.split("?")[0]
|
||||
st2, _ = http_get(unsigned)
|
||||
check("裸地址被拒(401/403)", st2 in (401, 403), f"HTTP {st2} —— 如果返回 200 说明空间是公开的")
|
||||
|
||||
print("\n[5] 原始件也取得回来(原始件绝不能丢)")
|
||||
u2 = call("GET", f"/api/v1/mp/files/{fid}/url?original=true", tok)
|
||||
signed2 = ((u2.get("data") or {}).get("url")) or ""
|
||||
st3, body3 = http_get(signed2)
|
||||
check("原始件签名链接返回 200", st3 == 200, f"HTTP {st3}")
|
||||
if st3 == 200:
|
||||
check("原始件字节与上传的**一模一样**",
|
||||
body3 == img, f"上传 {len(img)} 字节 / 取回 {len(body3)} 字节")
|
||||
finally:
|
||||
print("\n[6] 清理")
|
||||
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}")
|
||||
ok("已清理草稿与名册绑定")
|
||||
print("""
|
||||
⚠️ 对象存储里的文件**没有删** —— 那正是要的:原始件留着备查,
|
||||
认定机构退件时要能回答"是不是你们压坏的"。要清就从七牛控制台按前缀清。""")
|
||||
|
||||
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())
|
||||
Reference in New Issue
Block a user