Sanad_lite/dashboard/app.py

"""FastAPI application — Sanad Dashboard.

Each route module is imported INDIVIDUALLY inside try/except so that one
broken router (missing dep, syntax error in a sibling) cannot break the
entire dashboard. Failed routers are logged and the server starts without
them.
"""

from __future__ import annotations

import importlib
import logging
import secrets

# Backfill asyncio.to_thread on Python 3.8 — must run before any router import.
from Project.Sanad.core import asyncio_compat  # noqa: F401

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from starlette.middleware.sessions import SessionMiddleware

from Project.Sanad.config import BASE_DIR
from Project.Sanad.core.logger import get_logger

log = get_logger("dashboard.app")

from Project.Sanad.core.config_loader import section as _cfg_section
_APP_CFG = _cfg_section("dashboard", "app")
app = FastAPI(
    title=_APP_CFG.get("title", "Sanad Dashboard"),
    version=_APP_CFG.get("version", "1.0.0"),
)

# Auth gate — every request except /login, /api/auth/*, and /static/* needs
# a signed session cookie. HTML routes redirect to /login; API/WS routes 401.
#
# Middleware ordering matters: the LAST `add_middleware` call wraps the
# OUTSIDE of the stack, so we register the gate FIRST (innermost) and
# SessionMiddleware SECOND (outermost). That way the session is populated
# on `request.scope` before the gate dereferences `request.session`.
_AUTH_BYPASS_PREFIXES = ("/login", "/api/auth/", "/static/", "/favicon")


@app.middleware("http")
async def _auth_gate(request: Request, call_next):
    path = request.url.path
    if any(path == p or path.startswith(p) for p in _AUTH_BYPASS_PREFIXES):
        return await call_next(request)
    if request.session.get("user"):
        return await call_next(request)
    if path.startswith("/api/") or path.startswith("/ws/"):
        return JSONResponse({"detail": "Not authenticated"}, status_code=401)
    return RedirectResponse(f"/login?next={path}", status_code=303)


# Cookie session — secret regenerates on every restart, so all sessions
# invalidate on a server restart (acceptable for a local robot dashboard).
# https_only=True means the cookie is only sent over HTTPS — fine when this
# app sits behind a TLS-terminating reverse proxy (LiteSpeed) on the public
# domain. same_site="lax" plus the Secure flag mitigates CSRF on the login form.
app.add_middleware(SessionMiddleware, secret_key=secrets.token_urlsafe(32),
                   session_cookie="sanad_session", max_age=60 * 60 * 24,
                   https_only=True, same_site="lax")


# When LiteSpeed reverse-proxies into uvicorn, the request hits us as HTTP
# from 127.0.0.1. ProxyHeadersMiddleware reads X-Forwarded-Proto / -For so
# request.url.scheme reports https (for redirects) and request.client.host
# reports the real visitor IP (for logs/auth). Only trusts these headers
# when the proxy is on localhost.
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware  # noqa: E402
app.add_middleware(ProxyHeadersMiddleware, trusted_hosts="127.0.0.1")


# -- isolated route registration --

_REST_ROUTES: list[tuple[str, str, str]] = [
    # (module_name, prefix, tag)
    ("auth",            None,                   "auth"),
    ("health",          "/api",                 "health"),
    ("system",          "/api/system",          "system"),
    ("voice",           "/api/voice",           "voice"),
    ("logs",            "/api/logs",            "logs"),
    ("audio_control",   "/api/audio",           "audio"),
    ("scripts",         "/api/scripts",         "scripts"),
    ("records",         "/api/records",         "records"),
    ("prompt",          "/api/prompt",          "prompt"),
    ("typed_replay",    "/api/typed-replay",    "typed-replay"),
]

_WS_ROUTES: list[str] = ["log_stream"]

_loaded_routes: list[str] = []
_failed_routes: dict[str, str] = {}


def _register_router(module_name: str, prefix: str | None = None, tag: str | None = None,
                     package: str = "Project.Sanad.dashboard.routes"):
    """Import + register one router. Failures are logged, never raised."""
    full_name = f"{package}.{module_name}"
    try:
        mod = importlib.import_module(full_name)
        if not hasattr(mod, "router"):
            raise AttributeError(f"{full_name} has no 'router' attribute")
        kwargs: dict = {}
        if prefix is not None:
            kwargs["prefix"] = prefix
        if tag is not None:
            kwargs["tags"] = [tag]
        app.include_router(mod.router, **kwargs)
        _loaded_routes.append(module_name)
        log.info("Registered router: %s", module_name)
    except Exception as exc:
        _failed_routes[module_name] = str(exc)
        log.exception("Failed to register router %s — skipping", module_name)


# REST routes
for mod_name, prefix, tag in _REST_ROUTES:
    _register_router(mod_name, prefix=prefix, tag=tag)

# WebSocket routes
for mod_name in _WS_ROUTES:
    _register_router(
        mod_name,
        package="Project.Sanad.dashboard.websockets",
        tag="websocket",
    )


# -- Static files (dashboard UI) — best effort --
STATIC_DIR = BASE_DIR / _APP_CFG.get("static_subdir", "dashboard/static")
try:
    STATIC_DIR.mkdir(parents=True, exist_ok=True)
    app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
    log.info("Static dir mounted: %s", STATIC_DIR)
except Exception:
    log.exception("Could not mount static dir %s — serving without it", STATIC_DIR)


@app.get("/")
async def root():
    """Serve the dashboard SPA."""
    index = STATIC_DIR / "index.html"
    if index.exists():
        from fastapi.responses import HTMLResponse
        try:
            return HTMLResponse(index.read_text(encoding="utf-8"))
        except OSError as exc:
            return {"error": f"Could not read index.html: {exc}"}
    return {
        "message": "Sanad Dashboard — index.html not found",
        "loaded_routes": _loaded_routes,
        "failed_routes": _failed_routes,
    }


@app.get("/api/_dashboard_status")
async def dashboard_load_status():
    """Diagnostic — which routers loaded, which failed."""
    return {
        "loaded": _loaded_routes,
        "failed": _failed_routes,
        "total_loaded": len(_loaded_routes),
        "total_failed": len(_failed_routes),
    }