127 lines
5.9 KiB
Python
127 lines
5.9 KiB
Python
#!/usr/bin/env python3
|
||
"""Выгрузка блокировок хуков из транскриптов Claude Code в JSONL — сырьё для регрессионных тестов хуков (B2).
|
||
|
||
Запуск: python3 hook_blocks_export.py [--since ГГГГ-ММ-ДД] [--id-prefix <дата>] <каталог-транскриптов> [...]
|
||
Одна строка — одна блокировка: сессия, время, cwd и ветка на момент вызова, инструмент, команда или путь файла,
|
||
первая строка ответа хука. Поле expected не заполняется: вердикт (block/allow) проставляется при разметке.
|
||
Выгружаются только ошибки хуков (результат начинается с `PreToolUse:<инструмент> hook error`); вывод команд,
|
||
в котором встретилось слово BLOCKED (файлы памяти, исходники хуков), не попадает.
|
||
--since — только сессии с первой меткой времени не раньше даты; --id-prefix (по умолчанию — дата --since) —
|
||
поле id вида <префикс>-NNN (срез ретро, B12). Похожие на секреты подстроки заменяются на <redacted>.
|
||
"""
|
||
import glob
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
HOOK_ERROR = re.compile(r"\s*PreToolUse:\S+ hook error")
|
||
|
||
RE_SECRET = re.compile(
|
||
r'(?i)gh[pousr]_[A-Za-z0-9]{20,}|sk-[A-Za-z0-9_-]{20,}|\b[0-9a-f]{40}\b'
|
||
r'|(?P<kv>(?:token|password|passwd|secret|api[_-]?key)\s*[=:]\s*)\S+'
|
||
r'|(?P<flag>--(?:token|password)\s+)\S+|(?P<auth>Authorization:\s*\S+\s+)\S+'
|
||
)
|
||
|
||
|
||
def redact(s):
|
||
def sub(m):
|
||
keep = m.group("kv") or m.group("flag") or m.group("auth") or ""
|
||
return keep + "<redacted>"
|
||
return RE_SECRET.sub(sub, s)
|
||
|
||
|
||
def hook_of(txt, tool):
|
||
if "[guard-main]" in txt:
|
||
return "guard-main(Bash)" if tool == "Bash" else "guard-main(Edit/Write)"
|
||
if "[require-plan]" in txt:
|
||
return "require-plan"
|
||
if "shell write" in txt: # формулировки до 0.4.0
|
||
return "guard-main(Bash)"
|
||
if "you are on" in txt:
|
||
return "guard-main(Edit/Write)"
|
||
if "without a plan" in txt:
|
||
return "require-plan"
|
||
if "commit/push в main" in txt or "ветке main" in txt or "reference-old-arm" in txt or "docs/source" in txt:
|
||
return "workspace-guard"
|
||
return "other"
|
||
|
||
|
||
def _records(path):
|
||
for line in open(path, encoding="utf-8", errors="replace"):
|
||
try:
|
||
j = json.loads(line)
|
||
except ValueError:
|
||
continue
|
||
if isinstance(j, dict):
|
||
yield j
|
||
|
||
|
||
def first_ts(path):
|
||
for j in _records(path):
|
||
if isinstance(j.get("timestamp"), str) and j["timestamp"]:
|
||
return j["timestamp"]
|
||
return ""
|
||
|
||
|
||
def export(dirs, since=None, prefix=None):
|
||
"""Записи выгрузки по каталогам транскриптов, по порядку файлов и строк."""
|
||
n = 0
|
||
for d in dirs:
|
||
for f in sorted(glob.glob(os.path.join(os.path.expanduser(d), "*.jsonl"))):
|
||
if since and first_ts(f)[:10] < since:
|
||
continue
|
||
sid = os.path.basename(f)[:8]
|
||
pending = {} # tool_use_id -> (name, input, cwd, branch, ts)
|
||
for j in _records(f):
|
||
m = j.get("message") or {}
|
||
cont = m.get("content")
|
||
if j.get("type") == "assistant":
|
||
for b in cont or []:
|
||
if isinstance(b, dict) and b.get("type") == "tool_use":
|
||
pending[b.get("id")] = (b.get("name"), b.get("input") or {}, j.get("cwd", ""),
|
||
j.get("gitBranch", ""), j.get("timestamp", ""))
|
||
elif j.get("type") == "user" and isinstance(cont, list):
|
||
for b in cont:
|
||
if not (isinstance(b, dict) and b.get("type") == "tool_result"):
|
||
continue
|
||
txt = b.get("content", "")
|
||
if isinstance(txt, list):
|
||
txt = "\n".join(x.get("text", "") for x in txt if isinstance(x, dict))
|
||
if not isinstance(txt, str) or not HOOK_ERROR.match(txt):
|
||
continue # обычный вывод команды, даже если в нём есть слово BLOCKED
|
||
name, inp, cwd, branch, ts = pending.get(b.get("tool_use_id"), ("?", {}, "", "", ""))
|
||
if name not in ("Bash", "Edit", "Write", "MultiEdit"):
|
||
continue
|
||
msg = next((l for l in txt.splitlines() if "BLOCKED" in l), "")
|
||
rec = {
|
||
"session": sid, "ts": ts, "cwd": cwd, "branch": branch, "tool": name,
|
||
"command": redact(inp.get("command", "")) if name == "Bash" else None,
|
||
"file_path": inp.get("file_path") if name != "Bash" else None,
|
||
"hook": hook_of(txt, name), "message": redact(msg[:300]), "expected": None,
|
||
}
|
||
if prefix:
|
||
n += 1
|
||
rec = dict({"id": "%s-%03d" % (prefix, n)}, **rec)
|
||
yield rec
|
||
|
||
|
||
def main(argv):
|
||
args, dirs, since, prefix = argv[1:], [], None, None
|
||
i = 0
|
||
while i < len(args):
|
||
if args[i] == "--since" and i + 1 < len(args):
|
||
since, i = args[i + 1], i + 2
|
||
elif args[i] == "--id-prefix" and i + 1 < len(args):
|
||
prefix, i = args[i + 1], i + 2
|
||
else:
|
||
dirs.append(args[i])
|
||
i += 1
|
||
for rec in export(dirs, since, prefix or since):
|
||
print(json.dumps(rec, ensure_ascii=False))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main(sys.argv))
|