260 lines
8.5 KiB
Python
260 lines
8.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
monthly_audit.py — Wizard Council Monthly Environment Audit
|
|
|
|
Runs all three checks (bootstrap, skills audit, dependency check) and
|
|
produces a combined Markdown report. Designed to be invoked by cron or
|
|
manually.
|
|
|
|
Usage:
|
|
python wizard-bootstrap/monthly_audit.py
|
|
python wizard-bootstrap/monthly_audit.py --output /path/to/report.md
|
|
python wizard-bootstrap/monthly_audit.py --post-telegram # post to configured channel
|
|
|
|
The report is also written to ~/.hermes/wizard-council/audit-YYYY-MM.md
|
|
"""
|
|
|
|
import argparse
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
from contextlib import redirect_stdout
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
# Ensure repo root is importable
|
|
_REPO_ROOT = Path(__file__).parent.parent
|
|
sys.path.insert(0, str(_REPO_ROOT))
|
|
|
|
from wizard_bootstrap import run_all_checks
|
|
from skills_audit import run_audit
|
|
from dependency_checker import run_dep_check
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Report builder
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _emoji(ok: bool) -> str:
|
|
return "✅" if ok else "❌"
|
|
|
|
|
|
def build_report(repo_root: Path) -> str:
|
|
now = datetime.now(timezone.utc)
|
|
lines = [
|
|
f"# Wizard Council Environment Audit",
|
|
f"",
|
|
f"**Date:** {now.strftime('%Y-%m-%d %H:%M UTC')}",
|
|
f"",
|
|
f"---",
|
|
f"",
|
|
]
|
|
|
|
# 1. Bootstrap checks
|
|
lines.append("## 1. Environment Bootstrap")
|
|
lines.append("")
|
|
bootstrap = run_all_checks()
|
|
for check in bootstrap.checks:
|
|
icon = _emoji(check.passed)
|
|
label = check.name.replace("_", " ").title()
|
|
lines.append(f"- {icon} **{label}**: {check.message}")
|
|
if not check.passed and check.fix_hint:
|
|
lines.append(f" - _Fix_: {check.fix_hint}")
|
|
lines.append("")
|
|
if bootstrap.passed:
|
|
lines.append("**Environment: READY** ✅")
|
|
else:
|
|
failed = len(bootstrap.failed)
|
|
lines.append(f"**Environment: {failed} check(s) FAILED** ❌")
|
|
lines.append("")
|
|
lines.append("---")
|
|
lines.append("")
|
|
|
|
# 2. Skills audit
|
|
lines.append("## 2. Skills Drift Audit")
|
|
lines.append("")
|
|
skills_report = run_audit(repo_root)
|
|
missing = skills_report.by_status("MISSING")
|
|
extra = skills_report.by_status("EXTRA")
|
|
outdated = skills_report.by_status("OUTDATED")
|
|
ok_count = len(skills_report.by_status("OK"))
|
|
total = len(skills_report.drifts)
|
|
|
|
lines.append(f"| Status | Count |")
|
|
lines.append(f"|--------|-------|")
|
|
lines.append(f"| ✅ OK | {ok_count} |")
|
|
lines.append(f"| ❌ Missing | {len(missing)} |")
|
|
lines.append(f"| ⚠️ Extra | {len(extra)} |")
|
|
lines.append(f"| 🔄 Outdated | {len(outdated)} |")
|
|
lines.append(f"| **Total** | **{total}** |")
|
|
lines.append("")
|
|
|
|
if missing:
|
|
lines.append("### Missing Skills (in repo, not installed)")
|
|
for d in missing:
|
|
lines.append(f"- `{d.skill_path}`")
|
|
lines.append("")
|
|
|
|
if outdated:
|
|
lines.append("### Outdated Skills")
|
|
for d in outdated:
|
|
lines.append(f"- `{d.skill_path}` (repo: `{d.repo_hash}`, installed: `{d.installed_hash}`)")
|
|
lines.append("")
|
|
|
|
if extra:
|
|
lines.append("### Extra Skills (installed, not in repo)")
|
|
for d in extra:
|
|
lines.append(f"- `{d.skill_path}`")
|
|
lines.append("")
|
|
|
|
if not skills_report.has_drift:
|
|
lines.append("**Skills: IN SYNC** ✅")
|
|
else:
|
|
lines.append("**Skills: DRIFT DETECTED** ❌ — run `python wizard-bootstrap/skills_audit.py --fix`")
|
|
lines.append("")
|
|
lines.append("---")
|
|
lines.append("")
|
|
|
|
# 3. Dependency check
|
|
lines.append("## 3. Cross-Wizard Dependency Check")
|
|
lines.append("")
|
|
dep_report = run_dep_check()
|
|
|
|
if not dep_report.deps:
|
|
lines.append("No explicit dependencies declared in SKILL.md frontmatter.")
|
|
lines.append("")
|
|
lines.append(
|
|
"_Tip: Add a `dependencies` block to SKILL.md to make binary/env_var "
|
|
"requirements checkable automatically._"
|
|
)
|
|
else:
|
|
satisfied = sum(1 for d in dep_report.deps if d.satisfied)
|
|
total_deps = len(dep_report.deps)
|
|
lines.append(f"**{satisfied}/{total_deps} dependencies satisfied.**")
|
|
lines.append("")
|
|
if dep_report.unsatisfied:
|
|
lines.append("### Unsatisfied Dependencies")
|
|
for dep in dep_report.unsatisfied:
|
|
dep_type = "binary" if dep.binary else "env_var"
|
|
dep_name = dep.binary or dep.env_var
|
|
lines.append(f"- `[{dep.skill_path}]` {dep_type}:`{dep_name}` — {dep.detail}")
|
|
lines.append("")
|
|
|
|
if dep_report.all_satisfied:
|
|
lines.append("**Dependencies: ALL SATISFIED** ✅")
|
|
else:
|
|
lines.append("**Dependencies: ISSUES FOUND** ❌")
|
|
lines.append("")
|
|
lines.append("---")
|
|
lines.append("")
|
|
|
|
# Summary
|
|
overall_ok = bootstrap.passed and not skills_report.has_drift and dep_report.all_satisfied
|
|
lines.append("## Summary")
|
|
lines.append("")
|
|
lines.append(f"| Check | Status |")
|
|
lines.append(f"|-------|--------|")
|
|
lines.append(f"| Environment Bootstrap | {_emoji(bootstrap.passed)} |")
|
|
lines.append(f"| Skills Drift | {_emoji(not skills_report.has_drift)} |")
|
|
lines.append(f"| Dependency Check | {_emoji(dep_report.all_satisfied)} |")
|
|
lines.append("")
|
|
if overall_ok:
|
|
lines.append("**Overall: FORGE READY** ✅")
|
|
else:
|
|
lines.append("**Overall: ACTION REQUIRED** ❌")
|
|
lines.append("")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Output / delivery
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _save_report(report: str, output_path: Path) -> None:
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
output_path.write_text(report, encoding="utf-8")
|
|
print(f"Report saved to: {output_path}")
|
|
|
|
|
|
def _post_telegram(report: str) -> None:
|
|
"""Post the report summary to Telegram via hermes gateway if configured."""
|
|
token = os.environ.get("TELEGRAM_BOT_TOKEN")
|
|
channel = os.environ.get("TELEGRAM_HOME_CHANNEL") or os.environ.get("TELEGRAM_CHANNEL_ID")
|
|
if not (token and channel):
|
|
print("Telegram not configured (need TELEGRAM_BOT_TOKEN + TELEGRAM_HOME_CHANNEL).", file=sys.stderr)
|
|
return
|
|
|
|
try:
|
|
import requests # noqa: PLC0415
|
|
|
|
# Extract just the summary section for Telegram (keep it brief)
|
|
summary_start = report.find("## Summary")
|
|
summary_text = report[summary_start:] if summary_start != -1 else report[-1000:]
|
|
payload = {
|
|
"chat_id": channel,
|
|
"text": f"🧙 **Wizard Council Monthly Audit**\n\n{summary_text}",
|
|
"parse_mode": "Markdown",
|
|
}
|
|
resp = requests.post(
|
|
f"https://api.telegram.org/bot{token}/sendMessage",
|
|
json=payload,
|
|
timeout=15,
|
|
)
|
|
if resp.status_code == 200:
|
|
print("Report summary posted to Telegram.")
|
|
else:
|
|
print(f"Telegram post failed: HTTP {resp.status_code}", file=sys.stderr)
|
|
except Exception as exc:
|
|
print(f"Telegram post error: {exc}", file=sys.stderr)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI entry point
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(
|
|
description="Run the monthly Wizard Council environment audit."
|
|
)
|
|
parser.add_argument(
|
|
"--output",
|
|
default=None,
|
|
help="Path to save the Markdown report (default: ~/.hermes/wizard-council/audit-YYYY-MM.md)",
|
|
)
|
|
parser.add_argument(
|
|
"--repo-root",
|
|
default=str(_REPO_ROOT),
|
|
help="Root of the hermes-agent repo",
|
|
)
|
|
parser.add_argument(
|
|
"--post-telegram",
|
|
action="store_true",
|
|
help="Post the report summary to Telegram",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
repo_root = Path(args.repo_root).resolve()
|
|
report = build_report(repo_root)
|
|
|
|
# Print to stdout
|
|
print(report)
|
|
|
|
# Save to default location
|
|
now = datetime.now(timezone.utc)
|
|
if args.output:
|
|
output_path = Path(args.output)
|
|
else:
|
|
hermes_home = Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes"))
|
|
output_path = hermes_home / "wizard-council" / f"audit-{now.strftime('%Y-%m')}.md"
|
|
|
|
_save_report(report, output_path)
|
|
|
|
if args.post_telegram:
|
|
_post_telegram(report)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|