cloudinit-bot
a8ec7b80de
---ci--- project: inboxos phase: 1 milestone: v1.4 status: execute wave: 2 tasks: [T3, T4, T5, T6, T7, T8] ---/ci---
78 lines
2.4 KiB
Python
Executable File
78 lines
2.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""TUI/CLI dashboard for InboxOS operator.
|
|
|
|
v1.4 REQ-INB-50.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
|
|
ARTIFACTS_DIR = os.environ.get("INBOXOS_ARTIFACTS_DIR", "pipelines/_artifacts")
|
|
METRICS_DIR = os.environ.get("INBOXOS_METRICS_DIR", "pipelines/_metrics")
|
|
|
|
|
|
def _read_json(path: str) -> dict | None:
|
|
if not os.path.exists(path):
|
|
return None
|
|
try:
|
|
with open(path) as f:
|
|
return json.load(f)
|
|
except (json.JSONDecodeError, OSError):
|
|
return None
|
|
|
|
|
|
def render() -> str:
|
|
out = []
|
|
out.append("=" * 60)
|
|
out.append(f"InboxOS Operator Dashboard — {os.environ.get('HOSTNAME', '?')}")
|
|
out.append("=" * 60)
|
|
|
|
last_run = _read_json(os.path.join(ARTIFACTS_DIR, "last_run.json"))
|
|
if last_run:
|
|
out.append(f"Last run: {last_run.get('run_id', '?')} ({last_run.get('run_type', '?')})")
|
|
out.append(f" started={last_run.get('started_at', '?')} status={last_run.get('status', '?')}")
|
|
out.append(f" mids: total={last_run.get('mids_total', 0)} errored={last_run.get('mids_errored', 0)}")
|
|
mods = last_run.get("modules", {})
|
|
out.append(f" modules: finance={mods.get('finance', 0)} crm={mods.get('crm', 0)} "
|
|
f"intel={mods.get('intel', 0)} promotions={mods.get('promotions', 0)}")
|
|
else:
|
|
out.append("Last run: (no data)")
|
|
|
|
health = _read_json(os.path.join(METRICS_DIR, "healthcheck.json"))
|
|
if health:
|
|
out.append("")
|
|
out.append(f"Healthcheck: {health.get('last_run_status', '?')}")
|
|
out.append(f" pending={health.get('pending_count', 0)} error_24h={health.get('error_count_24h', 0)}")
|
|
out.append(f" s3_archive_gb={health.get('archive_size_gb', 0)}")
|
|
else:
|
|
out.append("")
|
|
out.append("Healthcheck: (no data)")
|
|
|
|
out.append("=" * 60)
|
|
return "\n".join(out)
|
|
|
|
|
|
def main() -> int:
|
|
p = argparse.ArgumentParser(description="InboxOS operator dashboard")
|
|
p.add_argument("--watch", action="store_true", help="Refresh every 30s")
|
|
args = p.parse_args()
|
|
|
|
if args.watch:
|
|
import time
|
|
while True:
|
|
sys.stdout.write("\033[2J\033[H")
|
|
sys.stdout.write(render() + "\n")
|
|
sys.stdout.flush()
|
|
time.sleep(30)
|
|
else:
|
|
print(render())
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|