bffce65aca
3가지 검사 모드(한 페이지, 사이트 크롤링, 목록 업로드) 모두에서 접근성 표준을 선택할 수 있도록 추가. WCAG 2.0 A/AA, 2.1 AA, 2.2 AA와 KWCAG 2.1, 2.2를 지원하며, KWCAG 선택 시 axe-core 결과를 KWCAG 검사항목으로 자동 매핑. - KWCAG 2.2 (33항목) / 2.1 (24항목) ↔ WCAG 매핑 테이블 (kwcag_mapping.py) - AccessibilityChecker에 표준 파싱 및 KWCAG 변환 로직 추가 - 전체 API 파이프라인에 accessibility_standard 파라미터 전파 - 프론트엔드 3개 폼에 공용 표준 선택 드롭다운 추가 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
256 lines
8.3 KiB
Python
256 lines
8.3 KiB
Python
"""
|
|
Inspections router.
|
|
Handles inspection lifecycle: start, SSE stream, result, issues, history, trend.
|
|
|
|
IMPORTANT: Static paths (/batch, /trend) must be registered BEFORE
|
|
dynamic paths (/{id}) to avoid routing conflicts.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
from pydantic import HttpUrl, ValidationError
|
|
from sse_starlette.sse import EventSourceResponse
|
|
|
|
from app.core.database import get_db
|
|
from app.core.redis import get_redis, get_current_progress
|
|
from app.models.schemas import StartInspectionRequest, StartInspectionResponse
|
|
from app.services.inspection_service import InspectionService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _get_service() -> InspectionService:
|
|
"""Get InspectionService instance."""
|
|
db = get_db()
|
|
redis = get_redis()
|
|
return InspectionService(db=db, redis=redis)
|
|
|
|
|
|
# ============================================================
|
|
# POST /api/inspections -- Start inspection
|
|
# ============================================================
|
|
|
|
@router.post("/inspections", status_code=202)
|
|
async def start_inspection(request: StartInspectionRequest):
|
|
"""
|
|
Start a new web inspection.
|
|
Returns 202 Accepted with inspection_id immediately.
|
|
Inspection runs asynchronously in the background.
|
|
"""
|
|
url = str(request.url)
|
|
|
|
# Validate URL scheme
|
|
if not url.startswith(("http://", "https://")):
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail="유효한 URL을 입력해주세요 (http:// 또는 https://로 시작해야 합니다)",
|
|
)
|
|
|
|
service = _get_service()
|
|
|
|
try:
|
|
inspection_id = await service.start_inspection(
|
|
url,
|
|
accessibility_standard=request.accessibility_standard,
|
|
)
|
|
except httpx.HTTPStatusError as e:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"해당 URL에 접근할 수 없습니다 (HTTP {e.response.status_code})",
|
|
)
|
|
except httpx.TimeoutException:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="해당 URL에 접근할 수 없습니다 (응답 시간 초과)",
|
|
)
|
|
except httpx.RequestError as e:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="해당 URL에 접근할 수 없습니다",
|
|
)
|
|
except Exception as e:
|
|
logger.error("Failed to start inspection: %s", str(e))
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="해당 URL에 접근할 수 없습니다",
|
|
)
|
|
|
|
return StartInspectionResponse(
|
|
inspection_id=inspection_id,
|
|
status="running",
|
|
url=url,
|
|
stream_url=f"/api/inspections/{inspection_id}/stream",
|
|
)
|
|
|
|
|
|
# ============================================================
|
|
# GET /api/inspections -- List inspections (history)
|
|
# IMPORTANT: This MUST be before /{inspection_id} routes
|
|
# ============================================================
|
|
|
|
@router.get("/inspections")
|
|
async def list_inspections(
|
|
page: int = Query(default=1, ge=1),
|
|
limit: int = Query(default=20, ge=1, le=100),
|
|
url: Optional[str] = Query(default=None),
|
|
sort: str = Query(default="-created_at"),
|
|
):
|
|
"""Get paginated inspection history."""
|
|
service = _get_service()
|
|
result = await service.get_inspection_list(
|
|
page=page,
|
|
limit=limit,
|
|
url_filter=url,
|
|
sort=sort,
|
|
)
|
|
return result
|
|
|
|
|
|
# ============================================================
|
|
# GET /api/inspections/trend -- Trend data
|
|
# IMPORTANT: Must be before /{inspection_id} to avoid conflict
|
|
# ============================================================
|
|
|
|
@router.get("/inspections/trend")
|
|
async def get_trend(
|
|
url: str = Query(..., description="Target URL for trend data"),
|
|
limit: int = Query(default=10, ge=1, le=50),
|
|
):
|
|
"""Get trend data (score history) for a specific URL."""
|
|
service = _get_service()
|
|
result = await service.get_trend(url=url, limit=limit)
|
|
return result
|
|
|
|
|
|
# ============================================================
|
|
# GET /api/inspections/{inspection_id}/stream -- SSE stream
|
|
# ============================================================
|
|
|
|
@router.get("/inspections/{inspection_id}/stream")
|
|
async def stream_progress(inspection_id: str):
|
|
"""Stream inspection progress via Server-Sent Events."""
|
|
|
|
async def event_generator():
|
|
redis = get_redis()
|
|
pubsub = redis.pubsub()
|
|
channel = f"inspection:{inspection_id}:events"
|
|
|
|
await pubsub.subscribe(channel)
|
|
|
|
try:
|
|
# Send current state immediately (client may connect mid-progress)
|
|
current = await get_current_progress(inspection_id)
|
|
if current:
|
|
# Build initial progress event
|
|
categories = {}
|
|
for cat in ["html_css", "accessibility", "seo", "performance_security"]:
|
|
cat_progress = int(current.get(f"{cat}_progress", 0))
|
|
cat_step = current.get(f"{cat}_step", "")
|
|
cat_status = current.get(f"{cat}_status", "pending")
|
|
categories[cat] = {
|
|
"status": cat_status,
|
|
"progress": cat_progress,
|
|
"current_step": cat_step,
|
|
}
|
|
|
|
total = sum(c["progress"] for c in categories.values())
|
|
overall = round(total / 4)
|
|
|
|
yield {
|
|
"event": "progress",
|
|
"data": json.dumps({
|
|
"inspection_id": inspection_id,
|
|
"status": "running",
|
|
"overall_progress": overall,
|
|
"categories": categories,
|
|
}, ensure_ascii=False),
|
|
}
|
|
|
|
# Listen for Pub/Sub messages
|
|
async for message in pubsub.listen():
|
|
if message["type"] == "message":
|
|
event_data = json.loads(message["data"])
|
|
event_type = event_data.pop("event_type", "progress")
|
|
|
|
yield {
|
|
"event": event_type,
|
|
"data": json.dumps(event_data, ensure_ascii=False),
|
|
}
|
|
|
|
# End stream on complete or error
|
|
if event_type in ("complete", "error"):
|
|
break
|
|
|
|
except Exception as e:
|
|
logger.error("SSE stream error for %s: %s", inspection_id, str(e))
|
|
yield {
|
|
"event": "error",
|
|
"data": json.dumps({
|
|
"inspection_id": inspection_id,
|
|
"status": "error",
|
|
"message": "스트리밍 중 오류가 발생했습니다",
|
|
}, ensure_ascii=False),
|
|
}
|
|
finally:
|
|
await pubsub.unsubscribe(channel)
|
|
await pubsub.aclose()
|
|
|
|
return EventSourceResponse(
|
|
event_generator(),
|
|
media_type="text/event-stream",
|
|
)
|
|
|
|
|
|
# ============================================================
|
|
# GET /api/inspections/{inspection_id} -- Get result
|
|
# ============================================================
|
|
|
|
@router.get("/inspections/{inspection_id}")
|
|
async def get_inspection(inspection_id: str):
|
|
"""Get inspection result by ID."""
|
|
service = _get_service()
|
|
result = await service.get_inspection(inspection_id)
|
|
|
|
if result is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="검사 결과를 찾을 수 없습니다",
|
|
)
|
|
|
|
# Remove MongoDB _id field if present
|
|
result.pop("_id", None)
|
|
return result
|
|
|
|
|
|
# ============================================================
|
|
# GET /api/inspections/{inspection_id}/issues -- Get issues
|
|
# ============================================================
|
|
|
|
@router.get("/inspections/{inspection_id}/issues")
|
|
async def get_issues(
|
|
inspection_id: str,
|
|
category: Optional[str] = Query(default=None),
|
|
severity: Optional[str] = Query(default=None),
|
|
):
|
|
"""Get filtered issue list for an inspection."""
|
|
service = _get_service()
|
|
result = await service.get_issues(
|
|
inspection_id=inspection_id,
|
|
category=category,
|
|
severity=severity,
|
|
)
|
|
|
|
if result is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="검사 결과를 찾을 수 없습니다",
|
|
)
|
|
|
|
return result
|