import logging import os import re from typing import Any from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from api.platform_router import router as platform_router from api.sys_router import router as sys_router from api.system_runtime_router import build_system_runtime_router from api.topic_router import router as topic_router from bootstrap.app_runtime import assemble_app_runtime from core.config_manager import BotConfigManager from core.docker_manager import BotDockerManager from core.settings import BOTS_WORKSPACE_ROOT, DATA_ROOT from core.speech_service import WhisperSpeechService app = FastAPI(title="Dashboard Nanobot API") logger = logging.getLogger("dashboard.backend") LAST_ACTION_MAX_LENGTH = 16000 def _normalize_last_action_text(value: Any) -> str: text = str(value or "").replace("\r\n", "\n").replace("\r", "\n").strip() if not text: return "" text = re.sub(r"\n{4,}", "\n\n\n", text) return text[:LAST_ACTION_MAX_LENGTH] def _apply_log_noise_guard() -> None: for name in ( "httpx", "httpcore", "uvicorn.access", "watchfiles.main", "watchfiles.watcher", ): logging.getLogger(name).setLevel(logging.WARNING) _apply_log_noise_guard() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) app.include_router(topic_router) app.include_router(platform_router) app.include_router(sys_router) os.makedirs(BOTS_WORKSPACE_ROOT, exist_ok=True) os.makedirs(DATA_ROOT, exist_ok=True) docker_manager = BotDockerManager(host_data_root=BOTS_WORKSPACE_ROOT) config_manager = BotConfigManager(host_data_root=BOTS_WORKSPACE_ROOT) speech_service = WhisperSpeechService() app.state.docker_manager = docker_manager app.state.speech_service = speech_service BOT_ID_PATTERN = re.compile(r"^[A-Za-z0-9_]+$") runtime_assembly = assemble_app_runtime( app=app, logger=logger, bots_workspace_root=BOTS_WORKSPACE_ROOT, data_root=DATA_ROOT, docker_manager=docker_manager, config_manager=config_manager, speech_service=speech_service, bot_id_pattern=BOT_ID_PATTERN, ) app.include_router(build_system_runtime_router(system_service=runtime_assembly.system_service)) @app.middleware("http") async def bot_access_password_guard(request: Request, call_next): return await runtime_assembly.dashboard_auth_service.guard(request, call_next) @app.on_event("startup") async def on_startup(): await runtime_assembly.app_lifecycle_service.on_startup()