#!/usr/bin/env python3 """ dashboard.py — FLEET COMMAND CENTER aggregator. Reads real state from all 7 agent profiles and emits: fleet/dashboard.json machine-readable snapshot fleet/dashboard.html self-contained live dashboard (open in browser) Handles missing/empty state gracefully (shows 0 / "no data yet"). No external deps. Run on cron (e.g. every 30m) so dashboard.json stays fresh. Usage: python3 dashboard.py [--html] """ import json import sys import datetime import html from pathlib import Path PROFILES = Path("/root/.hermes/profiles") OUT = PROFILES / "fleet" OUT.mkdir(exist_ok=True) AGENTS = [ ("default", "Command Center", "indigo", "⌘"), ("c3nter", "Command Bot", "indigo", "⌘"), ("coder", "DevForge", "cyan", "⌨"), ("hackbox", "ReconHub", "red", "⚔"), ("leadgen", "LeadForge", "blue", "◎"), ("realestate", "Maigida REI", "amber", "⌂"), ("seo", "ContentForge", "violet", "✦"), ("creative", "AI Studio", "pink", "◆"), ("govcontracts", "GovForge", "teal", "★"), ("tradermoe", "TradeDesk", "green", "$"), ("claimforge", "ClaimForge", "gold", "⚖"), ("domainhaus", "DomainHaus", "orange", "◈"), ("lifeagent", "LifeAssure", "emerald", "❖"), ("redteam", "RedSky Sec", "red", "⚔"), ] BOTS = { "default": "@undagroundbot", "c3nter": "@commandc3nterbot", "coder": "@fitatechdevbot", "hackbox": "@Reconhausbot", "leadgen": "@leadmasterdevbot", "realestate": "@MaigidaREIbot", "seo": "@mrseodevbot", "creative": "@AIStudiodevbot", "govcontracts": "@ongovcontractsbot", "tradermoe": "@m3tradebot", "claimforge": "@Claimforgebot", "domainhaus": "@domainhausbot", "lifeagent": "@AWInsurancebot", "redteam": "(own)", } def safe(fn, default=None): try: return fn() except Exception: return default def count_published(seo_dir): n = 0 arts = 0 p = seo_dir / "published" if p.exists(): for idx in p.glob("*/index.json"): try: d = json.loads(idx.read_text()) arts += len(d.get("articles", [])) n += 1 except Exception: pass return n, arts def gateway_status(pid): """Return (online:bool, detail:str) for a profile's gateway via process check.""" import subprocess try: out = subprocess.run(["pgrep", "-f", "hermes -p %s g" % pid], capture_output=True, text=True, timeout=5).stdout.split() if out: return True, "online" return False, "offline" except Exception: return False, "unknown" def agent_state(pid): w = PROFILES / pid / "workspace" g_on, g_det = gateway_status(pid) s = {"profile": pid, "bot": BOTS.get(pid, "?"), "metrics": {"gateway": "ONLINE" if g_on else "offline"}, "notes": ""} if pid == "leadgen": raw = sorted((w / "leadgen").glob("leads_raw_*.json")) if (w / "leadgen").exists() else [] if raw: d = json.loads(raw[-1].read_text()) s["metrics"]["leads_found"] = d.get("lead_count", len(d.get("leads", []))) s["metrics"]["last_run"] = d.get("generated_at", "")[:10] st = safe(lambda: json.loads((w / "leadgen/state.json").read_text())) if st: s["metrics"]["stage"] = st.get("stage", "?") s["notes"] = f"stage={st.get('stage')} awaiting={st.get('awaiting_approval')}" elif pid == "realestate": s["metrics"]["status"] = "awaiting /start" s["notes"] = "3 agents queued (motivated-sellers / agent-booking / deal-finder)" elif pid == "seo": n, arts = count_published(w / "seo") rev = safe(lambda: json.loads((w / "seo/revenue.json").read_text()), {"clients": {}}) total_articles = sum(c.get("articles", 0) for c in rev.get("clients", {}).values()) s["metrics"]["niches_live"] = n s["metrics"]["articles_published"] = arts s["metrics"]["billed_usd"] = round(sum(c.get("rate", 0) * c.get("articles", 0) for c in rev.get("clients", {}).values()), 2) s["notes"] = f"farm: {n} niches, {arts} articles" elif pid == "creative": q = safe(lambda: json.loads((w / "creative/qualified.json").read_text())) if q: s["metrics"]["prospects_qualified"] = q.get("count", len(q.get("prospects", []))) s["metrics"]["last_run"] = q.get("generated_at", "")[:10] s["notes"] = "demo-site + pentest outreach" elif pid == "govcontracts": p = safe(lambda: json.loads((w / "govcontracts/pipeline.json").read_text())) if p: s["metrics"]["opportunities"] = len(p.get("opportunities", [])) s["metrics"]["submitted_today"] = p.get("submitted_today", 0) s["notes"] = f"submitted today={p.get('submitted_today')}" elif pid == "tradermoe": s["metrics"]["status"] = "trading desk" s["notes"] = "autonomous v2 (see /opt/tradermoe)" elif pid == "redteam": t = safe(lambda: json.loads((w / "redteam/authorized_targets.json").read_text())) if t: s["metrics"]["authorized_targets"] = len(t) if isinstance(t, list) else len(t.get("targets", [])) s["notes"] = "authorized-only pentest" elif pid == "c3nter": s["metrics"]["role"] = "command bot" s["notes"] = "fleet command @commandc3nterbot" elif pid == "coder": s["notes"] = "dev / build agent" elif pid == "hackbox": s["notes"] = "recon helper" elif pid == "claimforge": import json as _j p = safe(lambda: _j.loads((w / "claimforge/data/scored_matches.json").read_text())) if p: s["metrics"]["claims_scored"] = len(p) s["metrics"]["high_value_5k"] = sum(1 for m in p if (m.get("amount_parsed") or 0) >= 5000.0) s["notes"] = "unclaimed-first >= $5K, then corp active/expired" elif pid == "lifeagent": s["notes"] = "life insurance CRM — leads, scoring, calls, pipeline" elif pid == "domainhaus": s["notes"] = "domain intelligence — expired/dropping/backordered domains" elif pid == "default": s["metrics"]["role"] = "orchestrator" s["notes"] = "fleet command" return s def build(): now = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC") agents = [agent_state(pid) for pid, _, _, _ in AGENTS] summary = { "generated_at": now, "profile_count": len(AGENTS), "bot_count": sum(1 for a in agents if a["bot"] != "(own)"), "totals": { "leads": sum(a["metrics"].get("leads_found", 0) for a in agents), "seo_articles": sum(a["metrics"].get("articles_published", 0) for a in agents), "seo_niches": sum(a["metrics"].get("niches_live", 0) for a in agents), "qualified_prospects": sum(a["metrics"].get("prospects_qualified", 0) for a in agents), "gov_opps": sum(a["metrics"].get("opportunities", 0) for a in agents), "billed_usd": sum(a["metrics"].get("billed_usd", 0) for a in agents), }, "agents": agents, } return summary CSS = """ :root{--bg:#0b0e14;--card:#141926;--fg:#e7ebf3;--mut:#8b93a7;--ok:#3ddc84;--warn:#ffb454;--bad:#ff5c5c;} *{box-sizing:border-box}body{margin:0;background:var(--bg);color:var(--fg);font:15px/1.5 -apple-system,Segoe UI,Roboto,Helvetica,Arial,sans-serif} header{padding:20px 28px;background:linear-gradient(120deg,#161b2e,#0b0e14);border-bottom:1px solid #222a3d;display:flex;justify-content:space-between;align-items:center} header h1{margin:0;font-size:20px} .sub{color:var(--mut);font-size:13px} .grid{display:grid;grid-template-columns:repeat(auto-fill,minmax(260px,1fr));gap:16px;padding:22px} .card{background:var(--card);border:1px solid #222a3d;border-radius:14px;padding:16px;position:relative;overflow:hidden} .card .bar{position:absolute;top:0;left:0;width:100%;height:4px} .card h3{margin:8px 0 4px;font-size:17px;display:flex;align-items:center;gap:8px} .glyph{font-size:20px} .bot{color:var(--mut);font-size:12px} .metrics{margin:10px 0 0;font-size:13px} .metrics div{display:flex;justify-content:space-between;padding:3px 0;border-bottom:1px solid #1c2333} .metrics b{color:#fff} .notes{color:var(--mut);font-size:12px;margin-top:8px;font-style:italic} .totals{display:flex;gap:14px;flex-wrap:wrap;padding:0 28px 8px} .tot{background:var(--card);border:1px solid #222a3d;border-radius:12px;padding:12px 18px} .tot b{display:block;font-size:22px;color:#fff}.tot span{color:var(--mut);font-size:12px} .gok{color:var(--ok)!important}.gdown{color:var(--bad)!important} footer{padding:18px 28px;color:var(--mut);font-size:12px;border-top:1px solid #222a3d} """ COLORS = {a[0]: a[2] for a in AGENTS} GLYPHS = {a[0]: a[3] for a in AGENTS} FLEET_NAV_HTML = '' def render_html(summary): tot = summary["totals"] cards = [] for a in summary["agents"]: pid = a["profile"] m = a["metrics"] def _mval(k, v): sv = str(v) if k == "gateway": cls = "gok" if sv == "ONLINE" else "gdown" return f"
{k}{sv}
" return f"
{k.replace('_',' ')}{v}
" mrows = "".join(_mval(k, v) for k, v in m.items()) cards.append(f"""

{GLYPHS[pid]}{html.escape(a['profile'])}

{html.escape(a['bot'])}
{mrows or '
no data yet—
'}
{html.escape(a.get('notes',''))}
""") return f""" Fleet Command Center {FLEET_NAV_HTML}

⚡ Fleet Command Center

{html.escape(summary['generated_at'])} · {summary['profile_count']} profiles · {summary['bot_count']} bots
{tot['leads']}leads found
{tot['seo_articles']}SEO articles
{tot['seo_niches']}niches live
{tot['qualified_prospects']}creative prospects
{tot['gov_opps']}gov opps
${tot['billed_usd']}billed (SEO)
{''.join(cards)}
""" def main(): summary = build() (OUT / "dashboard.json").write_text(json.dumps(summary, indent=2)) (OUT / "dashboard.html").write_text(render_html(summary)) print(f"[+] Fleet dashboard @ {summary['generated_at']}") print(f" profiles={summary['profile_count']} bots={summary['bot_count']}") for a in summary["agents"]: print(f" {a['profile']:14s} {a['bot']:18s} {a['metrics']}") if __name__ == "__main__": main()