141 lines
5.1 KiB
Python
141 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
||
"""le-app-design 证据账本：所有对上游 Skill 的真实调用经此包裹，落入 calls.jsonl。
|
||
|
||
用法:
|
||
执行型调用（CLI 类上游）:
|
||
python3 log_call.py <dep> --run-dir <RUN> -- python3 <DEP>/scripts/search.py "…" --design-system
|
||
装载型调用（文档型上游 / playbook）:
|
||
python3 log_call.py <dep> --run-dir <RUN> --load <DEP>/SKILL.md
|
||
|
||
dep ∈ {ui-ux-pro-max, frontend-design, impeccable, le-app-design(自检)}
|
||
--run-dir 缺省时按序: $LE_APP_DESIGN_RUN_DIR > <cwd>/le-app-design-runs/<UTC 时间戳>-run/
|
||
|
||
行为:
|
||
- exec: 运行命令，透传 stdout/stderr，透传退出码；成功失败都记账。
|
||
- load: 读文件记 sha256 与字节数（证明用的是原文而非转述），文件缺失记 FAIL 退出 1。
|
||
- 记录字段: ts / dep / kind / command / exit_code / duration_ms / stdout_tail /
|
||
stderr_tail / bytes / sha256 / note。stdout/stderr 仅保留尾部 4000 字符防账本膨胀。
|
||
|
||
仅 python3 标准库。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
TAIL = 4000
|
||
KNOWN_DEPS = {"ui-ux-pro-max", "frontend-design", "impeccable", "le-app-design"}
|
||
|
||
|
||
def default_run_dir() -> Path:
|
||
env = os.environ.get("LE_APP_DESIGN_RUN_DIR")
|
||
if env:
|
||
return Path(env).expanduser()
|
||
stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%SZ")
|
||
return Path.cwd() / "le-app-design-runs" / f"{stamp}-run"
|
||
|
||
|
||
def append_record(run_dir: Path, record: dict) -> Path:
|
||
run_dir.mkdir(parents=True, exist_ok=True)
|
||
ledger = run_dir / "calls.jsonl"
|
||
with ledger.open("a", encoding="utf-8") as f:
|
||
f.write(json.dumps(record, ensure_ascii=False) + "\n")
|
||
return ledger
|
||
|
||
|
||
def confirm(dep: str, kind: str, status: str, ledger: Path, extra: str = "") -> None:
|
||
print(
|
||
f"[log_call] {dep} {kind} {status} → {ledger}"
|
||
+ (f" ({extra})" if extra else ""),
|
||
file=sys.stderr,
|
||
)
|
||
|
||
|
||
def main() -> int:
|
||
# 手工在第一个裸 "--" 处切分：argparse 的 REMAINDER 会把 -- 之前的选项误吞进命令
|
||
argv = sys.argv[1:]
|
||
cmd: list[str] = []
|
||
if "--" in argv:
|
||
split = argv.index("--")
|
||
argv, cmd = argv[:split], argv[split + 1:]
|
||
|
||
ap = argparse.ArgumentParser(description="le-app-design 调用账本")
|
||
ap.add_argument("dep", help="上游标识: ui-ux-pro-max | frontend-design | impeccable | le-app-design")
|
||
ap.add_argument("--run-dir", default=None, help="运行目录（缺省自动创建）")
|
||
ap.add_argument("--note", default="", help="阶段/用途备注，如 S2-design-system")
|
||
ap.add_argument("--load", metavar="PATH", default=None, help="装载型调用：记录读取的原文文件")
|
||
args = ap.parse_args(argv)
|
||
|
||
dep = args.dep
|
||
if dep not in KNOWN_DEPS:
|
||
print(f"[log_call] 未知依赖标识 {dep!r}；允许: {sorted(KNOWN_DEPS)}", file=sys.stderr)
|
||
return 2
|
||
|
||
run_dir = Path(args.run_dir).expanduser() if args.run_dir else default_run_dir()
|
||
base = {"ts": datetime.now(timezone.utc).isoformat(timespec="seconds"), "dep": dep, "note": args.note}
|
||
|
||
if args.load:
|
||
path = Path(args.load).expanduser()
|
||
rec = dict(base, kind="load", command=f"load:{path}")
|
||
try:
|
||
data = path.read_bytes()
|
||
except OSError as e:
|
||
rec.update(exit_code=1, stderr_tail=f"read failed: {e}"[:TAIL])
|
||
ledger = append_record(run_dir, rec)
|
||
confirm(dep, "load", "FAIL", ledger, str(e))
|
||
return 1
|
||
rec.update(
|
||
exit_code=0,
|
||
bytes=len(data),
|
||
sha256=hashlib.sha256(data).hexdigest(),
|
||
)
|
||
ledger = append_record(run_dir, rec)
|
||
confirm(dep, "load", "OK", ledger, f"{len(data)} bytes sha256:{rec['sha256'][:12]}…")
|
||
return 0
|
||
|
||
if not cmd:
|
||
print("[log_call] 需要 --load PATH 或 -- <命令…>", file=sys.stderr)
|
||
return 2
|
||
|
||
rec = dict(base, kind="exec", command=json.dumps(cmd, ensure_ascii=False))
|
||
t0 = time.monotonic()
|
||
try:
|
||
proc = subprocess.run(
|
||
cmd, capture_output=True, text=True, timeout=900,
|
||
env=dict(os.environ, PYTHONDONTWRITEBYTECODE="1"),
|
||
)
|
||
rc, out, err = proc.returncode, proc.stdout or "", proc.stderr or ""
|
||
except subprocess.TimeoutExpired as e:
|
||
rc, out, err = 124, (e.stdout or b"").decode(errors="replace") if isinstance(e.stdout, bytes) else (e.stdout or ""), f"timeout after {e.timeout}s"
|
||
except FileNotFoundError as e:
|
||
rc, out, err = 127, "", f"command not found: {e}"
|
||
|
||
rec.update(
|
||
exit_code=rc,
|
||
duration_ms=int((time.monotonic() - t0) * 1000),
|
||
stdout_tail=out[-TAIL:],
|
||
stderr_tail=err[-TAIL:],
|
||
)
|
||
ledger = append_record(run_dir, rec)
|
||
if out:
|
||
sys.stdout.write(out)
|
||
if not out.endswith("\n"):
|
||
sys.stdout.write("\n")
|
||
if err:
|
||
sys.stderr.write(err)
|
||
if not err.endswith("\n"):
|
||
sys.stderr.write("\n")
|
||
confirm(dep, "exec", f"exit={rc}", ledger, f"{rec['duration_ms']}ms")
|
||
return rc
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|