from __future__ import annotations import asyncio from pathlib import Path from aiohttp import web import wbb from wbb.admin.api import AdminApi, build_admin_application _runner: web.AppRunner | None = None def _resolve_admin_asset(relative: str) -> Path | None: root = Path(str(getattr(wbb, "ADMIN_WEB_STATIC_DIR", "admin-web/dist"))).resolve() candidate = (root / relative).resolve() if relative and candidate.is_file() and root in candidate.parents: return candidate index = root / "index.html" if index.is_file(): return index return None async def _serve_spa(request: web.Request) -> web.StreamResponse: asset = _resolve_admin_asset(request.match_info.get("tail", "")) if asset is not None: return web.FileResponse(asset) return web.Response( text="Admin web assets are not built. Run `cd admin-web && npm run build`.", status=503, content_type="text/plain", ) async def _redirect_admin(_: web.Request) -> web.StreamResponse: raise web.HTTPFound("/admin/") async def start_admin_web() -> None: global _runner if _runner is not None or not bool(getattr(wbb, "ADMIN_WEB_ENABLED", True)): return application = build_admin_application() api: AdminApi = application["admin_api"] await api.initialize() application.router.add_get("/admin", _redirect_admin) application.router.add_get("/admin/{tail:.*}", _serve_spa) _runner = web.AppRunner(application, access_log=None) await _runner.setup() host = str(getattr(wbb, "ADMIN_WEB_HOST", "0.0.0.0")) port = int(getattr(wbb, "ADMIN_WEB_PORT", 8088)) site = web.TCPSite(_runner, host=host, port=port) await site.start() wbb.log.info(f"Admin web listening on http://{host}:{port}/admin") async def stop_admin_web() -> None: global _runner if _runner is not None: await _runner.cleanup() _runner = None def schedule_admin_web() -> None: try: loop = asyncio.get_running_loop() except RuntimeError: return loop.create_task(start_admin_web(), name="admin-web")