""" checklist_audit.py — 自检驱动实现机制 读取 G 规范 + H 需求的标注，对比系统实际状态，输出差距报告。 可被 self_todo_executor 调用作为自检环节。 """ import re, os, sys, json, urllib.request sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from service_registry import BASE, TEMP SPEC_PATH = os.path.normpath(os.path.join(BASE, "..", "..", ".memory", "dev-spec.md")) PRD_PATH = os.path.normpath(os.path.join(BASE, "docs", "PRD.md")) TODO_FILE = os.path.join(TEMP, "checklist_gaps.jsonl") DASHBOARD_URL = "http://127.0.0.1:5803" def extract_annotations(path): """Extract annotations with their context (section heading)""" if not os.path.exists(path): return [] with open(path, "r", encoding="utf-8") as f: content = f.read() results = [] lines = content.split("\n") current_section = "" for line in lines: if line.startswith("#"): current_section = line.strip() # Check for ❌ in table cells (e.g. | R01 | ... | ... | ... | ... | ❌ 未修复 |) # Check for ❌ in table cells (separate regexes for each status) for m in re.finditer(r'\|\s*❌\s*([^|]*?)\s*\|', line): cell_content = "❌ " + m.group(1).strip() status = "fail" if "暂缓" not in cell_content else "postponed" results.append({ "annotation": cell_content, "status": status, "section": current_section, "source": os.path.basename(path), "in_table": True, }) for m in re.finditer(r'\|\s*⏸️\s*([^|]*?)\s*\|', line): cell_content = "⏸️ " + m.group(1).strip() status = "postponed" results.append({ "annotation": cell_content, "status": status, "section": current_section, "source": os.path.basename(path), "in_table": True, }) for m in re.finditer(r'', line): anno = m.group(1) # Determine status if "✅" in anno: status = "done" elif "⏸️" in anno: status = "postponed" elif "❌" in anno: status = "fail" elif "🔄" in anno: status = "wip" else: status = "unknown" results.append({ "annotation": anno, "status": status, "section": current_section, "source": os.path.basename(path), }) return results def check_api(endpoint, timeout=5): """Check if an API endpoint returns OK""" try: r = urllib.request.urlopen(f"{DASHBOARD_URL}{endpoint}", timeout=timeout) return r.getcode() == 200, json.loads(r.read()) except: return False, None def main(): print("=" * 60) print(" 自检驱动实现 — CheckList Audit") print("=" * 60) # 1. Extract all annotations from both docs all_annos = extract_annotations(SPEC_PATH) + extract_annotations(PRD_PATH) print(f"\n📋 共发现 {len(all_annos)} 条标注") # 2. Categorize done = [a for a in all_annos if a["status"] == "done"] wip = [a for a in all_annos if a["status"] == "wip"] postponed = [a for a in all_annos if a["status"] == "postponed"] fails = [a for a in all_annos if a["status"] == "fail"] print(f" ✅ 已完成: {len(done)}") print(f" 🔄 进行中: {len(wip)}") print(f" ⏸️ 暂缓: {len(postponed)}") print(f" ❌ 未实现: {len(fails)}") # 3. Report actionable gaps (❌ that are not ⏸️) action_items = [] for a in fails: # Check if this item was already done (system reality) action_items.append(a) if action_items: print(f"\n🔴 需要实现的缺口 ({len(action_items)} 项):") for a in action_items: print(f" ❌ {a['annotation'][:80]}") print(f" → 来源: {a['source']} / {a['section'][:60]}") # Write to TODO file for the executor with open(TODO_FILE, "w", encoding="utf-8") as f: for a in action_items: entry = { "created": __import__("datetime").datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "source": a["source"], "section": a["section"], "gap": a["annotation"], "status": "pending", } f.write(json.dumps(entry, ensure_ascii=False) + "\n") print(f"\n 已写入 {len(action_items)} 条 TODO 到 {TODO_FILE}") else: print(f"\n✅ 无待实现缺口") # 4. Remove stale TODOs (for items that are now done) if os.path.exists(TODO_FILE): with open(TODO_FILE, "r", encoding="utf-8") as f: existing = f.readlines() # Keep only items whose annotation still has ❌ active_gaps = [a["annotation"] for a in action_items] new_lines = [] for line in existing: line = line.strip() if not line: continue try: entry = json.loads(line) if not any(gap in entry.get("gap", "") for gap in active_gaps): continue # skip stale new_lines.append(line) except: pass with open(TODO_FILE, "w", encoding="utf-8") as f: f.write("\n".join(new_lines) + "\n") print(f" TODO 文件已清理: {len(new_lines)} 活跃条目") # 5. Deploy status check (H0 规范) print(f"\n🚀 部署状态 (H0):") agents_dir = os.path.normpath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..")) try: r_local = subprocess.run(["git", "rev-parse", "HEAD"], cwd=agents_dir, capture_output=True, text=True, timeout=5) local_head = r_local.stdout.strip() if r_local.returncode == 0 else "?" r_remote = subprocess.run(["git", "rev-parse", "origin/master"], cwd=agents_dir, capture_output=True, text=True, timeout=5) remote_head = r_remote.stdout.strip() if r_remote.returncode == 0 else "?" r_unpushed = subprocess.run(["git", "log", "origin/master..HEAD", "--oneline"], cwd=agents_dir, capture_output=True, text=True, timeout=5) unpushed = r_unpushed.stdout.strip() if local_head == remote_head: print(f" ✅ 本地与 origin/master 一致: {local_head[:12]}") else: print(f" ⚠️ 本地与 origin/master 不同步") print(f" local: {local_head[:12]} remote: {remote_head[:12]}") if unpushed: print(f" ⚠️ 有 {len(unpushed.split(chr(10)))} 个本地提交未推送:") for line in unpushed.split(chr(10))[:3]: print(f" {line[:60]}") else: print(f" ✅ 无待推送提交") except Exception as e: print(f" ❌ 检查失败: {str(e)[:60]}") # 6. Run system reality checks print(f"\n🔍 系统状态快照:") endpoints = [ ("Services", "/api/services"), ("Git", "/api/git"), ("Monitor", "/api/monitor"), ("MetaGrowth", "/api/metagrowth"), ("Expected", "/api/expected"), ] for name, ep in endpoints: ok, data = check_api(ep) status = "✅" if ok else "❌" print(f" {status} {name}") if __name__ == "__main__": main()