supervisor-simulator/tools/ndbs/server/main.py

from __future__ import annotations

import os
import subprocess
from typing import Any, Dict, List, Optional

from fastapi import APIRouter, Depends, FastAPI, HTTPException, Query, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

from ndbs_core import (
    ALLOWED_DEFINITION_EXTS,
    ALLOWED_RULE_EXTS,
    NdbsPaths,
    create_rule as core_create_rule,
    ensure_within_root,
    infer_schema,
    list_definition_files,
    list_rule_files,
    load_definition_file,
    resolve_paths,
    write_definition_file,
)


class FileContentRequest(BaseModel):
    filename: str
    content: Any


class RuleCreateRequest(BaseModel):
    rule_id: str


class RuleContentRequest(BaseModel):
    filename: str
    content: str


api_router = APIRouter()
ui_router = APIRouter()


def get_paths(request: Request) -> NdbsPaths:
    return request.app.state.paths


def create_app(paths: Optional[NdbsPaths] = None) -> FastAPI:
    app = FastAPI(title="NDBS Server")
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    resolved_paths = paths or resolve_paths()
    app.state.paths = resolved_paths
    if resolved_paths.web_root.exists():
        app.mount(
            "/static",
            StaticFiles(directory=str(resolved_paths.web_root)),
            name="static",
        )
    app.include_router(ui_router)
    app.include_router(api_router, prefix="/api")
    return app


@ui_router.get("/")
def serve_index(paths: NdbsPaths = Depends(get_paths)) -> FileResponse:
    index_path = paths.web_root / "index.html"
    if not index_path.exists():
        raise HTTPException(status_code=500, detail="UI assets are missing.")
    return FileResponse(index_path)


@api_router.get("/files/definitions")
def list_definitions(paths: NdbsPaths = Depends(get_paths)) -> Dict[str, List[str]]:
    try:
        files = list_definition_files(paths)
    except FileNotFoundError:
        raise HTTPException(status_code=500, detail="Definitions root not found.")
    return {"files": files}


@api_router.get("/files/content")
def get_file_content(
    filename: str = Query(...),
    paths: NdbsPaths = Depends(get_paths),
) -> Dict[str, Any]:
    try:
        path = ensure_within_root(paths.definitions_root, filename, ALLOWED_DEFINITION_EXTS)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    if not path.exists():
        raise HTTPException(status_code=404, detail="File not found.")
    try:
        content = load_definition_file(path)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    return {"filename": filename, "content": content}


@api_router.post("/files/content")
def save_file_content(
    payload: FileContentRequest,
    paths: NdbsPaths = Depends(get_paths),
) -> Dict[str, str]:
    try:
        path = ensure_within_root(
            paths.definitions_root, payload.filename, ALLOWED_DEFINITION_EXTS
        )
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    if not path.exists():
        raise HTTPException(status_code=404, detail="File not found.")
    try:
        write_definition_file(path, payload.content, paths)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    return {"status": "ok"}


@api_router.get("/files/schema")
def get_file_schema(
    filename: str = Query(...),
    paths: NdbsPaths = Depends(get_paths),
) -> Dict[str, Any]:
    try:
        path = ensure_within_root(paths.definitions_root, filename, ALLOWED_DEFINITION_EXTS)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    if not path.exists():
        raise HTTPException(status_code=404, detail="File not found.")
    content = load_definition_file(path)
    return {"filename": filename, "schema": infer_schema(content)}


@api_router.post("/rules/create")
def create_rule(
    payload: RuleCreateRequest,
    paths: NdbsPaths = Depends(get_paths),
) -> Dict[str, str]:
    try:
        filename = core_create_rule(payload.rule_id, paths)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    except FileExistsError:
        raise HTTPException(status_code=409, detail="Rule already exists.")
    return {"status": "ok", "filename": filename}


@api_router.get("/rules/list")
def list_rules(paths: NdbsPaths = Depends(get_paths)) -> Dict[str, List[str]]:
    return {"files": list_rule_files(paths)}


@api_router.get("/rules/content")
def get_rule_content(
    filename: str = Query(...),
    paths: NdbsPaths = Depends(get_paths),
) -> Dict[str, str]:
    try:
        path = ensure_within_root(paths.rules_root, filename, ALLOWED_RULE_EXTS)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    if not path.exists():
        raise HTTPException(status_code=404, detail="Rule not found.")
    return {"filename": filename, "content": path.read_text(encoding="utf-8")}


@api_router.post("/rules/content")
def save_rule_content(
    payload: RuleContentRequest,
    paths: NdbsPaths = Depends(get_paths),
) -> Dict[str, str]:
    try:
        path = ensure_within_root(paths.rules_root, payload.filename, ALLOWED_RULE_EXTS)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    if not path.exists():
        raise HTTPException(status_code=404, detail="Rule not found.")
    path.write_text(payload.content, encoding="utf-8")
    return {"status": "ok"}


@api_router.post("/build")
def build_game(paths: NdbsPaths = Depends(get_paths)) -> Dict[str, Any]:
    if os.getenv("NDBS_DISABLE_BUILD") == "1":
        raise HTTPException(status_code=503, detail="Build disabled.")
    result = subprocess.run(
        ["dotnet", "build"],
        cwd=str(paths.project_root),
        capture_output=True,
        text=True,
        check=False,
    )
    return {
        "success": result.returncode == 0,
        "stdout": result.stdout,
        "stderr": result.stderr,
    }


app = create_app()


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)