AgentsMeeting/gateway/scripts/checklist_audit.py

"""
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'<!--\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()