xiaotutongxue/scripts/dev-clean-qiniu.py

#!/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())