c483018cc8
- scripts/patch_env.py：把 KEY=VALUE 文件合并进 .env，只回显键名与长度，不回显密钥 - scripts/check_s4.py：在生产机上验证「预签名 → PUT → 公开读回」整条直传链路 - .env.example：注明本地后端连生产库时 HASHID_SECRET 必须与线上一致 （不一致会把线上 brand_uid 解码成另一个 id，任务被判 brand not found）
95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
||
"""在生产机上验证 S4 直传链路：签发预签名 URL → PUT 一个小文件 → 公开读回。
|
||
|
||
只输出状态与长度，不输出密钥。
|
||
用法：python3 check_s4.py
|
||
"""
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import secrets
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
|
||
ENV_PATH = "/opt/fashion/backend_v2/.env"
|
||
ENDPOINT = "http://127.0.0.1:18092/admin/internal/upload-url"
|
||
|
||
|
||
def load_env(path):
|
||
env = {}
|
||
with open(path, encoding="utf-8") as fh:
|
||
for line in fh:
|
||
line = line.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
key, value = line.split("=", 1)
|
||
env[key.strip()] = value
|
||
return env
|
||
|
||
|
||
def main():
|
||
env = load_env(ENV_PATH)
|
||
secret = env.get("INGEST_SECRET", "")
|
||
bucket = env.get("S4_BUCKET", "")
|
||
if not secret:
|
||
print("!! .env 里没有 INGEST_SECRET")
|
||
return 1
|
||
|
||
key = f"runway/s4test-{int(time.time())}.jpg"
|
||
body = json.dumps({"items": [{"key": key, "content_type": "image/jpeg"}]}, separators=(",", ":"))
|
||
ts = str(int(time.time()))
|
||
nonce = secrets.token_hex(8)
|
||
sig = hmac.new(secret.encode(), f"{ts}.{nonce}.{body}".encode(), hashlib.sha256).hexdigest()
|
||
|
||
req = urllib.request.Request(
|
||
ENDPOINT,
|
||
data=body.encode(),
|
||
method="POST",
|
||
headers={
|
||
"Content-Type": "application/json",
|
||
"X-Signature": sig,
|
||
"X-Timestamp": ts,
|
||
"X-Nonce": nonce,
|
||
},
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=20) as resp:
|
||
print(f"1) 签发预签名: HTTP {resp.status}")
|
||
data = json.loads(resp.read())
|
||
except urllib.error.HTTPError as exc:
|
||
print(f"1) 签发预签名失败: HTTP {exc.code} {exc.read()[:200]!r}")
|
||
return 1
|
||
|
||
item = data["items"][0]
|
||
# 预签名 URL 里带签名参数，只显示 host 与 path
|
||
parsed = urllib.request.urlparse(item["url"])
|
||
print(f" 签名 URL: {parsed.scheme}://{parsed.netloc}{parsed.path[:40]}... 长度={len(item['url'])}")
|
||
|
||
payload = b"\xff\xd8\xff\xe0" + b"fashion-s4-probe" + b"\xff\xd9"
|
||
put = urllib.request.Request(
|
||
item["url"], data=payload, method="PUT", headers={"Content-Type": "image/jpeg"}
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(put, timeout=30) as resp:
|
||
print(f"2) 上传: HTTP {resp.status}")
|
||
except urllib.error.HTTPError as exc:
|
||
print(f"2) 上传失败: HTTP {exc.code} {exc.read()[:300]!r}")
|
||
return 1
|
||
|
||
public_url = f"https://{bucket}.s3.bitiful.net/{key}"
|
||
try:
|
||
with urllib.request.urlopen(public_url, timeout=20) as resp:
|
||
got = resp.read()
|
||
print(f"3) 公开读取: HTTP {resp.status} 字节={len(got)} 一致={got == payload}")
|
||
except urllib.error.HTTPError as exc:
|
||
print(f"3) 公开读取失败: HTTP {exc.code}（桶可能不是公开读）")
|
||
return 1
|
||
|
||
print(f" 测试对象: {key}（{len(payload)} 字节，可忽略）")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|