#!/usr/bin/env python3 """ Aria Dashboard — minimal HTML Dashboard for tester VPS. Runs on 0.0.0.0:9879 with HTTP Basic Auth. Endpoints: GET / -> HTML UI (auto-refresh) GET /api/status -> JSON: all aria-* services + tmux + last_chat GET /api/logs?s=X -> JSON: last 200 lines of journalctl for service X POST /api/restart -> body service=X, restarts that aria-* service User/PW from ~/.env (ARIA_DASH_USER, ARIA_DASH_PW); defaults: aria / changeme """ import base64 import json import os import subprocess import sys import time from http.server import HTTPServer, BaseHTTPRequestHandler from socketserver import ThreadingMixIn from urllib.parse import urlparse, parse_qs PORT = 9879 ENV_FILE = os.path.expanduser("~/.env") def env(key, default=""): if os.path.exists(ENV_FILE): for line in open(ENV_FILE): if line.startswith(key + "="): return line.split("=", 1)[1].strip().strip('"').strip("'") return os.environ.get(key, default) USER = env("ARIA_DASH_USER", "aria") PW = env("ARIA_DASH_PW", "changeme") ARIA_SERVICES = [ "aria.service", "aria-chat-logger.service", "aria-health.service", ] def svc_state(name): try: active = subprocess.run(["systemctl", "is-active", name], capture_output=True, text=True, timeout=3).stdout.strip() enabled = subprocess.run(["systemctl", "is-enabled", name], capture_output=True, text=True, timeout=3).stdout.strip() return {"name": name, "active": active, "enabled": enabled} except Exception as e: return {"name": name, "active": "unknown", "enabled": "unknown", "error": str(e)} def tmux_status(): r = subprocess.run(["tmux", "has-session", "-t", "aria"], capture_output=True) return "online" if r.returncode == 0 else "offline" def last_chat_ts(): db = os.path.expanduser("~/.aria-chat.db") if not os.path.exists(db): return None try: import sqlite3 conn = sqlite3.connect(db, timeout=3) row = conn.execute( "SELECT created_at FROM aria_chat_log ORDER BY created_at DESC LIMIT 1" ).fetchone() conn.close() return row[0] if row else None except Exception: return None def journal_lines(service, lines=200): if not service.startswith("aria"): return ["(refused: only aria-* services)"] try: r = subprocess.run( ["journalctl", "-u", service, "-n", str(lines), "--no-pager", "--output=short-iso"], capture_output=True, text=True, timeout=5, ) return r.stdout.splitlines()[-lines:] or ["(no log entries)"] except Exception as e: return [f"(journalctl failed: {e})"] HTML = """ Aria Dashboard

Aria Dashboard —

tmux aria-session…
letzte Chat-Nachricht…
Uptime…
Services
Logs 
(Service wählen + „laden")

Auto-Refresh alle 5s · Dashboard → HTTP Basic-Auth aus ~/.env

""" class Handler(BaseHTTPRequestHandler): def _auth_ok(self): hdr = self.headers.get("Authorization", "") if not hdr.startswith("Basic "): return False try: raw = base64.b64decode(hdr.split(" ", 1)[1]).decode() u, p = raw.split(":", 1) return u == USER and p == PW except Exception: return False def _send(self, code, body, ctype="application/json"): b = body if isinstance(body, bytes) else body.encode() self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(b))) self.end_headers() self.wfile.write(b) def _unauth(self): self.send_response(401) self.send_header("WWW-Authenticate", 'Basic realm="aria-dashboard"') self.end_headers() def do_GET(self): if not self._auth_ok(): return self._unauth() p = urlparse(self.path) if p.path in ("/", "/index.html"): return self._send(200, HTML, "text/html; charset=utf-8") if p.path == "/api/status": services = [svc_state(s) for s in ARIA_SERVICES] try: up = subprocess.run(["uptime", "-p"], capture_output=True, text=True, timeout=2).stdout.strip() except Exception: up = "" return self._send(200, json.dumps({ "tmux": tmux_status(), "services": services, "last_chat": last_chat_ts(), "uptime": up, "tester_code": env("ARIA_TESTER_CODE", ""), "ts": int(time.time()), })) if p.path == "/api/logs": q = parse_qs(p.query) svc = q.get("s", [""])[0] return self._send(200, json.dumps({"service": svc, "lines": journal_lines(svc)})) return self._send(404, json.dumps({"error": "not found"})) def do_POST(self): if not self._auth_ok(): return self._unauth() if self.path != "/api/restart": return self._send(404, json.dumps({"error": "not found"})) length = int(self.headers.get("Content-Length", "0")) body = self.rfile.read(length).decode() params = parse_qs(body) svc = params.get("service", [""])[0] if not svc.startswith("aria") or svc not in ARIA_SERVICES: return self._send(400, json.dumps({"ok": False, "error": "refused: only whitelisted aria-* services"})) try: subprocess.run(["systemctl", "restart", svc], check=True, timeout=15) return self._send(200, json.dumps({"ok": True, "service": svc})) except Exception as e: return self._send(500, json.dumps({"ok": False, "error": str(e)})) def log_message(self, fmt, *args): return class ThreadedHTTPServer(ThreadingMixIn, HTTPServer): daemon_threads = True if __name__ == "__main__": print(f"aria-dashboard listening on 0.0.0.0:{PORT} (user={USER})", file=sys.stderr) ThreadedHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()