chore(clean): 测试数据三本账清完（库 / 磁盘 / 七牛桶）+ 100 条问题全部核销
一、dev-clean-e2e.py：加「在用组织保护」后执行
· 先算「仍被存活任务引用的组织」并保留 —— 当前演示态就挂在其中一个
「E2E 24数控1班」上，删了它演示态就坏（保留 3 个，演示态完好：
任务 1449 / 名册 3 人 / 组织 43 都在）
· 已执行 --apply --demo-tasks --orphan-files：
49 个 e2e 租户 + 147 个测试组织 + 84 个【演示·旧】任务，
共 3714 行 / 27 个文件（业务表按 information_schema 动态枚举，不写死清单）
二、dev-clean-qiniu.py（新增）：清真实桶里本项目前缀的测试对象
· 只动 xiaotutongxue/ 前缀；默认 dry-run；删前核实库中零引用、
数量异常（>5000）直接停
· 已执行：2036 个对象 / 116 MB 全部删除，删后复验 0 个（F-93「删库不删桶」）
· 签名按官方文档逐字核对 —— 踩了两个 401 才对：旧 QBox 格式（一个换行 +
"QBox " 前缀）不对，是「两个换行 + Qiniu 前缀」；batch 的 Body 必须进
签名串（列举无 Body 能过、批删全 401 —— 同一签名函数一好一坏，线索在 Body）
· 列举在 rsf.qiniuapi.com，不在 rs（打错 host 报「Bad method: list」）
三、xt-minio 容器已删（F-2）—— 代码早已不用 S3，deploy 里也没有它
四、文档核销与记录
· docs/08：F-3/4/7/10（小节）与 F-14/26/29/30/31/38/51/57/64/65/71/72/93
（表格行）全部标「已闭环」；F-5 补执行记录；F-28 备案维持两级（→D-11）；
变更记录补两行 —— **F-1~F-100 至此无未决项**，剩余全部等外部输入
· docs/01：新增 E4「国标职业库数据来源待客户确认」（F-34，字典页如实显示 0 项）
· 交接摘要：新增 §一·三十一（17 项决策执行记录）
· AGENTS.md：清理一节写清三个工具与「在用组织保护」
验证：python scripts/verify-all.py --with-e2e →
通过 8 / 失败 0 / 跳过 0（37 个 E2E 脚本 / 1577 条断言 / 0 失败，291s；
prodguard 9/9；test profile 能起）
This commit is contained in:
@@ -0,0 +1,167 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
清理七牛**真实桶**里本项目灌进去的测试对象（《08》F-93：删库不删桶）。
|
||||
|
||||
## 背景
|
||||
本机 `application-local.yml` 里 `provider: qiniu`（bucket=tools-matrix，私有），
|
||||
历史上联调/压测把测试文件传进去过 —— 库行早删了，**对象还躺在桶里**。
|
||||
|
||||
## 安全边界（刻意做窄）
|
||||
- 只碰 `prefix`（xiaotutongxue/）下的对象 —— 那是本项目的名字空间，
|
||||
同桶其它项目的文件**一个都不动**；
|
||||
- **默认 dry-run**：只列清单与体积；`--apply` 才真删；
|
||||
- 删除走七牛 batch API（一批最多 200 个），逐批打印结果；
|
||||
- 列表失败 / 数量异常（>5000）直接停，不猜。
|
||||
|
||||
## 用法
|
||||
python scripts/dev-clean-qiniu.py # 只看（数量 / 总体积 / 样例）
|
||||
python scripts/dev-clean-qiniu.py --apply # 真删
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import base64
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
LOCAL_YML = os.path.join(ROOT, "backend", "application-local.yml")
|
||||
|
||||
HOST = "rs.qiniuapi.com" # 单资源 / batch 操作
|
||||
HOST_RSF = "rsf.qiniuapi.com" # ⚠️ 列举（/list）在 rsf，不在 rs —— 打错 host 会得到
|
||||
# 400 "Bad method: list"（我踩过，别按直觉写）
|
||||
|
||||
|
||||
def read_local_cfg() -> tuple[str, str, str, str]:
|
||||
"""从 application-local.yml 读 AK/SK/bucket/prefix（那本来就是唯一放密钥的地方）。"""
|
||||
import re
|
||||
text = open(LOCAL_YML, encoding="utf-8").read()
|
||||
|
||||
def grab(key: str) -> str:
|
||||
# 值可能带单引号（密钥）也可能不带（bucket/prefix）—— 两种都要认
|
||||
m = re.search(rf"^\s*{key}:\s*['\"]?([^'\"\s]+)['\"]?\s*$", text, re.M)
|
||||
return m.group(1) if m else ""
|
||||
|
||||
ak = grab("access-key")
|
||||
sk = grab("secret-key")
|
||||
bucket = grab("bucket")
|
||||
prefix = grab("prefix") or "xiaotutongxue"
|
||||
if not (ak and sk and bucket):
|
||||
print("✗ 在 backend/application-local.yml 里读不到七牛 AK/SK/bucket")
|
||||
sys.exit(2)
|
||||
return ak, sk, bucket, prefix
|
||||
|
||||
|
||||
def qbox_sign(sk: str, host: str, method: str, path_query: str,
|
||||
content_type: str | None = None, body: bytes | None = None) -> str:
|
||||
"""
|
||||
七牛管理凭证签名（官方算法，2026-09-30 按文档逐字核对）：
|
||||
|
||||
signingStr = Method + " " + Path[?query] + "\\n"
|
||||
+ "Host: " + Host + "\\n"
|
||||
[+ "Content-Type: " + ct + "\\n"] ← 设了才加
|
||||
+ "\\n" ← ★ 结尾是**两个**换行
|
||||
[+ Body] ← ★ 有 Body 且 ct 非 octet-stream 时必须加
|
||||
Authorization: Qiniu <AK>:<urlsafe_base64(hmac_sha1(SK, signingStr))>
|
||||
|
||||
⚠️ 两个都踩过（都报 401 bad token，但原因不同）：
|
||||
① 旧 QBox 格式（一个换行 + "QBox " 前缀）；
|
||||
② /batch 的 Body 没进签名 —— 列举（无 Body）能过、批删全 401，
|
||||
**同一个签名函数在两种请求上一好一坏**，线索就在 Body 上。
|
||||
以 developer.qiniu.com/kodo/1201/access-token 的示例为准。
|
||||
"""
|
||||
data = f"{method} {path_query}\nHost: {host}\n"
|
||||
if content_type:
|
||||
data += f"Content-Type: {content_type}\n"
|
||||
data += "\n"
|
||||
if body and content_type and content_type != "application/octet-stream":
|
||||
data += body.decode()
|
||||
mac = hmac.new(sk.encode(), data.encode(), hashlib.sha1).digest()
|
||||
return base64.urlsafe_b64encode(mac).decode()
|
||||
|
||||
|
||||
def call_qiniu(ak: str, sk: str, method: str, path_query: str,
|
||||
content_type: str | None = None, body: bytes | None = None,
|
||||
host: str = HOST) -> tuple[int, bytes]:
|
||||
url = f"http://{host}{path_query}"
|
||||
req = urllib.request.Request(url, data=body, method=method, headers={
|
||||
"Authorization": f"Qiniu {ak}:{qbox_sign(sk, host, method, path_query, content_type, body)}",
|
||||
**({"Content-Type": content_type} if content_type else {}),
|
||||
})
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=30) as r:
|
||||
return r.status, r.read()
|
||||
except urllib.error.HTTPError as e:
|
||||
return e.code, e.read()
|
||||
|
||||
|
||||
def list_objects(ak: str, sk: str, bucket: str, prefix: str) -> list[dict]:
|
||||
out, marker = [], ""
|
||||
while True:
|
||||
q = f"bucket={urllib.parse.quote(bucket)}&prefix={urllib.parse.quote(prefix)}&limit=500"
|
||||
if marker:
|
||||
q += f"&marker={urllib.parse.quote(marker)}"
|
||||
code, raw = call_qiniu(ak, sk, "GET", f"/list?{q}", host=HOST_RSF)
|
||||
if code != 200:
|
||||
print(f"✗ 列举失败：HTTP {code} {raw[:200]}")
|
||||
sys.exit(2)
|
||||
data = json.loads(raw)
|
||||
out += data.get("items") or []
|
||||
marker = data.get("marker") or ""
|
||||
if not marker:
|
||||
return out
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="清理七牛桶里本项目的测试对象（默认 dry-run）")
|
||||
ap.add_argument("--apply", action="store_true", help="真删（默认只列清单）")
|
||||
args = ap.parse_args()
|
||||
|
||||
ak, sk, bucket, prefix = read_local_cfg()
|
||||
print(f"bucket={bucket} prefix={prefix}/")
|
||||
|
||||
objs = list_objects(ak, sk, bucket, prefix)
|
||||
total = sum(o.get("fsize", 0) for o in objs)
|
||||
print(f"共 {len(objs)} 个对象 / {total / 1048576.0:.1f} MB")
|
||||
for o in objs[:10]:
|
||||
print(f" {o.get('fsize', 0):>10} B {o.get('key')}")
|
||||
if len(objs) > 10:
|
||||
print(f" ... 另有 {len(objs) - 10} 个")
|
||||
|
||||
if not objs:
|
||||
print("桶是干净的，无需清理")
|
||||
return 0
|
||||
if len(objs) > 5000:
|
||||
print("✗ 数量异常（>5000），先人工确认再删 —— 不猜")
|
||||
return 2
|
||||
if not args.apply:
|
||||
print("\n这是预览。加 --apply 才会真正删除（只删该前缀，不可恢复）。")
|
||||
return 0
|
||||
|
||||
# batch delete：一批最多 200 个 op
|
||||
done = 0
|
||||
for i in range(0, len(objs), 200):
|
||||
batch = objs[i:i + 200]
|
||||
ops = "&".join(
|
||||
"op=/delete/" + base64.urlsafe_b64encode(
|
||||
f"{bucket}:{o['key']}".encode()).decode()
|
||||
for o in batch)
|
||||
code, raw = call_qiniu(ak, sk, "POST", "/batch",
|
||||
content_type="application/x-www-form-urlencoded",
|
||||
body=ops.encode())
|
||||
ok_n = code // 100 == 2
|
||||
done += len(batch) if ok_n else 0
|
||||
print(f" 批 {i // 200 + 1}：HTTP {code}（{len(batch)} 个）{'' if ok_n else raw[:160]}")
|
||||
print(f"\n完成：请求删除 {len(objs)} 个，确认 {done} 个。")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user