arm/atxpi/webui/server.py

#!/usr/bin/env python3
"""atxpi — web-плоскость управления (FastAPI).

Отдаёт:
  /            браузерный UI (index.html)
  /static/*    статика (в т.ч. vendored xterm.js)
  /api/power   REST: GET — статус, POST on|off|reset|cycle
  /api/status  сводка по устройству
  /api/history журнал действий (audit)
  /ws/console  WebSocket — двунаправленная RS232-консоль
  /mcp         MCP-эндпоинт (streamable-http) для агентов

Аутентификация: Bearer-токен (ATXPI_WEB_TOKEN) в заголовке Authorization
(для /ws/console также принимается ?token=...).
"""
import asyncio
import json
import logging
import os
import time
from pathlib import Path

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles

from atxpi.core import power as power_mod
from atxpi.core.hardware import make_console, DRIVER, SERIAL_PORT, SERIAL_BAUD

from mcp.server.mcpserver import MCPServer

log = logging.getLogger("atxpi.webui")
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

STATIC = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static")
WEB_TOKEN = os.environ.get("ATXPI_WEB_TOKEN", "")
AUDIT = os.environ.get("ATXPI_AUDIT", "/etc/atxpi/audit.jsonl")


def audit(action: str, **kw):
    """Логируем действие в jsonl (для /api/history)."""
    entry = {"t": time.time(), "action": action, **kw}
    try:
        Path(AUDIT).parent.mkdir(parents=True, exist_ok=True)
        with open(AUDIT, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    except Exception as e:
        log.warning("audit: %s", e)
    return entry


def _history(n: int = 100):
    if not os.path.exists(AUDIT):
        return []
    try:
        with open(AUDIT, "r", encoding="utf-8") as f:
            lines = f.readlines()[-n:]
        return [json.loads(l) for l in lines if l.strip()]
    except Exception:
        return []


app = FastAPI(title="atxpi", version="0.1.0")


@app.middleware("http")
async def auth_mw(request: Request, call_next):
    """Требуем Bearer-токен для /api/* и /mcp/* (если токен задан)."""
    path = request.url.path
    if path.startswith(("/api", "/mcp")) and WEB_TOKEN:
        got = request.headers.get("authorization", "").replace("Bearer ", "").strip()
        if got != WEB_TOKEN:
            return JSONResponse({"detail": "unauthorized"}, status_code=401)
    return await call_next(request)


# ── REST: питание ──────────────────────────────────────────────────────────
@app.get("/api/power")
async def api_power_status():
    return power_mod.status()


@app.post("/api/power/on")
async def api_power_on():
    audit("power_on")
    return power_mod.press_power()


@app.post("/api/power/off")
async def api_power_off():
    audit("power_off")
    return power_mod.force_off()


@app.post("/api/power/reset")
async def api_power_reset():
    audit("power_reset")
    return power_mod.press_reset()


@app.post("/api/power/cycle")
async def api_power_cycle():
    audit("power_cycle")
    return power_mod.cycle()


@app.get("/api/status")
async def api_status():
    s = power_mod.status()
    return {
        "power": s["power"],
        "powered_on": s["powered_on"],
        "driver": DRIVER,
        "serial_port": SERIAL_PORT,
        "serial_baud": SERIAL_BAUD,
    }


@app.get("/api/history")
async def api_history():
    return _history()


# ── WebSocket консоль ───────────────────────────────────────────────────────
@app.websocket("/ws/console")
async def ws_console(ws: WebSocket):
    token = ws.query_params.get("token", "")
    if WEB_TOKEN and token != WEB_TOKEN:
        await ws.close(code=1008)
        return
    await ws.accept()
    console = make_console()
    try:
        console.open()
    except Exception as e:
        await ws.send_text(json.dumps({"error": f"console open failed: {e}"}))
        await ws.close()
        return

    console.send(("\r\nATXPI console ready. driver=%s\r\n" % DRIVER).encode())

    async def reader():
        try:
            while True:
                data = await asyncio.to_thread(console.read, 0.25)
                if data:
                    await ws.send_bytes(data)
        except Exception:
            pass

    read_task = asyncio.create_task(reader())
    try:
        while True:
            msg = await ws.receive()
            if msg["type"] == "websocket.disconnect":
                break
            data = msg.get("bytes") or msg.get("text")
            if data:
                if isinstance(data, str):
                    data = data.encode()
                console.send(data)
    except WebSocketDisconnect:
        pass
    finally:
        read_task.cancel()
        try:
            await ws.close()
        except Exception:
            pass
        console.close()


# ── MCP (streamable-http endpoint, для агентов) ─────────────────────────────
mcp = MCPServer("atxpi")


@mcp.tool()
def power_status() -> dict:
    """Текущее состояние питания сервера (on/off)."""
    return power_mod.status()


@mcp.tool()
def power_on() -> dict:
    """Нажать кнопку питания (вкл/выкл-переключатель)."""
    audit("power_on")
    return power_mod.press_power()


@mcp.tool()
def power_off() -> dict:
    """Принудительное выключение (длинное нажатие кнопки питания)."""
    audit("power_off")
    return power_mod.force_off()


@mcp.tool()
def power_reset() -> dict:
    """Аппаратный сброс сервера."""
    audit("power_reset")
    return power_mod.press_reset()


@mcp.tool()
def power_cycle() -> dict:
    """Жёсткий перезапуск: выключить, пауза, включить."""
    audit("power_cycle")
    return power_mod.cycle()


app.mount("/mcp", mcp.streamable_http_app())


# ── статика ─────────────────────────────────────────────────────────────────
@app.get("/")
async def index():
    return FileResponse(os.path.join(STATIC, "index.html"))


app.mount("/static", StaticFiles(directory=STATIC), name="static")