""" API endpoints для проверки синтаксиса ролей (lint) Автор: Сергей Антропов Сайт: https://devops.org.ru """ from fastapi import APIRouter, Request, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse from fastapi.templating import Jinja2Templates from pathlib import Path from typing import Optional from app.core.config import settings from app.services.lint_service import LintService router = APIRouter() templates_path = Path(__file__).parent.parent.parent.parent / "templates" templates = Jinja2Templates(directory=str(templates_path)) lint_service = LintService() @router.get("/lint", response_class=HTMLResponse) async def lint_page(request: Request): """Страница проверки синтаксиса ролей""" # Получение списка ролей roles_dir = settings.PROJECT_ROOT / "roles" roles = [] if roles_dir.exists(): for role_dir in roles_dir.iterdir(): if role_dir.is_dir() and role_dir.name != "deploy.yml": roles.append({ "name": role_dir.name, "path": str(role_dir) }) return templates.TemplateResponse( "pages/lint/index.html", { "request": request, "roles": sorted(roles, key=lambda x: x["name"]) } ) @router.get("/roles/{role_name}/lint", response_class=HTMLResponse) async def lint_role_page(request: Request, role_name: str): """Страница проверки синтаксиса конкретной роли""" # Проверка существования роли roles_dir = settings.PROJECT_ROOT / "roles" / role_name if not roles_dir.exists(): from fastapi import HTTPException raise HTTPException(status_code=404, detail=f"Роль '{role_name}' не найдена") return templates.TemplateResponse( "pages/lint/role.html", { "request": request, "role_name": role_name } ) @router.websocket("/ws/lint/{role_name}") async def lint_websocket(websocket: WebSocket, role_name: str): """WebSocket для live логов линтинга""" await websocket.accept() try: # Проверка существования роли (если указана) if role_name != "all": roles_dir = settings.PROJECT_ROOT / "roles" / role_name if not roles_dir.exists(): await websocket.send_json({ "type": "error", "data": f"Роль '{role_name}' не найдена" }) await websocket.close() return # Отправка начального сообщения if role_name == "all": await websocket.send_json({ "type": "info", "data": "🔍 Запуск проверки синтаксиса всех ролей..." }) else: await websocket.send_json({ "type": "info", "data": f"🔍 Запуск проверки синтаксиса роли '{role_name}'..." }) # Запуск линтинга role_name_param = None if role_name == "all" else role_name async for line in lint_service.lint_role( role_name=role_name_param, stream=True ): # Очистка строки от лишних символов line = line.rstrip() if not line: continue # Определение типа лога log_type = lint_service.detect_log_level(line) await websocket.send_json({ "type": "log", "level": log_type, "data": line }) # Завершение await websocket.send_json({ "type": "complete", "status": "success", "data": "✅ Линтинг завершен" }) except WebSocketDisconnect: pass except Exception as e: import traceback error_msg = f"❌ Ошибка: {str(e)}\n{traceback.format_exc()}" await websocket.send_json({ "type": "error", "data": error_msg }) finally: try: await websocket.close() except: pass