from __future__ import annotations
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib import error, request
from urllib.parse import parse_qs, urlparse
from .config import ConfigStore
from .exports import investigation_report, investigation_report_markdown
from .graylog_mcp import GraylogMcpClient
from .metrics import prometheus_metrics
from .feedback import FeedbackStore
from .incidents import IncidentStore
def _ollama_models(host: str = "http://127.0.0.1:11434") -> dict[str, object]:
try:
req = request.Request(f"{host.rstrip('/')}/api/tags", method="GET")
with request.urlopen(req, timeout=5) as response:
payload = json.loads(response.read().decode("utf-8"))
except error.URLError as exc:
return {"status": "error", "error": str(exc), "models": []}
except (json.JSONDecodeError, OSError) as exc:
return {"status": "error", "error": str(exc), "models": []}
models = payload.get("models", [])
if not isinstance(models, list):
models = []
output = []
for item in models:
if not isinstance(item, dict):
continue
output.append({
"name": str(item.get("name", "")),
"modified_at": str(item.get("modified_at", "")),
"size": int(item.get("size", 0) or 0),
})
return {"status": "ok", "models": sorted(output, key=lambda model: model["name"])}
HTML = """
SignalScope Monitor
Live Status
Waiting for monitor data.
Events and Anomalies
Baseline and Stream Health
AI Assessment
LLM assessment disabled.
Related Activity Across Sources
Recommended Stream Profiles
Waiting for observed stream data.
Installed Ollama Models
Loading local Ollama models.
"""
def serve_dashboard(host: str, port: int, status_file: str, *, image_dir: str | None = None, config_file: str = "state/fgai-config.json") -> None:
status_path = Path(status_file)
image_root = Path(image_dir) if image_dir else None
config_store = ConfigStore(config_file)
class Handler(BaseHTTPRequestHandler):
def do_GET(self) -> None:
if self.path == "/":
self._send(200, "text/html; charset=utf-8", HTML.encode("utf-8"))
return
if self.path == "/api/config":
self._send(200, "application/json", json.dumps(config_store.public()).encode("utf-8"))
return
if self.path == "/api/feedback":
self._send(200, "application/json", json.dumps(FeedbackStore().entries()).encode("utf-8"))
return
if self.path == "/api/incidents":
self._send(200, "application/json", json.dumps(IncidentStore().entries()).encode("utf-8"))
return
if self.path == "/api/ollama/models":
self._send(200, "application/json", json.dumps(_ollama_models()).encode("utf-8"))
return
if self.path == "/api/graylog/streams":
config = config_store.read()
try:
client = GraylogMcpClient(str(config.get("graylog_mcp_url", "")), str(config.get("graylog_mcp_token", "")))
client.probe()
result = client.call_tool("list_streams", {})
content = result.get("result", {}).get("content", [])
text = next((item.get("text", "") for item in content if isinstance(item, dict)), "")
if isinstance(result.get("result"), dict) and result["result"].get("isError"):
fallback = client.call_tool("list_resource", {"resource_type": "stream"})
resources = fallback.get("result", {}).get("structuredContent", {}).get("resources", [])
streams = [
{"id": str(item.get("uri", "")).rsplit(":", 1)[-1], "title": item.get("title", item.get("name", "")), "description": item.get("description", "")}
for item in resources if isinstance(item, dict) and item.get("uri")
]
else:
if not text:
raise RuntimeError("Graylog list_streams returned no text content")
streams = json.loads(text)
body = {"streams": streams, "selected": config.get("graylog_streams", [])}
self._send(200, "application/json", json.dumps(body).encode("utf-8"))
except Exception as exc:
self._send(400, "application/json", json.dumps({"error": str(exc)}).encode("utf-8"))
return
if self.path.startswith("/api/graylog/fields?"):
stream_id = self.path.split("stream_id=", 1)[-1].split("&", 1)[0]
config = config_store.read()
try:
client = GraylogMcpClient(str(config.get("graylog_mcp_url", "")), str(config.get("graylog_mcp_token", "")))
client.probe()
result = client.call_tool("list_fields", {"streams": [stream_id]})
content = result.get("result", {}).get("content", [])
text = next((item.get("text", "") for item in content if isinstance(item, dict)), "")
payload = json.loads(text)
fields = payload.get("fields", payload) if isinstance(payload, dict) else payload
self._send(200, "application/json", json.dumps({"fields": fields}).encode("utf-8"))
except Exception as exc:
self._send(400, "application/json", json.dumps({"error": str(exc)}).encode("utf-8"))
return
if self.path == "/api/status":
if status_path.exists():
body = status_path.read_bytes()
else:
body = json.dumps({"summary": {}, "anomalies": [], "block_candidates": []}).encode("utf-8")
self._send(200, "application/json", body)
return
if self.path.startswith("/api/export/incidents"):
parsed = urlparse(self.path)
params = parse_qs(parsed.query)
fmt = params.get("format", ["markdown"])[0]
incident_id = params.get("incident_id", [""])[0] or None
try:
status = json.loads(status_path.read_text(encoding="utf-8")) if status_path.exists() else {}
except json.JSONDecodeError:
status = {}
report = investigation_report(status, incident_id=incident_id)
if fmt == "json":
self._send(200, "application/json", json.dumps(report, indent=2, sort_keys=True).encode("utf-8"))
else:
self._send(200, "text/markdown; charset=utf-8", investigation_report_markdown(report).encode("utf-8"))
return
if self.path == "/metrics":
try:
status = json.loads(status_path.read_text(encoding="utf-8")) if status_path.exists() else {}
except json.JSONDecodeError:
status = {}
self._send(200, "text/plain; version=0.0.4; charset=utf-8", prometheus_metrics(status).encode("utf-8"))
return
if self.path.startswith("/images/") and image_root:
image_path = image_root / Path(self.path).name
if image_path.exists() and image_path.is_file():
content_type = "image/png" if image_path.suffix.lower() == ".png" else "application/octet-stream"
self._send(200, content_type, image_path.read_bytes())
return
self._send(404, "text/plain; charset=utf-8", b"not found")
def do_POST(self) -> None:
if self.path == "/api/feedback" and self._is_loopback_client():
try:
payload = json.loads(self.rfile.read(min(int(self.headers.get("Content-Length", "0")), 16_384)).decode("utf-8"))
self._send(200, "application/json", json.dumps(FeedbackStore().add(payload)).encode("utf-8"))
except (ValueError, json.JSONDecodeError) as exc:
self._send(400, "application/json", json.dumps({"error": str(exc)}).encode("utf-8"))
return
if self.path == "/api/incidents" and self._is_loopback_client():
try:
payload = json.loads(self.rfile.read(min(int(self.headers.get("Content-Length", "0")), 16_384)).decode("utf-8"))
self._send(200, "application/json", json.dumps(IncidentStore().update(str(payload.get("id", "")), str(payload.get("status", "")), str(payload.get("note", "")))).encode("utf-8"))
except (ValueError, json.JSONDecodeError) as exc:
self._send(400, "application/json", json.dumps({"error": str(exc)}).encode("utf-8"))
return
if self.path != "/api/config" or not self._is_loopback_client():
self._send(403, "application/json", b'{"error":"configuration is local-only"}')
return
try:
length = int(self.headers.get("Content-Length", "0"))
payload = json.loads(self.rfile.read(min(length, 32_768)).decode("utf-8"))
if not isinstance(payload, dict):
raise ValueError("configuration must be an object")
body = json.dumps(config_store.update(payload)).encode("utf-8")
except (ValueError, json.JSONDecodeError) as exc:
self._send(400, "application/json", json.dumps({"error": str(exc)}).encode("utf-8"))
return
self._send(200, "application/json", body)
def _is_loopback_client(self) -> bool:
return self.client_address[0] in {"127.0.0.1", "::1"}
def log_message(self, format: str, *args: object) -> None:
return
def _send(self, status: int, content_type: str, body: bytes) -> None:
self.send_response(status)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
server = ThreadingHTTPServer((host, port), Handler)
print(f"Dashboard listening on http://{host}:{port}")
print(f"Reading status from {status_path}")
server.serve_forever()