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-консоль

Аутентификация: Bearer-токен (ATXPI_WEB_TOKEN) в заголовке Authorization
(для /ws/console также принимается ?token=...).

MCP-эндпоинт (/mcp) реализуется отдельным процессом (см. atxpi/mcp_server.py),
т.к. mcp-SDK требует собственного run() для streamable-http транспорта.
"""
import asyncio
import json
import logging
import os
import time
from pathlib import Path

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request, HTTPException
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

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", "")
MODE = os.environ.get("ATXPI_MODE", "full")   # full | console (только консоль, питание отключено)
AUDIT = os.environ.get("ATXPI_AUDIT", "/etc/atxpi/audit.jsonl")


def _power_guard():
    """В режиме console питание выключено — деструктивные действия запрещены."""
    if MODE == "console":
        raise HTTPException(status_code=403, detail="console-only mode: power disabled")


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/* (если токен задан)."""
    path = request.url.path
    if path.startswith("/api") 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():
    _power_guard()
    return power_mod.status()


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


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


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


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


@app.get("/api/status")
async def api_status():
    try:
        s = power_mod.status()
    except Exception as e:
        log.warning("power status unavailable: %s", e)
        s = {"power": "unknown", "powered_on": None}
    return {
        "power": s["power"],
        "powered_on": s["powered_on"],
        "driver": DRIVER,
        "serial_port": SERIAL_PORT,
        "serial_baud": SERIAL_BAUD,
        "mode": MODE,
    }


@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

    # баннер шлём клиенту (в ws), а НЕ в порт — не пишем на экран сервера.
    await ws.send_bytes(("\r\nATXPI console ready. driver=%s\r\n" % DRIVER).encode())

    async def reader():
        while True:
            try:
                data = await asyncio.to_thread(console.read, 0.25)
            except Exception as e:
                log.warning("console.read error: %s", e)
                break
            if data:
                try:
                    await ws.send_bytes(data)
                except Exception as e:
                    log.warning("ws handle_console send error: %s", e)
                    break

    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()


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


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