web/server.py

"""FastAPI 路由与面板服务。

与 app.py 分离：app.py 负责 lifespan/agent 线程托管，本模块只管
HTTP 路由与文件读写。两者通过参数注入路径，便于测试与复用。
"""

import json
import secrets
import threading
import time
from pathlib import Path

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse, Response

from status import atomic_write_json, read_status

_PANEL_HTML = Path(__file__).parent / "panel.html"


def _tail_lines(path, n=200):
    """读文件最后 n 行；文件不存在返回 []。"""
    try:
        with open(path, "rb") as f:
            data = f.read()
    except FileNotFoundError:
        return []
    except OSError:
        return []
    lines = data.decode("utf-8", errors="replace").splitlines()
    return lines[-n:]


def _read_commands(path):
    """读 commands.json；不存在或损坏返回 []。"""
    try:
        data = json.loads(Path(path).read_text(encoding="utf-8") or "{}")
        return data.get("commands", [])
    except (FileNotFoundError, ValueError, OSError):
        return []


def _write_commands(path, commands):
    """原子写 commands.json（先临时文件再 rename）。"""
    atomic_write_json(path, {"commands": commands})


def build_app(lifespan, *, status_path, commands_path,
              key_events_path, history_path, memory_path,
              web_token=None, command_lock=None):
    app = FastAPI(title="arena-evolve", lifespan=lifespan)
    command_lock = command_lock or threading.Lock()

    def _authorized(request):
        if not web_token:
            return True
        scheme, _, token = request.headers.get("Authorization", "").partition(" ")
        return scheme.lower() == "bearer" and secrets.compare_digest(token, web_token)

    def _unauthorized():
        return JSONResponse({"ok": False, "error": "unauthorized"},
                            status_code=401,
                            headers={"WWW-Authenticate": "Bearer",
                                     "Cache-Control": "no-store"})

    @app.get("/", response_class=HTMLResponse)
    def index():
        return HTMLResponse(_PANEL_HTML.read_text(encoding="utf-8"))

    @app.get("/api/health")
    def health():
        st = getattr(app.state, "stop_event", None)
        thr = getattr(app.state, "agent_thread", None)
        err = getattr(app.state, "agent_error", None)
        return {
            "ok": True,
            "agent_thread_alive": bool(thr and thr.is_alive()),
            "agent_error": err,
            "genes_loaded": getattr(app.state, "genes", None) is not None,
        }

    @app.get("/api/status")
    def status(request: Request):
        if not _authorized(request):
            return _unauthorized()
        # stale_after=60：超过 60s 未更新标记为 stale（agent 卡住/断线）
        snap = read_status(status_path, stale_after=60)
        if snap is None:
            return JSONResponse(
                {"kind": "none", "stale": True,
                 "agent_error": getattr(app.state, "agent_error", None)},
                status_code=200)
        snap["agent_error"] = getattr(app.state, "agent_error", None)
        snap["agent_thread_alive"] = bool(
            getattr(app.state, "agent_thread", None) and
            app.state.agent_thread.is_alive())
        return snap

    @app.get("/api/memory")
    def memory(request: Request):
        if not _authorized(request):
            return _unauthorized()
        headers = {
            "Cache-Control": "no-store",
            "Content-Disposition": 'attachment; filename="memory.json"',
        }
        try:
            raw = Path(memory_path).read_bytes()
        except FileNotFoundError:
            return JSONResponse({"error": "memory snapshot not found"},
                                status_code=404, headers=headers)
        except OSError:
            return JSONResponse({"error": "memory snapshot unavailable"},
                                status_code=503, headers=headers)
        try:
            data = json.loads(raw)
        except (ValueError, UnicodeDecodeError):
            return JSONResponse({"error": "memory snapshot unavailable"},
                                status_code=503, headers=headers)
        if not isinstance(data, dict):
            return JSONResponse({"error": "memory snapshot unavailable"},
                                status_code=503, headers=headers)
        return Response(content=json.dumps(data, ensure_ascii=False),
                        media_type="application/json", headers=headers)

    @app.get("/api/key-events")
    def key_events(request: Request, n: int = 80):
        if not _authorized(request):
            return _unauthorized()
        return {"lines": _tail_lines(key_events_path, n)}

    @app.get("/api/history")
    def history(request: Request, n: int = 50):
        if not _authorized(request):
            return _unauthorized()
        rows = []
        for line in _tail_lines(history_path, n):
            try:
                rows.append(json.loads(line))
            except ValueError:
                continue
        return {"rows": rows}

    @app.post("/api/command")
    async def command(request: Request):
        if not _authorized(request):
            return _unauthorized()
        try:
            body = await request.json()
        except ValueError:
            return JSONResponse({"ok": False, "error": "invalid JSON"},
                                status_code=400)
        ctype = body.get("type")
        allowed = ("mark_resource", "mark_obstacle", "goto",
                   "cancel_goto", "clear_marks", "remove_mark",
                   "core_migrate", "core_auto",
                   "core_station", "core_enable_migration")
        if ctype not in allowed:
            return JSONResponse({"ok": False, "error": f"unknown type: {ctype}"},
                                status_code=400)
        direction = body.get("direction")
        if ctype == "core_migrate" and direction not in (
                "UP", "DOWN", "LEFT", "RIGHT"):
            return JSONResponse({"ok": False, "error": "invalid direction"},
                                status_code=400)
        with command_lock:
            cmds = _read_commands(commands_path)
            if len(cmds) >= 256:
                return JSONResponse({"ok": False, "error": "command queue full"},
                                    status_code=429)
            cmds.append({
                "id": f"{time.time_ns()}",
                "type": ctype,
                "pos": body.get("pos"),
                "uid": body.get("uid"),
                "direction": direction,
            })
            _write_commands(commands_path, cmds)
        return {"ok": True, "pending": len(cmds)}

    @app.get("/api/commands")
    def commands(request: Request):
        if not _authorized(request):
            return _unauthorized()
        with command_lock:
            return {"commands": _read_commands(commands_path)}

    return app