"""Simbiot Harness — дашборд задач: веб-страница + JSON-API. Задача t-260929164722-60dae2 [M]: страница со всеми задачами харнеса (статусы/размеры/уровни), автообновление. Плюс лестница уровней, эскалации и свежие дайджесты — состояние харнеса одним взглядом. Дизайн: stdlib-only (http.server + ThreadingHTTPServer), никаких зависимостей и лишних процессов. Читает событие-стор напрямую через run_projections() (та же проекция, что у scheduler/CLI — консистентно). Порты: 8765 (dashboard) / 8766 (healthcheck, без авторизации), SIMBIOT_DASH_PORT/SIMBIOT_DASH_ADDR/SIMBIOT_DASH_TOKEN — из env. Открытый мир — токен SIMBIOT_DASH_TOKEN (query ?token= или cookie): страница и API без токена отдают 401 (кроме порта 8766 — там только "ok", без данных; healthcheck для systemd не должен падать на auth). """ import json import os import sys import threading import time from http import cookies from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from harness import commands, es # noqa: E402 _HERE = os.path.dirname(os.path.abspath(__file__)) _INDEX = os.path.join(_HERE, "dashboard.html") _ADDR = os.environ.get("SIMBIOT_DASH_ADDR", "0.0.0.0") _PORT = int(os.environ.get("SIMBIOT_DASH_PORT", "8765")) _HEALTH_PORT = int(os.environ.get("SIMBIOT_DASH_HEALTH_PORT", "8766")) _TOKEN = os.environ.get("SIMBIOT_DASH_TOKEN", "") _REFRESH = int(os.environ.get("SIMBIOT_DASH_REFRESH", "5")) _STATUS_ORDER = {"ready": 0, "running": 1, "failed": 2, "done": 3} _SIZE_COLOR = {"S": "#1d7a3e", "M": "#a3690f", "L": "#FF4F00", "XL": "#b3261e"} def _utcnow() -> str: return time.strftime("%Y-%m-%dT%H:%M:%S+00:00", time.gmtime()) def build_state() -> dict: """Срез харнеса для API: задачи, лестница, эскалации, дайджесты, спенд. Полный run_projections() на каждый запрос — безопасно: инкрементальный чекпоинт в БД, работа пропорциональна дельте событий, не истории. """ model = es.run_projections() tasks = [] for t in model.get("tasks", {}).values(): tasks.append({ "id": t.get("id"), "title": t.get("title", ""), "size": t.get("size", "?"), "status": t.get("status", "?"), "level": t.get("level"), "created": t.get("created"), "finished": t.get("finished"), "error": t.get("error"), }) # два стабильных прохода: новые сверху внутри статуса, # статусы по порядку ready→running→failed→done tasks.sort(key=lambda x: x.get("created") or "", reverse=True) tasks.sort(key=lambda x: _STATUS_ORDER.get(x["status"], 9)) ladder = [] for name, cs in sorted(model.get("cycle_state", {}).items()): ladder.append({ "level": name, "busy": bool(cs.get("busy")), "paused": bool(cs.get("paused")), "ticks": cs.get("ticks", 0), "absorbed": cs.get("absorbed", 0), "errors": cs.get("errors", 0), "last_tick": cs.get("last_tick"), "last_error": cs.get("last_error"), }) escalations = [e for e in (model.get("escalations") or {}).values() if e.get("status") == "open"] escalations.sort(key=lambda e: e.get("ts", ""), reverse=True) digests = [] for d in model.get("digests", {}).values(): digests.append({ "level": d.get("level"), "period": d.get("period"), "text": d.get("text", ""), "ts": d.get("ts"), "summary_ref": d.get("summary_ref"), }) digests.sort(key=lambda d: d.get("ts") or "", reverse=True) spend = model.get("spend", [])[-12:] counts = {} for t in tasks: key = f"{t['status']}:{t['size']}" counts[key] = counts.get(key, 0) + 1 return { "generated_at": _utcnow(), "counts": { "tasks_total": len(tasks), "tasks_by_status": { s: sum(1 for t in tasks if t["status"] == s) for s in ("ready", "running", "failed", "done") }, "tasks_by_status_size": counts, "escalations_open": len(escalations), "events_total": es.count_events(), "digests_total": len(model.get("digests", {})), }, "ladder": ladder, "tasks": tasks, "escalations": escalations, "digests": digests[:12], "spend": spend, "meta": {"refresh_seconds": _REFRESH, "token_required": bool(_TOKEN)}, } class _Handler(BaseHTTPRequestHandler): server_version = "SimbiotDash/1.0" def log_message(self, fmt, *args): # тише в journalctl: одна строка print(f"[dash] {self.address_string()} {fmt % args}", flush=True) # ── auth ── def _authorized(self) -> bool: if not _TOKEN: return True supplied = None qs = self.path.partition("?")[2] from urllib.parse import parse_qs for v in parse_qs(qs).get("token", []): supplied = v if supplied is None: c = cookies.SimpleCookie(self.headers.get("Cookie", "")) if "simbiot_token" in c: supplied = c["simbiot_token"].value return supplied == _TOKEN def _forbidden(self): body = b'{"error": "unauthorized: pass ?token=SIMBIOT_DASH_TOKEN"}\n' self.send_response(401) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("WWW-Authenticate", 'Bearer realm="simbiot-dash"') self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _json(self, obj, code=200): body = json.dumps(obj, ensure_ascii=False, default=str).encode() self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _html(self, text: str): body = text.encode() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) # ── routes ── def do_GET(self): route = self.path.partition("?")[0] if route == "/api/state": if not self._authorized(): return self._forbidden() try: return self._json(build_state()) except Exception as e: # стор бит/занят — 503, фронт покажет retry return self._json({"error": str(e)[:500]}, code=503) if route == "/api/health": return self._json({"ok": True, "ts": _utcnow()}) if route in ("/", "/index.html"): if not self._authorized(): return self._forbidden() try: with open(_INDEX, encoding="utf-8") as f: return self._html(f.read()) except OSError as e: return self._json({"error": f"dashboard.html not found: {e}"}, code=500) return self._json({"error": "not found"}, code=404) def _health_server(): """Отдельный мини-порт для healthcheck: без auth, только {ok:true}.""" class H(BaseHTTPRequestHandler): def log_message(self, fmt, *args): pass def do_GET(self): body = b'{"ok":true}' self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) srv = ThreadingHTTPServer(("127.0.0.1", _HEALTH_PORT), H) srv.serve_forever() def main(): es.init_db() # идемпотентно if not os.path.exists(_INDEX): print(f"[dash] FATAL: {_INDEX} не найден", file=sys.stderr) sys.exit(1) threading.Thread(target=_health_server, daemon=True).start() print(f"[dash] http://{_ADDR}:{_PORT}/ (health 127.0.0.1:{_HEALTH_PORT}/api/health) " f"refresh={_REFRESH}s token={'on' if _TOKEN else 'off'}", flush=True) ThreadingHTTPServer((_ADDR, _PORT), _Handler).serve_forever() if __name__ == "__main__": main()