feat: 접근성 검사 표준 선택 기능 — WCAG/KWCAG 버전별 선택 지원

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>
This commit is contained in:
jungwoo choi
2026-02-14 08:36:14 +09:00
parent 21259eb40a
commit bffce65aca
19 changed files with 857 additions and 59 deletions

View File

@ -1,8 +1,9 @@
"""
Accessibility (WCAG) Checker Engine (F-003).
Accessibility (WCAG/KWCAG) Checker Engine (F-003).
Uses Playwright + axe-core for comprehensive accessibility testing.
Falls back to BeautifulSoup-based checks if Playwright is unavailable.
Supports WCAG version selection (2.0/2.1/2.2) via rules/accessibility.yaml.
Supports KWCAG (Korean) standard selection (2.1/2.2) via kwcag_mapping module.
"""
import json
@ -14,6 +15,11 @@ from typing import Optional
from bs4 import BeautifulSoup
from app.engines.base import BaseChecker
from app.engines.kwcag_mapping import (
convert_wcag_issue_to_kwcag,
get_kwcag_axe_tags,
get_kwcag_label,
)
from app.models.schemas import CategoryResult, Issue
from app.rules import get_rules
@ -115,13 +121,55 @@ def _get_wcag_level_label(wcag_version: str) -> str:
return labels.get(wcag_version, "WCAG 2.1 Level AA")
class AccessibilityChecker(BaseChecker):
"""Accessibility (WCAG) checker engine with version selection."""
# Standard parameter to internal version mapping
# Maps API-facing standard values to (is_kwcag, internal_version)
_STANDARD_MAP: dict[str, tuple[bool, str]] = {
"wcag_2.0_a": (False, "2.0_A"),
"wcag_2.0_aa": (False, "2.0_AA"),
"wcag_2.1_aa": (False, "2.1_AA"),
"wcag_2.2_aa": (False, "2.2_AA"),
"kwcag_2.1": (True, "kwcag_2.1"),
"kwcag_2.2": (True, "kwcag_2.2"),
}
def __init__(self, progress_callback=None, wcag_version: str = "2.1_AA"):
def _parse_standard(standard: str) -> tuple[bool, str]:
"""
Parse the accessibility standard parameter.
Args:
standard: API-facing standard string, e.g. "wcag_2.1_aa" or "kwcag_2.2"
Returns:
(is_kwcag, version_key) where version_key is the internal key
for axe tag selection and label generation.
"""
return _STANDARD_MAP.get(standard, (False, "2.1_AA"))
class AccessibilityChecker(BaseChecker):
"""Accessibility (WCAG/KWCAG) checker engine with standard selection."""
def __init__(
self,
progress_callback=None,
wcag_version: str = "2.1_AA",
standard: str = "wcag_2.1_aa",
):
super().__init__(progress_callback)
self.wcag_version = wcag_version
self.axe_tags = _get_axe_tags_for_version(wcag_version)
# Parse the standard parameter to determine mode
self.is_kwcag, self._version_key = _parse_standard(standard)
self.standard = standard
if self.is_kwcag:
# KWCAG mode: use KWCAG-specific axe tags
self.wcag_version = wcag_version # Keep for internal compatibility
self.axe_tags = get_kwcag_axe_tags(self._version_key)
else:
# WCAG mode: use the internal version key for axe tags
self.wcag_version = self._version_key
self.axe_tags = _get_axe_tags_for_version(self._version_key)
@property
def category_name(self) -> str:
@ -172,7 +220,12 @@ class AccessibilityChecker(BaseChecker):
}
""")
await self.update_progress(60, f"접근성 검사 실행 중 ({_get_wcag_level_label(self.wcag_version)})...")
level_label = (
get_kwcag_label(self._version_key)
if self.is_kwcag
else _get_wcag_level_label(self.wcag_version)
)
await self.update_progress(60, f"접근성 검사 실행 중 ({level_label})...")
# Build axe-core tag list including best-practice
axe_tag_list = self.axe_tags + ["best-practice"]
@ -203,11 +256,17 @@ class AccessibilityChecker(BaseChecker):
await browser.close()
await self.update_progress(100, "완료")
wcag_level = (
get_kwcag_label(self._version_key)
if self.is_kwcag
else _get_wcag_level_label(self.wcag_version)
)
return self._build_result(
category="accessibility",
score=score,
issues=issues,
wcag_level=_get_wcag_level_label(self.wcag_version),
wcag_level=wcag_level,
)
async def _check_with_beautifulsoup(self, url: str, html_content: str) -> CategoryResult:
@ -239,11 +298,16 @@ class AccessibilityChecker(BaseChecker):
score = self._calculate_score_by_deduction(issues)
await self.update_progress(100, "완료")
wcag_level = (
get_kwcag_label(self._version_key)
if self.is_kwcag
else _get_wcag_level_label(self.wcag_version)
)
return self._build_result(
category="accessibility",
score=score,
issues=issues,
wcag_level=_get_wcag_level_label(self.wcag_version),
wcag_level=wcag_level,
)
def _parse_axe_results(self, axe_results: dict) -> list[Issue]:
@ -287,14 +351,22 @@ class AccessibilityChecker(BaseChecker):
node_count = len(nodes)
count_info = f" ({node_count}개 요소)" if node_count > 1 else ""
issues.append(self._create_issue(
issue = self._create_issue(
code=code,
severity=severity,
message=f"{korean_msg}{detail}{count_info}",
element=element,
suggestion=violation.get("helpUrl", "해당 WCAG 기준을 확인하고 수정하세요"),
wcag_criterion=wcag,
))
)
# Convert to KWCAG criterion if in KWCAG mode
if self.is_kwcag:
issue_dict = issue.model_dump()
issue_dict = convert_wcag_issue_to_kwcag(issue_dict, self._version_key)
issue = Issue(**issue_dict)
issues.append(issue)
return issues

View File

@ -0,0 +1,392 @@
"""
KWCAG (Korean Web Content Accessibility Guidelines) mapping module.
Provides bidirectional mapping between KWCAG 2.1/2.2 and WCAG criteria,
enabling axe-core results to be re-interpreted under Korean accessibility standards.
KWCAG 2.2 contains 33 inspection items across 4 principles.
KWCAG 2.1 contains 24 inspection items (a subset of 2.2).
KWCAG 2.2 aligns with WCAG 2.1 Level AA and includes additional items.
"""
import logging
from typing import Optional
logger = logging.getLogger(__name__)
# =============================================================================
# KWCAG 2.2 Full Mapping (33 items)
# =============================================================================
# Key: KWCAG criterion number
# Value: {
# "wcag_criteria": [list of corresponding WCAG criterion numbers],
# "name": Korean name of the KWCAG item,
# "principle": KWCAG principle name,
# "kwcag_21": True if this item also exists in KWCAG 2.1,
# }
KWCAG_TO_WCAG: dict[str, dict] = {
# ========================================
# 원칙 1. 인식의 용이성
# ========================================
"1.1.1": {
"wcag_criteria": ["1.1.1"],
"name": "적절한 대체 텍스트",
"principle": "인식의 용이성",
"kwcag_21": True,
},
"1.2.1": {
"wcag_criteria": ["1.2.2", "1.2.4"],
"name": "자막 제공",
"principle": "인식의 용이성",
"kwcag_21": True,
},
"1.3.1": {
"wcag_criteria": ["1.4.1"],
"name": "색에 무관한 콘텐츠 인식",
"principle": "인식의 용이성",
"kwcag_21": True,
},
"1.4.1": {
"wcag_criteria": ["1.3.3"],
"name": "명확한 지시 사항",
"principle": "인식의 용이성",
"kwcag_21": True,
},
"1.5.1": {
"wcag_criteria": ["1.4.3"],
"name": "텍스트 콘텐츠의 명도 대비",
"principle": "인식의 용이성",
"kwcag_21": True,
},
"1.6.1": {
"wcag_criteria": ["1.4.2"],
"name": "자동 재생 금지",
"principle": "인식의 용이성",
"kwcag_21": True,
},
"1.7.1": {
"wcag_criteria": ["1.3.1"],
"name": "콘텐츠 간의 구분",
"principle": "인식의 용이성",
"kwcag_21": True,
},
# ========================================
# 원칙 2. 운용의 용이성
# ========================================
"2.1.1": {
"wcag_criteria": ["2.1.1", "2.1.2"],
"name": "키보드 사용 보장",
"principle": "운용의 용이성",
"kwcag_21": True,
},
"2.2.1": {
"wcag_criteria": ["2.4.3", "2.4.7"],
"name": "초점 이동과 표시",
"principle": "운용의 용이성",
"kwcag_21": True,
},
"2.3.1": {
"wcag_criteria": ["2.5.1"],
"name": "조작 가능",
"principle": "운용의 용이성",
"kwcag_21": False,
},
"2.4.1": {
"wcag_criteria": ["2.2.1"],
"name": "응답시간 조절",
"principle": "운용의 용이성",
"kwcag_21": True,
},
"2.5.1": {
"wcag_criteria": ["2.2.2"],
"name": "정지 기능 제공",
"principle": "운용의 용이성",
"kwcag_21": True,
},
"2.6.1": {
"wcag_criteria": ["2.3.1"],
"name": "깜빡임과 번쩍임 사용 제한",
"principle": "운용의 용이성",
"kwcag_21": True,
},
"2.7.1": {
"wcag_criteria": ["2.4.1"],
"name": "반복 영역 건너뛰기",
"principle": "운용의 용이성",
"kwcag_21": True,
},
"2.8.1": {
"wcag_criteria": ["2.4.2", "2.4.5", "2.4.6", "2.4.8"],
"name": "쉬운 내비게이션",
"principle": "운용의 용이성",
"kwcag_21": True,
},
# ========================================
# 원칙 3. 이해의 용이성
# ========================================
"3.1.1": {
"wcag_criteria": ["3.1.1"],
"name": "기본 언어 표시",
"principle": "이해의 용이성",
"kwcag_21": True,
},
"3.2.1": {
"wcag_criteria": ["3.2.1", "3.2.2"],
"name": "사용자 요구에 따른 실행",
"principle": "이해의 용이성",
"kwcag_21": True,
},
"3.3.1": {
"wcag_criteria": ["1.3.2"],
"name": "콘텐츠의 선형구조",
"principle": "이해의 용이성",
"kwcag_21": True,
},
"3.4.1": {
"wcag_criteria": ["1.3.1"],
"name": "표의 구성",
"principle": "이해의 용이성",
"kwcag_21": True,
},
"3.5.1": {
"wcag_criteria": ["1.3.1", "3.3.2"],
"name": "레이블 제공",
"principle": "이해의 용이성",
"kwcag_21": True,
},
"3.6.1": {
"wcag_criteria": ["3.3.1", "3.3.3"],
"name": "오류 정정",
"principle": "이해의 용이성",
"kwcag_21": True,
},
# ========================================
# 원칙 4. 견고성
# ========================================
"4.1.1": {
"wcag_criteria": ["4.1.1"],
"name": "마크업 오류 방지",
"principle": "견고성",
"kwcag_21": True,
},
"4.2.1": {
"wcag_criteria": ["4.1.2"],
"name": "웹 애플리케이션 접근성 준수",
"principle": "견고성",
"kwcag_21": True,
},
# ========================================
# KWCAG 2.2 추가 항목 (KWCAG 2.1에 없는 항목)
# ========================================
"1.1.2": {
"wcag_criteria": ["1.1.1"],
"name": "대체 콘텐츠의 동등성",
"principle": "인식의 용이성",
"kwcag_21": False,
},
"1.2.2": {
"wcag_criteria": ["1.2.1", "1.2.3"],
"name": "음성 및 영상 대체 수단",
"principle": "인식의 용이성",
"kwcag_21": False,
},
"1.5.2": {
"wcag_criteria": ["1.4.11"],
"name": "비텍스트 콘텐츠의 명도 대비",
"principle": "인식의 용이성",
"kwcag_21": False,
},
"2.1.2": {
"wcag_criteria": ["2.1.4"],
"name": "키보드 단축키",
"principle": "운용의 용이성",
"kwcag_21": False,
},
"2.2.2": {
"wcag_criteria": ["2.4.11"],
"name": "초점 감춤 방지",
"principle": "운용의 용이성",
"kwcag_21": False,
},
"2.3.2": {
"wcag_criteria": ["2.5.1", "2.5.2"],
"name": "포인터 입력의 취소",
"principle": "운용의 용이성",
"kwcag_21": False,
},
"2.3.3": {
"wcag_criteria": ["2.5.8"],
"name": "포인터 대상 크기",
"principle": "운용의 용이성",
"kwcag_21": False,
},
"2.8.2": {
"wcag_criteria": ["2.4.4"],
"name": "링크 텍스트",
"principle": "운용의 용이성",
"kwcag_21": False,
},
"3.2.2": {
"wcag_criteria": ["3.2.6"],
"name": "일관된 도움 제공",
"principle": "이해의 용이성",
"kwcag_21": False,
},
"3.6.2": {
"wcag_criteria": ["3.3.8"],
"name": "접근 가능한 인증",
"principle": "이해의 용이성",
"kwcag_21": False,
},
"3.6.3": {
"wcag_criteria": ["3.3.7"],
"name": "중복 입력 방지",
"principle": "이해의 용이성",
"kwcag_21": False,
},
}
# =============================================================================
# Reverse mapping: WCAG criterion -> KWCAG criterion(s)
# Built at module load time for O(1) lookup during result conversion.
# =============================================================================
def _build_wcag_to_kwcag_map() -> dict[str, list[str]]:
"""Build reverse mapping from WCAG criterion to KWCAG criterion(s)."""
reverse_map: dict[str, list[str]] = {}
for kwcag_id, info in KWCAG_TO_WCAG.items():
for wcag_id in info["wcag_criteria"]:
if wcag_id not in reverse_map:
reverse_map[wcag_id] = []
reverse_map[wcag_id].append(kwcag_id)
return reverse_map
WCAG_TO_KWCAG: dict[str, list[str]] = _build_wcag_to_kwcag_map()
# =============================================================================
# axe-core tag sets for KWCAG versions
# =============================================================================
# KWCAG 2.2 -> WCAG 2.1 AA level + selected 2.2 AA rules
# KWCAG 2.1 -> WCAG 2.0 AA level + selected 2.1 AA rules
_KWCAG_AXE_TAGS: dict[str, list[str]] = {
"kwcag_2.1": ["wcag2a", "wcag2aa", "wcag21a", "wcag21aa"],
"kwcag_2.2": ["wcag2a", "wcag2aa", "wcag21a", "wcag21aa", "wcag22aa"],
}
def get_kwcag_axe_tags(version: str) -> list[str]:
"""
Get axe-core tag list for a KWCAG version.
Args:
version: KWCAG version string, e.g. "kwcag_2.1" or "kwcag_2.2"
Returns:
List of axe-core tags to run.
"""
return _KWCAG_AXE_TAGS.get(version, _KWCAG_AXE_TAGS["kwcag_2.2"])
# =============================================================================
# Utility functions for result conversion
# =============================================================================
def convert_wcag_issue_to_kwcag(
issue_dict: dict,
kwcag_version: str = "kwcag_2.2",
) -> dict:
"""
Convert a single issue's wcag_criterion field from WCAG to KWCAG numbering.
If the WCAG criterion maps to a KWCAG item, the issue is updated with:
- kwcag_criterion: The KWCAG criterion number (e.g. "1.1.1")
- kwcag_name: The Korean name of the KWCAG item
- kwcag_principle: The KWCAG principle name
The original wcag_criterion field is preserved.
If no mapping exists, the issue is returned unchanged.
Args:
issue_dict: Dictionary representing a single inspection issue.
kwcag_version: "kwcag_2.1" or "kwcag_2.2"
Returns:
The issue_dict with KWCAG fields added (mutated in place and returned).
"""
wcag_criterion = issue_dict.get("wcag_criterion")
if not wcag_criterion:
return issue_dict
kwcag_ids = WCAG_TO_KWCAG.get(wcag_criterion, [])
if not kwcag_ids:
return issue_dict
# Filter to items that exist in the requested KWCAG version
is_21 = kwcag_version == "kwcag_2.1"
valid_ids = []
for kid in kwcag_ids:
info = KWCAG_TO_WCAG.get(kid)
if info is None:
continue
if is_21 and not info.get("kwcag_21", False):
continue
valid_ids.append(kid)
if not valid_ids:
return issue_dict
# Use the first matching KWCAG criterion
primary_kwcag_id = valid_ids[0]
info = KWCAG_TO_WCAG[primary_kwcag_id]
issue_dict["kwcag_criterion"] = primary_kwcag_id
issue_dict["kwcag_name"] = info["name"]
issue_dict["kwcag_principle"] = info["principle"]
# If multiple KWCAG items match, include all
if len(valid_ids) > 1:
issue_dict["kwcag_criteria_all"] = valid_ids
return issue_dict
def get_kwcag_label(version: str) -> str:
"""
Get human-readable label for a KWCAG version.
Args:
version: "kwcag_2.1" or "kwcag_2.2"
Returns:
Korean label string, e.g. "KWCAG 2.2 (한국형 웹 콘텐츠 접근성 지침 2.2)"
"""
labels = {
"kwcag_2.1": "KWCAG 2.1 (한국형 웹 콘텐츠 접근성 지침 2.1)",
"kwcag_2.2": "KWCAG 2.2 (한국형 웹 콘텐츠 접근성 지침 2.2)",
}
return labels.get(version, labels["kwcag_2.2"])
def get_kwcag_item_count(version: str) -> int:
"""
Get the total number of inspection items for a KWCAG version.
Args:
version: "kwcag_2.1" or "kwcag_2.2"
Returns:
Number of items: 24 for KWCAG 2.1, 33 for KWCAG 2.2
"""
if version == "kwcag_2.1":
return sum(1 for info in KWCAG_TO_WCAG.values() if info.get("kwcag_21", False))
return len(KWCAG_TO_WCAG)

View File

@ -39,6 +39,7 @@ class BatchPage(BaseModel):
class BatchInspectionConfig(BaseModel):
"""배치 검사 설정."""
concurrency: int = 4
accessibility_standard: str = "wcag_2.1_aa"
# --- Response Models ---

View File

@ -34,6 +34,10 @@ class CategoryName(str, Enum):
class StartInspectionRequest(BaseModel):
url: HttpUrl
accessibility_standard: str = Field(
default="wcag_2.1_aa",
description="접근성 검사 표준 (wcag_2.0_a, wcag_2.0_aa, wcag_2.1_aa, wcag_2.2_aa, kwcag_2.1, kwcag_2.2)",
)
# --- Core Data Models ---
@ -47,6 +51,11 @@ class Issue(BaseModel):
line: Optional[int] = None
suggestion: str
wcag_criterion: Optional[str] = None
# KWCAG mapping fields (populated when standard is kwcag_2.1 or kwcag_2.2)
kwcag_criterion: Optional[str] = None
kwcag_name: Optional[str] = None
kwcag_principle: Optional[str] = None
kwcag_criteria_all: Optional[list[str]] = None
class CategoryResult(BaseModel):
@ -93,6 +102,7 @@ class InspectionResult(BaseModel):
grade: str
categories: dict[str, CategoryResult]
summary: IssueSummary
accessibility_standard: Optional[str] = None
class InspectionResultResponse(BaseModel):
@ -107,6 +117,7 @@ class InspectionResultResponse(BaseModel):
grade: str
categories: dict[str, CategoryResult]
summary: IssueSummary
accessibility_standard: Optional[str] = None
class IssueListResponse(BaseModel):

View File

@ -7,6 +7,9 @@ from typing import Optional
from datetime import datetime
from enum import Enum
# Default accessibility standard
DEFAULT_ACCESSIBILITY_STANDARD = "wcag_2.1_aa"
# --- Enums ---
@ -31,6 +34,10 @@ class StartSiteInspectionRequest(BaseModel):
max_pages: int = Field(default=20, ge=0, le=500, description="최대 크롤링 페이지 수 (0=무제한)")
max_depth: int = Field(default=2, ge=1, le=3, description="최대 크롤링 깊이")
concurrency: int = Field(default=4, ge=1, le=8, description="동시 검사 수")
accessibility_standard: str = Field(
default=DEFAULT_ACCESSIBILITY_STANDARD,
description="접근성 검사 표준 (wcag_2.0_a, wcag_2.0_aa, wcag_2.1_aa, wcag_2.2_aa, kwcag_2.1, kwcag_2.2)",
)
class InspectPageRequest(BaseModel):
@ -69,6 +76,7 @@ class SiteInspectionConfig(BaseModel):
max_pages: int = 20
max_depth: int = 2
concurrency: int = 4
accessibility_standard: str = DEFAULT_ACCESSIBILITY_STANDARD
# --- Response Models ---

View File

@ -73,6 +73,7 @@ async def start_batch_inspection(
file: UploadFile = File(...),
name: str = Form(...),
concurrency: int = Form(default=4),
accessibility_standard: str = Form(default="wcag_2.1_aa"),
):
"""
Start a new batch inspection from an uploaded URL file.
@ -154,6 +155,7 @@ async def start_batch_inspection(
name=name,
urls=urls,
concurrency=concurrency,
accessibility_standard=accessibility_standard,
)
except Exception as e:
logger.error("Failed to start batch inspection: %s", str(e))

View File

@ -55,7 +55,10 @@ async def start_inspection(request: StartInspectionRequest):
service = _get_service()
try:
inspection_id = await service.start_inspection(url)
inspection_id = await service.start_inspection(
url,
accessibility_standard=request.accessibility_standard,
)
except httpx.HTTPStatusError as e:
raise HTTPException(
status_code=400,

View File

@ -67,6 +67,7 @@ async def start_site_inspection(request: StartSiteInspectionRequest):
max_pages=request.max_pages,
max_depth=request.max_depth,
concurrency=request.concurrency,
accessibility_standard=request.accessibility_standard,
)
except httpx.HTTPStatusError as e:
raise HTTPException(

View File

@ -46,6 +46,7 @@ class BatchInspectionService:
name: str,
urls: list[str],
concurrency: int = 4,
accessibility_standard: str = "wcag_2.1_aa",
) -> str:
"""
Start a batch inspection.
@ -92,6 +93,7 @@ class BatchInspectionService:
"completed_at": None,
"config": {
"concurrency": concurrency,
"accessibility_standard": accessibility_standard,
},
"source_urls": urls,
"discovered_pages": discovered_pages,
@ -100,13 +102,13 @@ class BatchInspectionService:
await self.db.batch_inspections.insert_one(doc)
logger.info(
"Batch inspection started: id=%s, name=%s, total_urls=%d, concurrency=%d",
batch_inspection_id, name, len(urls), concurrency,
"Batch inspection started: id=%s, name=%s, total_urls=%d, concurrency=%d, standard=%s",
batch_inspection_id, name, len(urls), concurrency, accessibility_standard,
)
# Launch background task
asyncio.create_task(
self._inspect_all(batch_inspection_id, urls, concurrency)
self._inspect_all(batch_inspection_id, urls, concurrency, accessibility_standard)
)
return batch_inspection_id
@ -205,6 +207,7 @@ class BatchInspectionService:
batch_inspection_id: str,
urls: list[str],
concurrency: int = 4,
accessibility_standard: str = "wcag_2.1_aa",
) -> None:
"""
Background task that inspects all URLs in parallel.
@ -225,6 +228,7 @@ class BatchInspectionService:
page_url=url,
page_index=idx,
total_pages=len(urls),
accessibility_standard=accessibility_standard,
)
for idx, url in enumerate(urls)
]
@ -287,6 +291,7 @@ class BatchInspectionService:
page_url: str,
page_index: int,
total_pages: int,
accessibility_standard: str = "wcag_2.1_aa",
) -> None:
"""Inspect a single page with semaphore-controlled concurrency."""
async with semaphore:
@ -295,6 +300,7 @@ class BatchInspectionService:
page_url=page_url,
page_index=page_index,
total_pages=total_pages,
accessibility_standard=accessibility_standard,
)
async def _inspect_single_page(
@ -303,6 +309,7 @@ class BatchInspectionService:
page_url: str,
page_index: int,
total_pages: int,
accessibility_standard: str = "wcag_2.1_aa",
) -> None:
"""Run inspection for a single page in the batch."""
inspection_id = str(uuid.uuid4())
@ -347,6 +354,7 @@ class BatchInspectionService:
url=page_url,
inspection_id=inspection_id,
progress_callback=page_progress_callback,
accessibility_standard=accessibility_standard,
)
overall_score = result.get("overall_score", 0)

View File

@ -49,7 +49,11 @@ class InspectionService:
self.db = db
self.redis = redis
async def start_inspection(self, url: str) -> str:
async def start_inspection(
self,
url: str,
accessibility_standard: str = "wcag_2.1_aa",
) -> str:
"""
Start an inspection and return the inspection_id.
1. Validate URL accessibility (timeout 10s)
@ -70,7 +74,7 @@ class InspectionService:
# 4. Run inspection as background task
asyncio.create_task(
self._run_inspection(inspection_id, url, response)
self._run_inspection(inspection_id, url, response, accessibility_standard)
)
return inspection_id
@ -80,6 +84,7 @@ class InspectionService:
url: str,
inspection_id: Optional[str] = None,
progress_callback: Optional[object] = None,
accessibility_standard: str = "wcag_2.1_aa",
) -> tuple[str, dict]:
"""
Run a full inspection synchronously (inline) and return the result.
@ -93,6 +98,7 @@ class InspectionService:
inspection_id: Optional pre-generated ID. If None, a new UUID is generated.
progress_callback: Optional async callback(category, progress, current_step).
If None, progress is not reported.
accessibility_standard: Accessibility standard to use for inspection.
Returns:
(inspection_id, result_dict) where result_dict is the MongoDB document.
@ -120,7 +126,10 @@ class InspectionService:
# Create 4 checker engines
checkers = [
HtmlCssChecker(progress_callback=progress_callback),
AccessibilityChecker(progress_callback=progress_callback),
AccessibilityChecker(
progress_callback=progress_callback,
standard=accessibility_standard,
),
SeoChecker(progress_callback=progress_callback),
PerformanceSecurityChecker(progress_callback=progress_callback),
]
@ -190,6 +199,7 @@ class InspectionService:
grade=grade,
categories=categories,
summary=summary,
accessibility_standard=accessibility_standard,
)
# Store in MongoDB
@ -203,14 +213,18 @@ class InspectionService:
await cache_result(inspection_id, doc)
logger.info(
"Inspection %s completed (inline): score=%d, duration=%.1fs",
inspection_id, overall_score, duration,
"Inspection %s completed (inline): score=%d, duration=%.1fs, standard=%s",
inspection_id, overall_score, duration, accessibility_standard,
)
return inspection_id, doc
async def _run_inspection(
self, inspection_id: str, url: str, response: httpx.Response
self,
inspection_id: str,
url: str,
response: httpx.Response,
accessibility_standard: str = "wcag_2.1_aa",
) -> None:
"""
Execute 4 category checks in parallel and store results.
@ -234,7 +248,10 @@ class InspectionService:
# Create 4 checker engines
checkers = [
HtmlCssChecker(progress_callback=progress_callback),
AccessibilityChecker(progress_callback=progress_callback),
AccessibilityChecker(
progress_callback=progress_callback,
standard=accessibility_standard,
),
SeoChecker(progress_callback=progress_callback),
PerformanceSecurityChecker(progress_callback=progress_callback),
]
@ -322,6 +339,7 @@ class InspectionService:
grade=grade,
categories=categories,
summary=summary,
accessibility_standard=accessibility_standard,
)
# Store in MongoDB
@ -347,8 +365,8 @@ class InspectionService:
})
logger.info(
"Inspection %s completed: score=%d, duration=%.1fs",
inspection_id, overall_score, duration,
"Inspection %s completed: score=%d, duration=%.1fs, standard=%s",
inspection_id, overall_score, duration, accessibility_standard,
)
except Exception as e:

View File

@ -50,6 +50,7 @@ class SiteInspectionService:
max_pages: int = 20,
max_depth: int = 2,
concurrency: int = 4,
accessibility_standard: str = "wcag_2.1_aa",
) -> str:
"""
Start a site-wide inspection.
@ -84,6 +85,7 @@ class SiteInspectionService:
"max_pages": max_pages,
"max_depth": max_depth,
"concurrency": concurrency,
"accessibility_standard": accessibility_standard,
},
"discovered_pages": [],
"aggregate_scores": None,
@ -91,13 +93,16 @@ class SiteInspectionService:
await self.db.site_inspections.insert_one(doc)
logger.info(
"Site inspection started: id=%s, url=%s, max_pages=%d, max_depth=%d, concurrency=%d",
site_inspection_id, url, max_pages, max_depth, concurrency,
"Site inspection started: id=%s, url=%s, max_pages=%d, max_depth=%d, concurrency=%d, standard=%s",
site_inspection_id, url, max_pages, max_depth, concurrency, accessibility_standard,
)
# Launch background task
asyncio.create_task(
self._crawl_and_inspect(site_inspection_id, url, max_pages, max_depth, concurrency)
self._crawl_and_inspect(
site_inspection_id, url, max_pages, max_depth, concurrency,
accessibility_standard=accessibility_standard,
)
)
return site_inspection_id
@ -272,6 +277,7 @@ class SiteInspectionService:
max_pages: int,
max_depth: int,
concurrency: int = 4,
accessibility_standard: str = "wcag_2.1_aa",
) -> None:
"""
Background task that runs in two phases:
@ -366,6 +372,7 @@ class SiteInspectionService:
page_url=page["url"],
page_index=idx,
total_pages=len(discovered_pages),
accessibility_standard=accessibility_standard,
)
for idx, page in enumerate(discovered_pages)
]
@ -428,6 +435,7 @@ class SiteInspectionService:
page_url: str,
page_index: int,
total_pages: int,
accessibility_standard: str = "wcag_2.1_aa",
) -> None:
"""Inspect a single page with semaphore-controlled concurrency."""
async with semaphore:
@ -436,6 +444,7 @@ class SiteInspectionService:
page_url=page_url,
page_index=page_index,
total_pages=total_pages,
accessibility_standard=accessibility_standard,
)
async def _inspect_single_page(
@ -444,6 +453,7 @@ class SiteInspectionService:
page_url: str,
page_index: int,
total_pages: int,
accessibility_standard: str = "wcag_2.1_aa",
) -> None:
"""Run inspection for a single discovered page."""
inspection_id = str(uuid.uuid4())
@ -480,6 +490,7 @@ class SiteInspectionService:
url=page_url,
inspection_id=inspection_id,
progress_callback=page_progress_callback,
accessibility_standard=accessibility_standard,
)
overall_score = result.get("overall_score", 0)