chore: 补 S4 启用与配置合并的运维脚本

- scripts/patch_env.py：把 KEY=VALUE 文件合并进 .env，只回显键名与长度，不回显密钥
- scripts/check_s4.py：在生产机上验证「预签名 → PUT → 公开读回」整条直传链路
- .env.example：注明本地后端连生产库时 HASHID_SECRET 必须与线上一致
  （不一致会把线上 brand_uid 解码成另一个 id，任务被判 brand not found）

.env.example

@@ -20,6 +20,9 @@ PG_CONTAINER=1Panel-postgresql-4Ndp
 
 # ---- 后端密钥（必填，缺了容器拒绝启动）----
 # 公开 ID 混淆盐：上线后不要再改，改了所有旧链接失效
+# ⚠️ 本地开发后端若连的是生产库（同一套 DB），这个值必须与线上完全一致。
+# 不一致时本地后端会把 crawl/ingest 里线上编码的 brand_uid 解码成另一个 id，
+# 任务被判「brand not found」——本地机器于是把生产队列里的任务做成失败。
 HASHID_SECRET=
 JWT_SECRET=
 # 爬虫上报 HMAC 密钥：须与爬虫端 config.yml 的 ingest.secret 完全一致

scripts/check_s4.py

@@ -0,0 +1,94 @@
+#!/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())

scripts/patch_env.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+"""把 KEY=VALUE 文件里的键合并进目标 .env（存在则替换，不存在则追加）。
+
+用途：在不把密钥回显到终端的前提下，更新服务器上的生产配置。
+用法：python3 patch_env.py <kv文件> <目标.env>
+"""
+import re
+import sys
+
+
+def main() -> int:
+    if len(sys.argv) != 3:
+        print("用法: patch_env.py <kv文件> <目标.env>")
+        return 2
+    kv_path, env_path = sys.argv[1], sys.argv[2]
+
+    kv = {}
+    with open(kv_path, encoding="utf-8") as fh:
+        for line in fh:
+            line = line.strip()
+            if not line or "=" not in line or line.startswith("#"):
+                continue
+            key, value = line.split("=", 1)
+            kv[key.strip()] = value
+
+    with open(env_path, encoding="utf-8") as fh:
+        lines = fh.read().splitlines()
+
+    seen = set()
+    out = []
+    for line in lines:
+        m = re.match(r"^([A-Z0-9_]+)=", line)
+        if m and m.group(1) in kv:
+            key = m.group(1)
+            out.append(f"{key}={kv[key]}")
+            seen.add(key)
+        else:
+            out.append(line)
+    for key, value in kv.items():
+        if key not in seen:
+            out.append(f"{key}={value}")
+
+    with open(env_path, "w", encoding="utf-8") as fh:
+        fh.write("\n".join(out) + "\n")
+
+    # 只打印键名与长度，绝不打印值
+    for key in sorted(kv):
+        print(f"  {key}: {'已更新' if key in seen else '已追加'} 长度={len(kv[key])}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())