406 lines
11 KiB
Python
406 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""Host metrics, button action log, and clock control for the web configurator."""
|
||
import json
|
||
import subprocess
|
||
import threading
|
||
import time
|
||
from collections import deque
|
||
from datetime import datetime
|
||
|
||
from flask import Blueprint, abort, jsonify, request
|
||
|
||
from auth import login_required
|
||
from config_store import load_config, save_config
|
||
from paths import run_dir, scripts_dir
|
||
|
||
system_bp = Blueprint("system", __name__)
|
||
|
||
_WEB_LOG = deque(maxlen=200)
|
||
_LOCK = threading.Lock()
|
||
_SKIP_PREFIXES = ("/static/", "/healthz", "/favicon.ico")
|
||
_prev_cpu = None
|
||
BUTTON_LOG = "button-log.jsonl"
|
||
BUTTON_LOG_LIMIT = 200
|
||
|
||
|
||
def _read_meminfo():
|
||
info = {}
|
||
try:
|
||
with open("/proc/meminfo", encoding="utf-8") as f:
|
||
for line in f:
|
||
key, _, val = line.partition(":")
|
||
info[key.strip()] = int(val.strip().split()[0])
|
||
except OSError:
|
||
pass
|
||
return info
|
||
|
||
|
||
def _cpu_percent():
|
||
global _prev_cpu
|
||
try:
|
||
with open("/proc/stat", encoding="utf-8") as f:
|
||
parts = [int(x) for x in f.readline().split()[1:]]
|
||
except (OSError, ValueError, IndexError):
|
||
return None
|
||
idle = parts[3]
|
||
total = sum(parts)
|
||
if _prev_cpu is None:
|
||
_prev_cpu = (idle, total)
|
||
return None
|
||
idle_d = idle - _prev_cpu[0]
|
||
total_d = total - _prev_cpu[1]
|
||
_prev_cpu = (idle, total)
|
||
if total_d <= 0:
|
||
return 0.0
|
||
return round(100.0 * (1.0 - idle_d / total_d), 1)
|
||
|
||
|
||
def _format_uptime(seconds):
|
||
s = int(seconds)
|
||
days, s = divmod(s, 86400)
|
||
hours, s = divmod(s, 3600)
|
||
minutes, _ = divmod(s, 60)
|
||
parts = []
|
||
if days:
|
||
parts.append(f"{days}д")
|
||
if hours:
|
||
parts.append(f"{hours}ч")
|
||
parts.append(f"{minutes}м")
|
||
return " ".join(parts)
|
||
|
||
|
||
def current_clock():
|
||
now = datetime.now()
|
||
return {
|
||
"datetime": now.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"iso": now.strftime("%Y-%m-%dT%H:%M"),
|
||
}
|
||
|
||
|
||
def collect_stats():
|
||
uptime_secs = 0.0
|
||
load = (0.0, 0.0, 0.0)
|
||
try:
|
||
with open("/proc/uptime", encoding="utf-8") as f:
|
||
uptime_secs = float(f.read().split()[0])
|
||
except (OSError, ValueError, IndexError):
|
||
pass
|
||
try:
|
||
with open("/proc/loadavg", encoding="utf-8") as f:
|
||
load_parts = f.read().split()[:3]
|
||
load = tuple(float(x) for x in load_parts)
|
||
except (OSError, ValueError, IndexError):
|
||
pass
|
||
|
||
mem = _read_meminfo()
|
||
total_kb = mem.get("MemTotal", 0)
|
||
avail_kb = mem.get("MemAvailable", mem.get("MemFree", 0))
|
||
used_kb = max(0, total_kb - avail_kb)
|
||
clock = current_clock()
|
||
|
||
return {
|
||
"clock": clock["datetime"],
|
||
"uptime_secs": round(uptime_secs, 1),
|
||
"uptime": _format_uptime(uptime_secs),
|
||
"cpu_percent": _cpu_percent(),
|
||
"load": {"1m": load[0], "5m": load[1], "15m": load[2]},
|
||
"memory": {
|
||
"total_kb": total_kb,
|
||
"used_kb": used_kb,
|
||
"available_kb": avail_kb,
|
||
"used_percent": round(100.0 * used_kb / total_kb, 1) if total_kb else 0.0,
|
||
},
|
||
"ts": time.time(),
|
||
}
|
||
|
||
|
||
def read_button_log(limit=BUTTON_LOG_LIMIT):
|
||
path = run_dir() / BUTTON_LOG
|
||
if not path.exists():
|
||
return []
|
||
try:
|
||
lines = path.read_text(encoding="utf-8").splitlines()
|
||
except OSError:
|
||
return []
|
||
entries = []
|
||
for line in lines[-limit:]:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
entries.append(json.loads(line))
|
||
except ValueError:
|
||
continue
|
||
entries.reverse()
|
||
return entries
|
||
|
||
|
||
def record_request(method, path, status, duration_ms, remote):
|
||
if any(path.startswith(p) for p in _SKIP_PREFIXES):
|
||
return
|
||
entry = {
|
||
"ts": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
"method": method,
|
||
"path": path,
|
||
"status": status,
|
||
"ms": round(duration_ms, 1),
|
||
"remote": remote,
|
||
}
|
||
with _LOCK:
|
||
_WEB_LOG.appendleft(entry)
|
||
|
||
|
||
def get_request_log():
|
||
with _LOCK:
|
||
return list(_WEB_LOG)
|
||
|
||
|
||
def apply_system_time(dt_str):
|
||
import re
|
||
if not re.fullmatch(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", dt_str):
|
||
raise ValueError("invalid datetime")
|
||
script = str(scripts_dir() / "bt-settime")
|
||
proc = subprocess.run(
|
||
["sudo", "-n", script, dt_str],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=15,
|
||
check=False,
|
||
)
|
||
if proc.returncode != 0:
|
||
err = (proc.stderr or proc.stdout or "set time failed").strip()
|
||
raise RuntimeError(err)
|
||
return proc.stdout.strip()
|
||
|
||
|
||
def _run_access(*args, timeout=20):
|
||
script = str(scripts_dir() / "bt-access")
|
||
proc = subprocess.run(
|
||
["sudo", "-n", script, *args],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=timeout,
|
||
check=False,
|
||
)
|
||
for line in reversed((proc.stdout or "").splitlines()):
|
||
line = line.strip()
|
||
if line.startswith("{") and line.endswith("}"):
|
||
return json.loads(line)
|
||
if proc.returncode != 0:
|
||
err = (proc.stderr or proc.stdout or "bt-access failed").strip()
|
||
raise RuntimeError(err)
|
||
raise RuntimeError((proc.stdout or "bt-access returned no status").strip())
|
||
|
||
|
||
def access_status():
|
||
cfg = load_config()
|
||
web_port = int(cfg.get("settings", {}).get("webPort", 8080))
|
||
status = _run_access("status")
|
||
status["webPort"] = web_port
|
||
return status
|
||
|
||
|
||
MIN_USER_PORT = 1024
|
||
|
||
|
||
def set_web_port(port):
|
||
try:
|
||
port = int(port)
|
||
except (TypeError, ValueError):
|
||
raise ValueError("invalid web port")
|
||
if port < MIN_USER_PORT or port > 65535:
|
||
raise ValueError(f"web port must be {MIN_USER_PORT}–65535")
|
||
cfg = load_config()
|
||
cfg.setdefault("settings", {})["webPort"] = port
|
||
save_config(cfg)
|
||
subprocess.Popen(
|
||
["sudo", "-n", str(scripts_dir() / "bt-service"), "restart"],
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.DEVNULL,
|
||
start_new_session=True,
|
||
)
|
||
return port
|
||
|
||
|
||
def _run_brightness(*args, timeout=10):
|
||
script = str(scripts_dir() / "bt-brightness")
|
||
proc = subprocess.run(
|
||
["sudo", "-n", script, *args],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=timeout,
|
||
check=False,
|
||
)
|
||
for line in reversed((proc.stdout or "").splitlines()):
|
||
line = line.strip()
|
||
if line.startswith("{") and line.endswith("}"):
|
||
return json.loads(line)
|
||
if proc.returncode != 0:
|
||
err = (proc.stderr or proc.stdout or "bt-brightness failed").strip()
|
||
raise RuntimeError(err)
|
||
raise RuntimeError((proc.stdout or "bt-brightness returned no status").strip())
|
||
|
||
|
||
def brightness_status():
|
||
live = _run_brightness("status")
|
||
cfg = load_config()
|
||
saved = cfg.get("settings", {}).get("brightness")
|
||
try:
|
||
saved = int(saved) if saved is not None else None
|
||
except (TypeError, ValueError):
|
||
saved = None
|
||
live["saved"] = saved
|
||
live["ok"] = True
|
||
return live
|
||
|
||
|
||
def set_brightness(percent):
|
||
try:
|
||
percent = int(percent)
|
||
except (TypeError, ValueError):
|
||
raise ValueError("percent must be an integer 0-100")
|
||
if percent < 0 or percent > 100:
|
||
raise ValueError("percent must be 0-100")
|
||
live = _run_brightness("set", str(percent))
|
||
cfg = load_config()
|
||
cfg.setdefault("settings", {})["brightness"] = percent
|
||
save_config(cfg)
|
||
live["saved"] = percent
|
||
live["ok"] = True
|
||
return live
|
||
|
||
|
||
def apply_saved_brightness():
|
||
"""Restore brightness from config at service start. No-op if unset."""
|
||
cfg = load_config()
|
||
saved = cfg.get("settings", {}).get("brightness")
|
||
if saved is None:
|
||
return None
|
||
try:
|
||
percent = int(saved)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if percent < 0 or percent > 100:
|
||
return None
|
||
try:
|
||
return _run_brightness("set", str(percent))
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
@system_bp.route("/api/system/stats", methods=["GET"])
|
||
@login_required
|
||
def api_system_stats():
|
||
return jsonify(collect_stats())
|
||
|
||
|
||
@system_bp.route("/api/system/access", methods=["GET"])
|
||
@login_required
|
||
def api_access_get():
|
||
try:
|
||
return jsonify(access_status())
|
||
except RuntimeError as e:
|
||
return jsonify({"ok": False, "error": str(e)}), 500
|
||
|
||
|
||
@system_bp.route("/api/system/access", methods=["POST"])
|
||
@login_required
|
||
def api_access_set():
|
||
data = request.get_json(force=True, silent=True) or {}
|
||
action = (data.get("action") or "").strip()
|
||
value = data.get("value")
|
||
try:
|
||
if action == "icmp":
|
||
if value not in ("on", "off"):
|
||
abort(400, "icmp value must be on/off")
|
||
return jsonify(_run_access("icmp", value))
|
||
if action == "ssh":
|
||
if value not in ("on", "off"):
|
||
abort(400, "ssh value must be on/off")
|
||
return jsonify(_run_access("ssh", value))
|
||
if action == "sshPort":
|
||
return jsonify(_run_access("ssh-port", str(value)))
|
||
if action == "webPort":
|
||
port = set_web_port(value)
|
||
return jsonify({"ok": True, "webPort": port})
|
||
abort(400, "unknown action")
|
||
except ValueError as e:
|
||
abort(400, str(e))
|
||
except RuntimeError as e:
|
||
return jsonify({"ok": False, "error": str(e)}), 500
|
||
|
||
|
||
@system_bp.route("/api/system/brightness", methods=["GET"])
|
||
@login_required
|
||
def api_brightness_get():
|
||
try:
|
||
return jsonify(brightness_status())
|
||
except RuntimeError as e:
|
||
return jsonify({"ok": False, "error": str(e)}), 500
|
||
|
||
|
||
@system_bp.route("/api/system/brightness", methods=["POST"])
|
||
@login_required
|
||
def api_brightness_set():
|
||
data = request.get_json(force=True, silent=True) or {}
|
||
try:
|
||
return jsonify(set_brightness(data.get("percent")))
|
||
except ValueError as e:
|
||
abort(400, str(e))
|
||
except RuntimeError as e:
|
||
return jsonify({"ok": False, "error": str(e)}), 500
|
||
|
||
|
||
@system_bp.route("/api/system/time", methods=["GET"])
|
||
@login_required
|
||
def api_system_time_get():
|
||
return jsonify(current_clock())
|
||
|
||
|
||
@system_bp.route("/api/system/time", methods=["POST"])
|
||
@login_required
|
||
def api_system_time_set():
|
||
data = request.get_json(force=True, silent=True) or {}
|
||
dt = (data.get("datetime") or "").strip()
|
||
if not dt:
|
||
abort(400, "datetime required (YYYY-MM-DD HH:MM:SS)")
|
||
try:
|
||
apply_system_time(dt)
|
||
except ValueError as e:
|
||
abort(400, str(e))
|
||
except RuntimeError as e:
|
||
return jsonify({"ok": False, "error": str(e)}), 500
|
||
return jsonify({"ok": True, **current_clock()})
|
||
|
||
|
||
def trigger_reboot():
|
||
script = str(scripts_dir() / "bt-reboot")
|
||
subprocess.Popen(
|
||
["sudo", "-n", script],
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.DEVNULL,
|
||
start_new_session=True,
|
||
)
|
||
|
||
|
||
@system_bp.route("/api/system/reboot", methods=["POST"])
|
||
@login_required
|
||
def api_system_reboot():
|
||
try:
|
||
trigger_reboot()
|
||
except FileNotFoundError as e:
|
||
return jsonify({"ok": False, "error": str(e)}), 500
|
||
return jsonify({"ok": True})
|
||
|
||
|
||
@system_bp.route("/api/system/button-log", methods=["GET"])
|
||
@login_required
|
||
def api_system_button_log():
|
||
return jsonify({"entries": read_button_log()})
|
||
|
||
|
||
@system_bp.route("/api/system/log", methods=["GET"])
|
||
@login_required
|
||
def api_system_log():
|
||
return jsonify({"entries": get_request_log()})
|