108 lines
2.9 KiB
Python
108 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
||
"""audit_reminder — Agent 结束时检查是否有待审计改动，15 分钟限频提醒。
|
||
|
||
替代原 PowerShell 版本，避免 Windows PowerShell 5.1 解析器 bug。
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
from datetime import datetime, timezone, timedelta
|
||
|
||
TZ_TAIPEI = timezone(timedelta(hours=8))
|
||
STATE_PATH = os.path.join(".kiro", "state", ".audit_state.json")
|
||
MIN_INTERVAL = timedelta(minutes=15)
|
||
|
||
|
||
def now_taipei():
|
||
return datetime.now(TZ_TAIPEI)
|
||
|
||
|
||
def load_state():
|
||
if not os.path.isfile(STATE_PATH):
|
||
return None
|
||
try:
|
||
with open(STATE_PATH, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def save_state(state):
|
||
os.makedirs(os.path.join(".kiro", "state"), exist_ok=True)
|
||
with open(STATE_PATH, "w", encoding="utf-8") as f:
|
||
json.dump(state, f, indent=2, ensure_ascii=False)
|
||
|
||
|
||
def get_real_changes():
|
||
"""获取排除噪声后的变更文件"""
|
||
try:
|
||
r = subprocess.run(["git", "status", "--porcelain"], capture_output=True, text=True, timeout=10)
|
||
if r.returncode != 0:
|
||
return []
|
||
except Exception:
|
||
return []
|
||
files = []
|
||
for line in r.stdout.splitlines():
|
||
if len(line) < 4:
|
||
continue
|
||
path = line[3:].strip().strip('"').replace("\\", "/")
|
||
if " -> " in path:
|
||
path = path.split(" -> ")[-1]
|
||
# 排除审计产物、.kiro 配置、临时文件
|
||
if path and not path.startswith("docs/audit/") and not path.startswith(".kiro/") and not path.startswith("tmp/") and not path.startswith(".hypothesis/"):
|
||
files.append(path)
|
||
return sorted(set(files))
|
||
|
||
|
||
def main():
|
||
state = load_state()
|
||
if not state:
|
||
sys.exit(0)
|
||
|
||
if not state.get("audit_required"):
|
||
sys.exit(0)
|
||
|
||
# 工作树干净时清除审计状态
|
||
real_files = get_real_changes()
|
||
if not real_files:
|
||
state["audit_required"] = False
|
||
state["reasons"] = []
|
||
state["changed_files"] = []
|
||
state["last_reminded_at"] = None
|
||
save_state(state)
|
||
sys.exit(0)
|
||
|
||
now = now_taipei()
|
||
|
||
# 15 分钟限频
|
||
last_str = state.get("last_reminded_at")
|
||
if last_str:
|
||
try:
|
||
last = datetime.fromisoformat(last_str)
|
||
if (now - last) < MIN_INTERVAL:
|
||
sys.exit(0)
|
||
except Exception:
|
||
pass
|
||
|
||
# 更新提醒时间
|
||
state["last_reminded_at"] = now.isoformat()
|
||
save_state(state)
|
||
|
||
reasons = state.get("reasons", [])
|
||
reason_text = ", ".join(reasons) if reasons else "high-risk paths changed"
|
||
sys.stderr.write(
|
||
f"[AUDIT REMINDER] Pending audit detected ({reason_text}). "
|
||
f"Run /audit (Manual: Run /audit hook) to sync docs & write audit artifacts. "
|
||
f"(rate limit: 15min)\n"
|
||
)
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except Exception:
|
||
sys.exit(0)
|