Leo
c2638668e3
leo 2026-08-20：「同一個 plugin 你用，薄殼也用，保證兩邊同步」
「我要你幫雲端做薄殼，永遠都有問題，你要做的就是這組設定
你自己可以 dogfooding」
搬進來：41 支 hook（51 條註冊）／7 支 command／2 支 skill／23 支腳本。
不搬 .env、wiki、docs——那些是知識不是環境。
51 條 hook 路徑全部從 $CLAUDE_PROJECT_DIR/.claude/hooks/ 改成 ${CLAUDE_PLUGIN_ROOT}/hooks/，
零漏網。那正是薄殼一直壞掉的根：雲端 cwd 不是真身，寫死路徑就斷。
尚未驗證：Claude Code 能不能從私有 Gitea repo 裝 marketplace（要憑證）。
下一步就是在本機實際裝一次，通了才動雲端 bootstrap.sh。
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
395 lines
16 KiB
Python
395 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
kv-generation-merge.py — 兩代 KV 櫃子的「逐把合併」工具（唯讀優先、絕不刪除、可回滾）
|
||
|
||
背景：arcrun-rag 安裝器對「帳號裡本來就有庫」的實例會另開一整套 KV
|
||
（`arcrun-rag-<suffix>-kv-*`），把 worker 綁到新櫃子，舊櫃子原封不動但沒人讀。
|
||
本工具處理「決定用哪一代 ＋ 把另一代獨有的東西搬過來 ＋ 把 worker 綁定指到同一代」。
|
||
|
||
設計硬規則
|
||
1. **永遠不刪除任何 key、任何 namespace。**（沒有 delete 子指令，寫不出來）
|
||
2. **預設不覆蓋**已存在的 key；要覆蓋必須用 `--overwrite` 逐把點名。
|
||
3. **可重跑**：copy 是冪等的（值相同就跳過），中途死掉重跑即可，不會產生半套狀態。
|
||
4. **綁定先快照再改**：`bindings-snapshot` 產生的 JSON 就是 rollback 的唯一依據。
|
||
5. **不印任何值**：診斷只印 key 名、長度、sha256 前 12 碼。
|
||
|
||
用法（token 只用「環境變數名字」傳入，值不進指令列——D36）
|
||
export CF_TOKEN_ENV=CLOUDFLARE_API_TOKEN_leo21c
|
||
export CF_ACCOUNT=51a01bfa2665bd7bc3fd080dc40cf3e1
|
||
|
||
# 1) 盤點：兩代逐把 key 級比對（唯讀）
|
||
python3 kv-generation-merge.py diff --old <ns_id_old> --new <ns_id_new>
|
||
|
||
# 2) 搬 key（只搬點名的那幾把；不點名什麼都不做）
|
||
python3 kv-generation-merge.py copy --from <ns_src> --to <ns_dst> --key K1 --key K2
|
||
# 要覆蓋目標已存在的 key（例：把新版 notify_leo 蓋進舊櫃）
|
||
python3 kv-generation-merge.py copy --from <ns_src> --to <ns_dst> --key K --overwrite
|
||
|
||
# 3) 綁定：先快照，再逐把重指，壞了用快照還原
|
||
python3 kv-generation-merge.py bindings-snapshot --script arcrun-cypher-executor -o snap.json
|
||
python3 kv-generation-merge.py bindings-repoint --script arcrun-cypher-executor \
|
||
--set WEBHOOKS=<ns_id> --set RECIPES=<ns_id> ...
|
||
python3 kv-generation-merge.py bindings-rollback --script arcrun-cypher-executor -i snap.json
|
||
|
||
# 4) 驗收：綁定實況 + 每把櫃子的內容統計
|
||
python3 kv-generation-merge.py audit --script arcrun-cypher-executor --script arcrun-registry ...
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
API = "https://api.cloudflare.com/client/v4"
|
||
|
||
|
||
def _token() -> str:
|
||
name = os.environ.get("CF_TOKEN_ENV")
|
||
if not name:
|
||
sys.exit("需要 CF_TOKEN_ENV=<存放 token 的環境變數名字>（只傳名字，不傳值）")
|
||
val = os.environ.get(name)
|
||
if not val:
|
||
sys.exit(f"環境變數 {name} 是空的")
|
||
return val
|
||
|
||
|
||
def _account() -> str:
|
||
acc = os.environ.get("CF_ACCOUNT")
|
||
if not acc:
|
||
sys.exit("需要 CF_ACCOUNT=<cloudflare account id>（明示，絕不靠 CLOUDFLARE_ACCOUNT_ID 預設值）")
|
||
return acc
|
||
|
||
|
||
def req(method: str, path: str, body=None, raw=False, retries=3):
|
||
url = f"{API}/accounts/{_account()}{path}"
|
||
data = None
|
||
headers = {"Authorization": f"Bearer {_token()}"}
|
||
if body is not None:
|
||
data = body if isinstance(body, bytes) else json.dumps(body).encode()
|
||
headers["Content-Type"] = "application/json"
|
||
last = None
|
||
for attempt in range(retries):
|
||
r = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||
try:
|
||
with urllib.request.urlopen(r, timeout=60) as resp:
|
||
payload = resp.read()
|
||
return payload if raw else json.loads(payload)
|
||
except Exception as e: # noqa: BLE001 — 網路類錯誤一律重試
|
||
last = e
|
||
body_txt = ""
|
||
if hasattr(e, "read"):
|
||
try:
|
||
body_txt = e.read().decode()[:400]
|
||
except Exception:
|
||
pass
|
||
if attempt == retries - 1:
|
||
sys.exit(f"CF API {method} {path} 失敗：{e} {body_txt}")
|
||
time.sleep(1.5 * (attempt + 1))
|
||
raise AssertionError(last)
|
||
|
||
|
||
def list_keys(ns: str) -> list[dict]:
|
||
out, cursor = [], ""
|
||
while True:
|
||
q = f"?limit=1000{'&cursor=' + urllib.parse.quote(cursor) if cursor else ''}"
|
||
d = req("GET", f"/storage/kv/namespaces/{ns}/keys{q}")
|
||
if not d.get("success"):
|
||
sys.exit(f"列 key 失敗：{d.get('errors')}")
|
||
out += d["result"]
|
||
cursor = (d.get("result_info") or {}).get("cursor") or ""
|
||
if not cursor:
|
||
return out
|
||
|
||
|
||
def get_value(ns: str, key: str) -> bytes | None:
|
||
try:
|
||
return req("GET", f"/storage/kv/namespaces/{ns}/values/{urllib.parse.quote(key, safe='')}",
|
||
raw=True, retries=2)
|
||
except SystemExit:
|
||
return None
|
||
|
||
|
||
def put_value(ns: str, key: str, value: bytes, expiration: int | None, metadata) -> None:
|
||
# multipart/form-data：CF KV 的 value+metadata 只吃這一種
|
||
boundary = "----arcrunkvmerge" + hashlib.sha256(key.encode()).hexdigest()[:16]
|
||
parts = []
|
||
|
||
def field(name, content: bytes, ctype=None):
|
||
h = f'--{boundary}\r\nContent-Disposition: form-data; name="{name}"\r\n'
|
||
if ctype:
|
||
h += f"Content-Type: {ctype}\r\n"
|
||
parts.append(h.encode() + b"\r\n" + content + b"\r\n")
|
||
|
||
field("value", value)
|
||
field("metadata", json.dumps(metadata or {}).encode(), "application/json")
|
||
parts.append(f"--{boundary}--\r\n".encode())
|
||
payload = b"".join(parts)
|
||
|
||
q = f"?expiration={expiration}" if expiration else ""
|
||
url = (f"{API}/accounts/{_account()}/storage/kv/namespaces/{ns}"
|
||
f"/values/{urllib.parse.quote(key, safe='')}{q}")
|
||
r = urllib.request.Request(
|
||
url, data=payload, method="PUT",
|
||
headers={"Authorization": f"Bearer {_token()}",
|
||
"Content-Type": f"multipart/form-data; boundary={boundary}"})
|
||
with urllib.request.urlopen(r, timeout=60) as resp:
|
||
d = json.loads(resp.read())
|
||
if not d.get("success"):
|
||
sys.exit(f"寫入失敗 {key}：{d.get('errors')}")
|
||
|
||
|
||
def sha(b: bytes | None) -> str:
|
||
return "-" if b is None else hashlib.sha256(b).hexdigest()[:12]
|
||
|
||
|
||
def patch_settings(script: str, settings: dict) -> dict:
|
||
"""PATCH /workers/scripts/{name}/settings —— 只改 metadata，不重傳程式碼。
|
||
|
||
⚠️ 這支端點**只吃 multipart/form-data**（送 application/json 會回 415 / code 10001）。
|
||
這是 2026-08-10 在 stage 演練時實撞出來的，別改回 json。
|
||
"""
|
||
boundary = "----arcrunsettings" + hashlib.sha256(script.encode()).hexdigest()[:16]
|
||
payload = (f'--{boundary}\r\nContent-Disposition: form-data; name="settings"\r\n'
|
||
f"Content-Type: application/json\r\n\r\n").encode()
|
||
payload += json.dumps(settings).encode() + f"\r\n--{boundary}--\r\n".encode()
|
||
url = f"{API}/accounts/{_account()}/workers/scripts/{script}/settings"
|
||
r = urllib.request.Request(
|
||
url, data=payload, method="PATCH",
|
||
headers={"Authorization": f"Bearer {_token()}",
|
||
"Content-Type": f"multipart/form-data; boundary={boundary}"})
|
||
try:
|
||
with urllib.request.urlopen(r, timeout=60) as resp:
|
||
return json.loads(resp.read())
|
||
except Exception as e: # noqa: BLE001
|
||
detail = e.read().decode()[:500] if hasattr(e, "read") else ""
|
||
sys.exit(f"改綁定失敗：{e} {detail}")
|
||
|
||
|
||
# ── 子指令 ────────────────────────────────────────────────────────────────────
|
||
|
||
def cmd_diff(a):
|
||
o = {k["name"]: k for k in list_keys(a.old)}
|
||
n = {k["name"]: k for k in list_keys(a.new)}
|
||
both = sorted(set(o) & set(n))
|
||
print(f"舊 {len(o)} 把｜新 {len(n)} 把｜共有 {len(both)}｜只在舊 {len(set(o)-set(n))}｜只在新 {len(set(n)-set(o))}")
|
||
print("\n[只在舊櫃子]")
|
||
for k in sorted(set(o) - set(n)):
|
||
print(" +old", k)
|
||
print("\n[只在新櫃子]")
|
||
for k in sorted(set(n) - set(o)):
|
||
print(" +new", k)
|
||
print("\n[兩代都有 → 逐把比對內容]")
|
||
same = diff = 0
|
||
for k in both:
|
||
vo, vn = get_value(a.old, k), get_value(a.new, k)
|
||
if sha(vo) == sha(vn):
|
||
same += 1
|
||
if a.verbose:
|
||
print(f" = {k} {sha(vo)}")
|
||
else:
|
||
diff += 1
|
||
print(f" ≠ {k} old={sha(vo)}({len(vo or b'')}B) new={sha(vn)}({len(vn or b'')}B)")
|
||
print(f"\n共有的 {len(both)} 把：內容相同 {same}、**內容不同 {diff}**（不同的那些才是會咬人的）")
|
||
|
||
|
||
def cmd_copy(a):
|
||
src_keys = {k["name"]: k for k in list_keys(getattr(a, "from"))}
|
||
dst_keys = {k["name"]: k for k in list_keys(a.to)}
|
||
todo, skip = [], []
|
||
for k in a.key:
|
||
if k not in src_keys:
|
||
sys.exit(f"來源櫃子沒有這把 key：{k}（先跑 diff 對一次名字）")
|
||
if k in dst_keys:
|
||
vs, vd = get_value(getattr(a, "from"), k), get_value(a.to, k)
|
||
if sha(vs) == sha(vd):
|
||
skip.append((k, "目標已有且內容相同"))
|
||
continue
|
||
if not a.overwrite:
|
||
skip.append((k, "目標已有且內容不同 → 需 --overwrite 才動"))
|
||
continue
|
||
todo.append(k)
|
||
for k, why in skip:
|
||
print(f" 跳過 {k}：{why}")
|
||
if a.dry_run:
|
||
for k in todo:
|
||
print(f" [dry-run] 會寫入 {k}")
|
||
print(f"\ndry-run：會寫 {len(todo)} 把、跳過 {len(skip)} 把。加 --apply 才真的寫。")
|
||
return
|
||
for k in todo:
|
||
meta = src_keys[k]
|
||
v = get_value(getattr(a, "from"), k)
|
||
if v is None:
|
||
sys.exit(f"讀不到來源值：{k}（中止，已寫入的部分不受影響，重跑即可續）")
|
||
put_value(a.to, k, v, meta.get("expiration"), meta.get("metadata"))
|
||
print(f" 寫入 {k} {sha(v)} ({len(v)}B)")
|
||
# 立即回讀複驗
|
||
bad = []
|
||
for k in todo:
|
||
if sha(get_value(getattr(a, "from"), k)) != sha(get_value(a.to, k)):
|
||
bad.append(k)
|
||
print(f"\n寫入 {len(todo)} 把、跳過 {len(skip)} 把；回讀複驗不符 {len(bad)} 把 {bad}")
|
||
if bad:
|
||
sys.exit(1)
|
||
|
||
|
||
def _script_settings(script: str):
|
||
d = req("GET", f"/workers/scripts/{script}/settings")
|
||
if not d.get("success"):
|
||
sys.exit(f"讀 {script} settings 失敗：{d.get('errors')}")
|
||
return d["result"]
|
||
|
||
|
||
def cmd_bindings_snapshot(a):
|
||
snap = {}
|
||
for s in a.script:
|
||
snap[s] = _script_settings(s)
|
||
out = json.dumps(snap, ensure_ascii=False, indent=2)
|
||
if a.out:
|
||
open(a.out, "w", encoding="utf-8").write(out)
|
||
print(f"快照已存 → {a.out}")
|
||
else:
|
||
print(out)
|
||
for s, st in snap.items():
|
||
for b in st.get("bindings", []):
|
||
if b.get("type") == "kv_namespace":
|
||
print(f" {s:26} {b['name']:16} {b['namespace_id']}")
|
||
|
||
|
||
def _sanitize(bindings: list[dict]) -> list[dict]:
|
||
"""把讀回來的 bindings 變成可以寫回去的形狀：secret 用 inherit（不碰真身，D36）。"""
|
||
out = []
|
||
for b in bindings:
|
||
t = b.get("type")
|
||
if t in ("secret_text", "secret_key"):
|
||
out.append({"type": "inherit", "name": b["name"]})
|
||
else:
|
||
out.append({k: v for k, v in b.items() if v is not None})
|
||
return out
|
||
|
||
|
||
def cmd_bindings_repoint(a):
|
||
want = dict(s.split("=", 1) for s in a.set)
|
||
st = _script_settings(a.script)
|
||
bindings = st.get("bindings", [])
|
||
names = {b["name"] for b in bindings if b.get("type") == "kv_namespace"}
|
||
missing = set(want) - names
|
||
if missing:
|
||
sys.exit(f"{a.script} 上沒有這些 KV 綁定名：{sorted(missing)}")
|
||
new_bindings, changes = [], []
|
||
for b in _sanitize(bindings):
|
||
if b.get("type") == "kv_namespace" and b["name"] in want:
|
||
old = b["namespace_id"]
|
||
if old != want[b["name"]]:
|
||
changes.append((b["name"], old, want[b["name"]]))
|
||
b = {**b, "namespace_id": want[b["name"]]}
|
||
new_bindings.append(b)
|
||
for n, o, w in changes:
|
||
print(f" {a.script}: {n} {o} → {w}")
|
||
if not changes:
|
||
print(" （沒有任何綁定需要變更——已經是目標狀態，冪等）")
|
||
return
|
||
if a.dry_run:
|
||
print("\ndry-run：加 --apply 才真的改。")
|
||
return
|
||
keep = ("compatibility_date", "compatibility_flags", "logpush", "placement",
|
||
"tail_consumers", "observability", "limits", "migrations")
|
||
settings = {k: v for k, v in st.items() if k in keep and v is not None}
|
||
settings["bindings"] = new_bindings
|
||
d = patch_settings(a.script, settings)
|
||
if not d.get("success"):
|
||
sys.exit(f"改綁定失敗：{d.get('errors')}")
|
||
after = {b["name"]: b["namespace_id"] for b in _script_settings(a.script).get("bindings", [])
|
||
if b.get("type") == "kv_namespace"}
|
||
bad = [n for n, v in want.items() if after.get(n) != v]
|
||
print(f"\n改完複驗：{'✅ 全部到位' if not bad else '❌ 沒到位 ' + str(bad)}")
|
||
if bad:
|
||
sys.exit(1)
|
||
|
||
|
||
def cmd_bindings_rollback(a):
|
||
snap = json.load(open(a.inp, encoding="utf-8"))
|
||
st = snap.get(a.script)
|
||
if not st:
|
||
sys.exit(f"快照裡沒有 {a.script}")
|
||
want = {b["name"]: b["namespace_id"] for b in st["bindings"] if b.get("type") == "kv_namespace"}
|
||
ns = argparse.Namespace(script=a.script, set=[f"{k}={v}" for k, v in want.items()],
|
||
dry_run=a.dry_run)
|
||
cmd_bindings_repoint(ns)
|
||
|
||
|
||
def cmd_audit(a):
|
||
ns_title = {}
|
||
d = req("GET", "/storage/kv/namespaces?per_page=100")
|
||
for n in d.get("result", []):
|
||
ns_title[n["id"]] = n["title"]
|
||
seen = {}
|
||
for s in a.script:
|
||
st = _script_settings(s)
|
||
kv = [b for b in st.get("bindings", []) if b.get("type") == "kv_namespace"]
|
||
if not kv:
|
||
continue
|
||
print(f"\n{s}")
|
||
for b in kv:
|
||
title = ns_title.get(b["namespace_id"], "?")
|
||
gen = "新" if title.startswith("arcrun-rag-") else "舊"
|
||
cnt = len(list_keys(b["namespace_id"]))
|
||
print(f" {b['name']:16} {b['namespace_id']} [{gen}] {title:38} {cnt:>4} 把")
|
||
seen.setdefault(b["name"], set()).add(b["namespace_id"])
|
||
print("\n── 分裂檢查（同一個綁定名被指到不同 namespace = 分裂）──")
|
||
split = {k: v for k, v in seen.items() if len(v) > 1}
|
||
print("✅ 沒有分裂：所有 worker 的同名綁定都指到同一顆" if not split else f"❌ 分裂：{split}")
|
||
if split:
|
||
sys.exit(1)
|
||
|
||
|
||
def main():
|
||
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
sub = p.add_subparsers(dest="cmd", required=True)
|
||
|
||
d = sub.add_parser("diff", help="兩代逐把 key 級比對（唯讀）")
|
||
d.add_argument("--old", required=True)
|
||
d.add_argument("--new", required=True)
|
||
d.add_argument("--verbose", action="store_true")
|
||
d.set_defaults(func=cmd_diff)
|
||
|
||
c = sub.add_parser("copy", help="把點名的 key 從一個 namespace 搬到另一個（冪等、預設不覆蓋、永不刪除）")
|
||
c.add_argument("--from", required=True, dest="from")
|
||
c.add_argument("--to", required=True)
|
||
c.add_argument("--key", action="append", required=True)
|
||
c.add_argument("--overwrite", action="store_true", help="目標已有且內容不同時才需要，逐把點名")
|
||
c.add_argument("--apply", dest="dry_run", action="store_false", default=True)
|
||
c.set_defaults(func=cmd_copy)
|
||
|
||
s = sub.add_parser("bindings-snapshot", help="把 worker 現況綁定存成 JSON（rollback 的唯一依據）")
|
||
s.add_argument("--script", action="append", required=True)
|
||
s.add_argument("-o", "--out")
|
||
s.set_defaults(func=cmd_bindings_snapshot)
|
||
|
||
r = sub.add_parser("bindings-repoint", help="把某個 worker 的 KV 綁定改指到別顆 namespace（不動程式碼、不動 secret）")
|
||
r.add_argument("--script", required=True)
|
||
r.add_argument("--set", action="append", required=True, help="BINDING_NAME=namespace_id")
|
||
r.add_argument("--apply", dest="dry_run", action="store_false", default=True)
|
||
r.set_defaults(func=cmd_bindings_repoint)
|
||
|
||
b = sub.add_parser("bindings-rollback", help="用快照把綁定還原")
|
||
b.add_argument("--script", required=True)
|
||
b.add_argument("-i", "--inp", required=True)
|
||
b.add_argument("--apply", dest="dry_run", action="store_false", default=True)
|
||
b.set_defaults(func=cmd_bindings_rollback)
|
||
|
||
a = sub.add_parser("audit", help="列出每顆 worker 綁到哪一代、各幾把 key，並檢查有沒有分裂")
|
||
a.add_argument("--script", action="append", required=True)
|
||
a.set_defaults(func=cmd_audit)
|
||
|
||
args = p.parse_args()
|
||
args.func(args)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|