144 lines
4.8 KiB
Python
144 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
||
"""Le SPEC 调用账本: 记录每次依赖真实交互, 供审查与交付作证据。
|
||
|
||
用法:
|
||
python3 ledger.py append --dep grilling --kind skill-load --method file-read \
|
||
--summary "Phase2 R1 frontier 4问" [--cmd "..."] [--exit 0] [--file PATH]
|
||
python3 ledger.py check [--file PATH] [--json]
|
||
python3 ledger.py tail [-n 5] [--file PATH]
|
||
|
||
账本定位优先级: --file > openspec root 下 openspec/le-spec/ledger.jsonl > ./.le-spec/ledger.jsonl
|
||
"""
|
||
import argparse
|
||
import datetime
|
||
import json
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
KINDS = {"skill-load", "cli", "subagent"}
|
||
METHODS = {"skill-tool", "file-read", "bash", "agent"}
|
||
|
||
|
||
def default_ledger():
|
||
try:
|
||
out = subprocess.run(["openspec", "context", "--json"],
|
||
capture_output=True, text=True, timeout=15)
|
||
ctx = json.loads(out.stdout or "{}")
|
||
root = (ctx.get("root") or {}).get("path")
|
||
if root:
|
||
p = Path(root) / "openspec" / "le-spec" / "ledger.jsonl"
|
||
return p, True
|
||
except Exception:
|
||
pass
|
||
return Path.cwd() / ".le-spec" / "ledger.jsonl", False
|
||
|
||
|
||
def resolve_ledger(args):
|
||
"""返回 (path, has_openspec_root)。"""
|
||
if getattr(args, "file", None):
|
||
return Path(args.file).expanduser(), None
|
||
return default_ledger()
|
||
|
||
|
||
def cmd_append(args):
|
||
if args.kind not in KINDS or args.method not in METHODS:
|
||
print(f"kind 须为 {KINDS}, method 须为 {METHODS}", file=sys.stderr)
|
||
return 2
|
||
path, has_root = resolve_ledger(args)
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
entry = {
|
||
"ts": datetime.datetime.now().isoformat(timespec="seconds"),
|
||
"phase": args.phase, "dep": args.dep, "kind": args.kind,
|
||
"method": args.method, "cmd": args.cmd, "exit": args.exit,
|
||
"summary": args.summary,
|
||
}
|
||
with path.open("a", encoding="utf-8") as f:
|
||
f.write(json.dumps({k: v for k, v in entry.items() if v is not None},
|
||
ensure_ascii=False) + "\n")
|
||
print(f"账本已记: {path}")
|
||
if has_root is False and not args.file:
|
||
print("(未找到 openspec root, 暂写 .le-spec/; Phase 5 init 后迁移)", file=sys.stderr)
|
||
return 0
|
||
|
||
|
||
def _load(path):
|
||
if not path.exists():
|
||
return []
|
||
rows = []
|
||
for line in path.read_text(encoding="utf-8").splitlines():
|
||
line = line.strip()
|
||
if line:
|
||
try:
|
||
rows.append(json.loads(line))
|
||
except json.JSONDecodeError:
|
||
rows.append({"_unparsed": line})
|
||
return rows
|
||
|
||
|
||
def cmd_check(args):
|
||
path, _ = resolve_ledger(args)
|
||
rows = _load(path)
|
||
by_dep = {}
|
||
for r in rows:
|
||
d = by_dep.setdefault(r.get("dep", "?"), {"count": 0, "last": None, "kinds": {}})
|
||
d["count"] += 1
|
||
d["last"] = r.get("ts", d["last"])
|
||
d["kinds"][r.get("kind", "?")] = d["kinds"].get(r.get("kind", "?"), 0) + 1
|
||
if args.json:
|
||
print(json.dumps({"ledger": str(path), "total": len(rows), "by_dep": by_dep},
|
||
ensure_ascii=False, indent=2))
|
||
else:
|
||
print(f"账本: {path} ｜ 总计 {len(rows)} 条")
|
||
for dep, d in sorted(by_dep.items()):
|
||
kinds = ", ".join(f"{k}×{v}" for k, v in d["kinds"].items())
|
||
print(f" {dep:<16} {d['count']:>3} 次 ({kinds}) 末次 {d['last']}")
|
||
expected = ["grilling", "think", "openspec"]
|
||
missing = [e for e in expected if e not in by_dep]
|
||
if missing:
|
||
hint = {"think": "（若 Phase 3 合规跳过——无实质方案分歧——属正常，确认 state.md 有跳过记录即可）"}
|
||
for m in missing:
|
||
print(f"⚠️ 未见 {m} 的调用记录 {hint.get(m, '')} — 否则 Phase 6 审查前需真实补调或如实声明")
|
||
return 0
|
||
|
||
|
||
def cmd_tail(args):
|
||
path, _ = resolve_ledger(args)
|
||
rows = _load(path)[-args.n:]
|
||
for r in rows:
|
||
print(json.dumps(r, ensure_ascii=False))
|
||
return 0
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||
|
||
a = sub.add_parser("append")
|
||
a.add_argument("--dep", required=True)
|
||
a.add_argument("--kind", required=True)
|
||
a.add_argument("--method", required=True)
|
||
a.add_argument("--phase")
|
||
a.add_argument("--cmd")
|
||
a.add_argument("--exit", type=int)
|
||
a.add_argument("--summary")
|
||
a.add_argument("--file")
|
||
a.set_defaults(fn=cmd_append)
|
||
|
||
c = sub.add_parser("check")
|
||
c.add_argument("--file")
|
||
c.add_argument("--json", action="store_true")
|
||
c.set_defaults(fn=cmd_check)
|
||
|
||
t = sub.add_parser("tail")
|
||
t.add_argument("-n", type=int, default=5)
|
||
t.add_argument("--file")
|
||
t.set_defaults(fn=cmd_tail)
|
||
|
||
args = ap.parse_args()
|
||
sys.exit(args.fn(args))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|