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>
350 lines
16 KiB
Python
350 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
KBDB 按鈕殼 —— **受測者只拿得到這支腳本**。
|
||
|
||
leo 2026-08-15：「如果明天只有這幾個按鈕可按，至少保證不再出錯。」
|
||
⇒ 這支就是那幾個按鈕。它刻意**只**暴露使用者真的走得到的那條路上真的存在的動作
|
||
（portal 資料面 ＝ MCP 知識面工具打的同一組端點、同一道閘）。
|
||
沒有 update、沒有 delete/archive、沒有 link、沒有 add-field、沒有 add-record-to-sheet
|
||
——因為那條路上真的沒有（見 system-dev/docs/4-guides/kbdb-動作對照表.md）。
|
||
|
||
兩個後端，**寫入語意刻意寫成同一份**（mirror 自 matrix/arcrun/kbdb/src/actions/record-crud.ts）：
|
||
local:<path> 本機 sqlite，用來做判分器自我驗證與乾跑（**不碰任何實例**）
|
||
portal:<url> 真的打 youlin 的 portal 資料面（正式考試用）
|
||
|
||
⚠️ 下面兩句 CREATE TABLE 標了 kbdb-sql-ok：那是**本機拋棄式 sqlite 的空白畫布**，
|
||
不是 KBDB 的 D1，也沒有替 KBDB 新增任何表。乾跑用完即丟。
|
||
"""
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import subprocess
|
||
import uuid
|
||
|
||
SYS_ROOT, SYS_BELONGS, SYS_FIELD_OF = "sys_root", "sys_belongs", "sys_field_of"
|
||
|
||
# 本機空白畫布：0007 之後的 entries（含三根指標欄）＋ 遷移期仍在的 templates。
|
||
SCHEMA = """
|
||
CREATE TABLE IF NOT EXISTS entries ( -- kbdb-sql-ok: 本機拋棄式 sqlite 畫布，非 KBDB 的 D1
|
||
id TEXT PRIMARY KEY, content TEXT, entry_type TEXT NOT NULL, owner_id TEXT,
|
||
parent_id TEXT, page_name TEXT, refs_json TEXT DEFAULT '[]', tags_json TEXT DEFAULT '[]',
|
||
task_status TEXT, content_hash TEXT, is_embedded INTEGER DEFAULT 0,
|
||
confidence REAL, metadata_json TEXT,
|
||
created_at INTEGER DEFAULT (unixepoch()), updated_at INTEGER DEFAULT (unixepoch()),
|
||
src_id TEXT, rel_id TEXT, dst_id TEXT);
|
||
CREATE TABLE IF NOT EXISTS templates ( -- kbdb-sql-ok: 同上，本機畫布
|
||
id TEXT PRIMARY KEY, name TEXT UNIQUE NOT NULL, description TEXT,
|
||
slots_json TEXT NOT NULL, created_by TEXT,
|
||
created_at INTEGER DEFAULT (unixepoch()), updated_at INTEGER DEFAULT (unixepoch()));
|
||
"""
|
||
|
||
|
||
def uid(p):
|
||
return f"{p}_{uuid.uuid4()}"
|
||
|
||
|
||
# ══════════════════════════ 後端 A：本機 sqlite ══════════════════════════
|
||
# 這一段是 record-crud.ts 的逐句對照移植。**包括它的沉默**：
|
||
# createRecord 只寫 template 宣告過的 slot（`slots.filter(s => s in values)`），
|
||
# 沒宣告的 key **靜默消失**且仍回 200 —— 那正是最值得考出來的一種安靜的錯。
|
||
class Local:
|
||
def __init__(self, path):
|
||
self.db = sqlite3.connect(path)
|
||
self.db.row_factory = sqlite3.Row
|
||
self.db.executescript(SCHEMA)
|
||
self._anchors()
|
||
|
||
def _anchors(self):
|
||
for i, c in ((SYS_ROOT, "root"), (SYS_BELONGS, "belongs"), (SYS_FIELD_OF, "field_of")):
|
||
self.db.execute(
|
||
"INSERT OR IGNORE INTO entries (id, content, entry_type) VALUES (?,?, 'system')",
|
||
(i, c))
|
||
self.db.commit()
|
||
|
||
def _ensure_fields(self, tid, slots):
|
||
for s in slots:
|
||
fid = f"fld_{tid}_{s}"
|
||
self.db.execute(
|
||
"INSERT OR IGNORE INTO entries (id, content, entry_type) VALUES (?,?,'field')",
|
||
(fid, s))
|
||
self.db.execute(
|
||
"INSERT OR IGNORE INTO entries (id, entry_type, src_id, rel_id, dst_id) "
|
||
"VALUES (?, 'relation', ?, ?, ?)", (f"relf_{tid}_{s}", fid, SYS_FIELD_OF, tid))
|
||
|
||
def list_sheets(self):
|
||
return [dict(r) for r in self.db.execute(
|
||
"SELECT id, name, slots_json FROM templates ORDER BY created_at DESC")]
|
||
|
||
def create_sheet(self, name, fields, description=None):
|
||
tid = uid("tpl")
|
||
self.db.execute(
|
||
"INSERT INTO templates (id, name, description, slots_json) VALUES (?,?,?,?)",
|
||
(tid, name, description, json.dumps(fields, ensure_ascii=False)))
|
||
self.db.execute(
|
||
"INSERT OR IGNORE INTO entries (id, content, entry_type) VALUES (?,?,'sheet')",
|
||
(tid, name))
|
||
self.db.execute(
|
||
"INSERT OR IGNORE INTO entries (id, entry_type, src_id, rel_id, dst_id) "
|
||
"VALUES (?, 'relation', ?, ?, ?)", (f"relb_{tid}", tid, SYS_BELONGS, SYS_ROOT))
|
||
self._ensure_fields(tid, fields)
|
||
self.db.commit()
|
||
return {"id": tid, "name": name, "slots": fields}
|
||
|
||
def _tpl(self, name_or_id):
|
||
r = self.db.execute("SELECT * FROM templates WHERE id=? OR name=? LIMIT 1",
|
||
(name_or_id, name_or_id)).fetchone()
|
||
return dict(r) if r else None
|
||
|
||
def append_record(self, sheet, values, metadata_json=None):
|
||
tpl = self._tpl(sheet)
|
||
if not tpl:
|
||
return {"error": f"sheet not found: {sheet}"}
|
||
slots = json.loads(tpl["slots_json"])
|
||
rid = uid("rec")
|
||
self.db.execute("INSERT OR IGNORE INTO entries (id, entry_type) VALUES (?, 'record')", (rid,))
|
||
self.db.execute(
|
||
"INSERT OR IGNORE INTO entries (id, entry_type, src_id, rel_id, dst_id) "
|
||
"VALUES (?, 'relation', ?, ?, ?)",
|
||
(f"relb_{rid}_{tpl['id']}", rid, SYS_BELONGS, tpl["id"]))
|
||
written = [s for s in slots if s in values] # ← 沒宣告的 key 在這裡靜默消失
|
||
self._ensure_fields(tpl["id"], written)
|
||
for s in written:
|
||
eid = uid("e")
|
||
self.db.execute(
|
||
"INSERT INTO entries (id, content, entry_type, metadata_json) "
|
||
"VALUES (?,?, 'value', ?)", (eid, values[s], metadata_json))
|
||
self.db.execute(
|
||
"INSERT INTO entries (id, entry_type, src_id, rel_id, dst_id) "
|
||
"VALUES (?, 'relation', ?, ?, ?)",
|
||
(uid("relv"), rid, f"fld_{tpl['id']}_{s}", eid))
|
||
self.db.commit()
|
||
return {"record_id": rid, "sheet": tpl["name"],
|
||
"values": {s: values[s] for s in written}}
|
||
|
||
# ⚠️ 只給判分器自我驗證用的正控制組，**不對受測者開放**：
|
||
# 「指向既有的那一顆，而不是複製一份」是 KBDB 的核心賣點，
|
||
# kbdb/src 內部確實有這個通道（createRecord 的 entry_ids），
|
||
# 但 **portal 資料面與 MCP 都只轉送 {template, values}，沒有把它露出來**
|
||
# ⇒ 使用者那條路上做不到。這支方法存在的意義就是把那個洞量出來。
|
||
def _append_record_pointer(self, sheet, values, pointers):
|
||
tpl = self._tpl(sheet)
|
||
rid = uid("rec")
|
||
self.db.execute("INSERT OR IGNORE INTO entries (id, entry_type) VALUES (?, 'record')", (rid,))
|
||
self.db.execute(
|
||
"INSERT OR IGNORE INTO entries (id, entry_type, src_id, rel_id, dst_id) "
|
||
"VALUES (?, 'relation', ?, ?, ?)",
|
||
(f"relb_{rid}_{tpl['id']}", rid, SYS_BELONGS, tpl["id"]))
|
||
slots = json.loads(tpl["slots_json"])
|
||
self._ensure_fields(tpl["id"], slots)
|
||
for s in slots:
|
||
if s in pointers:
|
||
dst = pointers[s]
|
||
elif s in values:
|
||
dst = uid("e")
|
||
self.db.execute(
|
||
"INSERT INTO entries (id, content, entry_type) VALUES (?,?, 'value')",
|
||
(dst, values[s]))
|
||
else:
|
||
continue
|
||
self.db.execute(
|
||
"INSERT INTO entries (id, entry_type, src_id, rel_id, dst_id) "
|
||
"VALUES (?, 'relation', ?, ?, ?)",
|
||
(uid("relv"), rid, f"fld_{tpl['id']}_{s}", dst))
|
||
self.db.commit()
|
||
return {"record_id": rid}
|
||
|
||
def _make_shared_value(self, content):
|
||
eid = uid("e")
|
||
self.db.execute("INSERT INTO entries (id, content, entry_type) VALUES (?,?, 'value')",
|
||
(eid, content))
|
||
self.db.commit()
|
||
return eid
|
||
|
||
def get_record(self, rid):
|
||
rows = self.db.execute(
|
||
"SELECT f.content AS field, v.content AS value FROM entries r "
|
||
"JOIN entries f ON f.id = r.rel_id JOIN entries v ON v.id = r.dst_id "
|
||
"WHERE r.src_id = ? AND r.rel_id != ?", (rid, SYS_BELONGS)).fetchall()
|
||
return {"record_id": rid, "values": {r["field"]: r["value"] for r in rows}}
|
||
|
||
def get_records(self, sheet):
|
||
tpl = self._tpl(sheet)
|
||
if not tpl:
|
||
return []
|
||
ids = [r["src_id"] for r in self.db.execute(
|
||
"SELECT src_id FROM entries WHERE rel_id=? AND dst_id=?", (SYS_BELONGS, tpl["id"]))]
|
||
return [self.get_record(i) for i in ids]
|
||
|
||
def search(self, q):
|
||
return [dict(r) for r in self.db.execute(
|
||
"SELECT id, content, entry_type FROM entries WHERE content LIKE ? LIMIT 50",
|
||
(f"%{q}%",))]
|
||
|
||
|
||
# ══════════════════════════ 後端 B：youlin portal 資料面 ══════════════════
|
||
# 使用者真的會走的那條路：portal 帳密登入 → /portal/data/*。
|
||
# MCP 的 kbdb_* 工具（identity.kind='portal'）打的是同一組端點、同一道閘。
|
||
class Portal:
|
||
def __init__(self, base, email, password):
|
||
self.base = base.rstrip("/")
|
||
self.session = self._login(email, password)
|
||
|
||
def _curl(self, method, path, body=None, auth=True):
|
||
cmd = ["curl", "-s", "--max-time", "40", "-X", method, f"{self.base}{path}",
|
||
"-A", "Mozilla/5.0", "-H", "Content-Type: application/json"]
|
||
if auth:
|
||
cmd += ["-H", f"Authorization: Bearer {self.session}"]
|
||
if body is not None:
|
||
cmd += ["-d", json.dumps(body, ensure_ascii=False)]
|
||
out = subprocess.run(cmd, capture_output=True, text=True, timeout=60).stdout
|
||
try:
|
||
return json.loads(out)
|
||
except Exception:
|
||
return {"error": "non-json response", "raw": out[:300]}
|
||
|
||
def _login(self, email, password):
|
||
d = self._curl("POST", "/portal/login", {"email": email, "password": password}, auth=False)
|
||
tok = d.get("session") or d.get("token") or d.get("access_token")
|
||
if not tok:
|
||
raise SystemExit(json.dumps({"error": "portal login failed", "detail": d},
|
||
ensure_ascii=False))
|
||
return tok
|
||
|
||
def list_sheets(self):
|
||
return self._curl("GET", "/portal/data/templates")
|
||
|
||
def create_sheet(self, name, fields, description=None):
|
||
return self._curl("POST", "/portal/data/templates",
|
||
{"name": name, "slots": fields, "description": description})
|
||
|
||
def append_record(self, sheet, values, metadata_json=None):
|
||
return self._curl("POST", "/portal/data/records", {"template": sheet, "values": values})
|
||
|
||
def get_record(self, rid):
|
||
return self._curl("GET", f"/portal/data/records/{rid}")
|
||
|
||
def get_records(self, sheet):
|
||
return self._curl("GET", f"/portal/data/records/by-template/{sheet}")
|
||
|
||
def search(self, q):
|
||
return self._curl("GET", f"/portal/data/search?q={q}")
|
||
|
||
|
||
# ══════════════════════════ 後端 C：acr CLI（leo 2026-08-15 指定）══════════════
|
||
# 「你可以叫它用 CLI 考試。」CLI／MCP／portal 是同一套 API 的三個薄殼
|
||
# （`cli/src/commands/kbdb.ts` 檔頭：能力長在基本盤 API，CLI 只做介面轉換）。
|
||
# `acr kbdb` 的動作**剛好就是那六個按鈕**，多一個少一個都沒有。
|
||
#
|
||
# 🔴 **它打哪一台，由 cwd 決定**：解析順序是
|
||
# env > 資料夾層 `.arcrun.yaml`（就近往上找）> 全域 `~/.arcrun/config.yaml`，
|
||
# 而**全域指的是 leo21c（leo 的真庫，47.9 萬筆）**。
|
||
# ⇒ 本後端在建構時強制跑一次 `acr whoami`，**確認 CF 帳號是預期那台才准往下走**。
|
||
# 不憑上一次的結果假設這一次也一樣（2026-08-15 就是這一步救了總管）。
|
||
YOULIN_ACCOUNT = "1129efd7df2e8899d537e9c8fbabb6cb"
|
||
# 🔴 2026-08-16 補：只比對 CF 帳號**不夠**——那不是決定資料落到誰名下的那一項。
|
||
# 實撞：專案層 `.arcrun.yaml` 只寫 cypher_executor_url ＋ cloudflare_account_id 時，
|
||
# `acr whoami` 印出——
|
||
# 帳號 bfezv28v ← 沒被覆蓋，從全域掉下來的（leo21c）
|
||
# 連哪台 youlin 的 cypher
|
||
# CF 帳號 1129efd7…（youlin）
|
||
# ⇒ 「打 youlin 這台機器，但用 leo21c 的身分寫入」。
|
||
# 而舊的防呆只找 CF 帳號字串，那一項是對的 ⇒ **它會放行**，
|
||
# 考完會拿到一份看起來正常、實際寫進錯地方的成績。
|
||
# ⇒ 判準：防呆要比對「**決定後果的那一項**」，不是「剛好看得到的那一項」。
|
||
# namespace 才是資料的歸屬鍵 ⇒ 三項一起驗，缺一不可。
|
||
YOULIN_NAMESPACE = "yuga3bse"
|
||
YOULIN_CYPHER = "arcrun-cypher-executor.youlin-hsieh-dev.workers.dev"
|
||
|
||
|
||
class Acr:
|
||
def __init__(self, workdir, expect_account=YOULIN_ACCOUNT,
|
||
expect_namespace=YOULIN_NAMESPACE, expect_cypher=YOULIN_CYPHER):
|
||
self.cwd = workdir
|
||
who = subprocess.run(["acr", "whoami"], cwd=workdir, capture_output=True,
|
||
text=True, timeout=60).stdout
|
||
missing = [label for label, token in (
|
||
("CF 帳號", expect_account),
|
||
("namespace（資料歸屬鍵）", expect_namespace),
|
||
("cypher 主機", expect_cypher),
|
||
) if token and token not in who]
|
||
if missing:
|
||
raise SystemExit(
|
||
"🔴 acr 指到的不是預期的實例，拒絕往下走。\n"
|
||
" 對不上：" + "、".join(missing) + "\n"
|
||
" ⚠️ 三項分別決定「哪個 CF 帳號」「資料算誰的」「打哪台機器」，"
|
||
"缺一項就可能考在錯的地方。\n" + who)
|
||
self.whoami = who
|
||
|
||
def _run(self, args):
|
||
r = subprocess.run(["acr", "kbdb"] + args, cwd=self.cwd,
|
||
capture_output=True, text=True, timeout=120)
|
||
return {"exit": r.returncode, "out": r.stdout.strip(), "err": r.stderr.strip()}
|
||
|
||
def list_sheets(self):
|
||
return self._run(["template", "list"])
|
||
|
||
def create_sheet(self, name, fields, description=None):
|
||
return self._run(["template", "create", name, "--slots", ",".join(fields)])
|
||
|
||
def append_record(self, sheet, values, metadata_json=None):
|
||
args = ["record", "create", sheet]
|
||
for k, v in values.items():
|
||
args += ["--values", f"{k}={v}"]
|
||
return self._run(args)
|
||
|
||
def get_record(self, rid):
|
||
return self._run(["record", "get", rid])
|
||
|
||
def get_records(self, sheet):
|
||
return self._run(["query", sheet])
|
||
|
||
def search(self, q):
|
||
return self._run(["search", q])
|
||
|
||
|
||
def make_backend(spec):
|
||
kind, _, arg = spec.partition(":")
|
||
if kind == "local":
|
||
return Local(arg)
|
||
if kind == "portal":
|
||
return Portal(arg, os.environ["PORTAL_EMAIL"], os.environ["PORTAL_PASSWORD"])
|
||
if kind == "acr":
|
||
# arg ＝ 考場資料夾（裡面放 .arcrun.yaml，決定打哪一台）
|
||
return Acr(arg)
|
||
raise SystemExit(f"unknown backend: {spec}")
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--backend", required=True)
|
||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||
sub.add_parser("list_sheets")
|
||
p = sub.add_parser("create_sheet"); p.add_argument("--name", required=True)
|
||
p.add_argument("--fields", nargs="+", required=True); p.add_argument("--description")
|
||
p = sub.add_parser("append_record"); p.add_argument("--sheet", required=True)
|
||
p.add_argument("--values", required=True, help="JSON 物件 {欄位名: 內容}")
|
||
p = sub.add_parser("get_record"); p.add_argument("--id", required=True)
|
||
p = sub.add_parser("get_records"); p.add_argument("--sheet", required=True)
|
||
p = sub.add_parser("search"); p.add_argument("--q", required=True)
|
||
a = ap.parse_args()
|
||
|
||
b = make_backend(a.backend)
|
||
if a.cmd == "list_sheets":
|
||
out = b.list_sheets()
|
||
elif a.cmd == "create_sheet":
|
||
out = b.create_sheet(a.name, a.fields, a.description)
|
||
elif a.cmd == "append_record":
|
||
out = b.append_record(a.sheet, json.loads(a.values))
|
||
elif a.cmd == "get_record":
|
||
out = b.get_record(a.id)
|
||
elif a.cmd == "get_records":
|
||
out = b.get_records(a.sheet)
|
||
else:
|
||
out = b.search(a.q)
|
||
print(json.dumps(out, ensure_ascii=False, indent=2))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|