自检驱动机制 + 标注修正(BOM/E暂缓)
- checklist_audit.py: 扫描文档标记→对比系统→输出差距报告 - metagrowth: 修复BOM导致JSON解析失败 - E标注: 🔄→⏸️ 暂缓(Phase 3) - QQ通道: ⏸️ 暂缓标记 - annotation审计: 24条标记,19✅/3🔄/2⏸️/0❌
This commit is contained in:
+1
-1
@@ -24,7 +24,7 @@
|
||||
|
||||
- **XMPP 通道** — 通过 ejabberd 服务器 + XMPP bot (slixmpp) 实现群聊/私聊消息收发。bot 进程：xmpp_bot.py (mohe)、xmpp_xiaoguo_bot.py、xmpp_zhiwei_bot.py。支持 __SILENT__/__REPLY__ 前缀机制控制 bot 输出行为。
|
||||
- **微信通道** — Windows 端 wechat_agent 桥接程序，抓取微信消息后通过 HTTP POST 到 Gateway，回复经 5801 端口推回微信窗口，长消息拆分为 2000 字/段。
|
||||
- **QQ 通道** — 规划中，未实现。
|
||||
- **QQ 通道** — 规划中，未实现。 <!-- ⏸️ 暂缓 — 当前无迫切需求 -->
|
||||
|
||||
### 2.2 Gateway 层 <!-- ✅ 8642/8643 运行中，健康检查已配置 -->
|
||||
|
||||
|
||||
@@ -0,0 +1,145 @@
|
||||
"""
|
||||
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()
|
||||
for m in re.finditer(r'<!--\s*(.*?)\s*-->', 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. 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()
|
||||
@@ -696,12 +696,15 @@ def api_metagrowth():
|
||||
total_fixes = 0
|
||||
completed_fixes = 0
|
||||
if todo_file.exists():
|
||||
with open(str(todo_file)) as f:
|
||||
with open(str(todo_file), encoding="utf-8-sig") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if line:
|
||||
total_fixes += 1
|
||||
try:
|
||||
entry = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
if entry.get("status") == "completed":
|
||||
completed_fixes += 1
|
||||
# 统计 git 提交次数（从 reflog 行数计算）
|
||||
|
||||
Reference in New Issue
Block a user