TelegramBotAdmin/wbb/web_admin.py

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