Feat(#22): AI 분석 결과 이벤트 및 Worker 장애 처리 구현

.dockerignore

@@ -0,0 +1,17 @@
+.git
+.github
+.vscode
+
+.venv
+__pycache__
+.pytest_cache
+*.py[cod]
+
+.env
+.env.*
+
+tests
+requirements-dev.txt
+
+*.log
+.DS_Store

.github/workflows/ai-ci.yml

@@ -0,0 +1,54 @@
+name: AI CI
+
+on:
+  pull_request:
+  push:
+    branches:
+      - develop
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    name: Python tests
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+          cache: pip
+          cache-dependency-path: |
+            requirements.txt
+            requirements-dev.txt
+
+      - name: Install dependencies
+        run: pip install -r requirements-dev.txt
+
+      - name: Check dependency compatibility
+        run: pip check
+
+      - name: Run tests
+        run: python -m pytest -q
+
+  docker-build:
+    name: Docker build
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+
+      - name: Build Docker image
+        run: docker build -t safefam-ai:test .

Dockerfile

@@ -1,17 +1,26 @@
 FROM python:3.11-slim
 
-# 작업 디렉토리 설정
 WORKDIR /workspace
 
-# 필수 패키지 설치를 위한 레이어 캐싱
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+
 COPY requirements.txt .
+
 RUN pip install --no-cache-dir -r requirements.txt
 
-# 소스코드 전체 복사
-COPY . .
+COPY app app
+
+RUN useradd --create-home --shell /usr/sbin/nologin safefam
+
+USER safefam
 
-# FastAPI 기본 포트 개방
 EXPOSE 8000
 
-# 로컬 개발 및 컨테이너 구동을 위한 기본 명령어
-CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
+HEALTHCHECK \
+    --interval=30s \
+    --timeout=5s \
+    --retries=3 \
+    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/', timeout=3)"
+
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

app/analysis/execution.py

@@ -0,0 +1,84 @@
+from dataclasses import dataclass
+from enum import Enum
+
+from app.analysis.schemas import SmishingAnalysisResponse
+
+
+class AnalysisExecutionStatus(str, Enum):
+    """분석 실행 상태."""
+
+    COMPLETED = "COMPLETED"
+    PARTIAL = "PARTIAL"
+    FAILED = "FAILED"
+
+
+@dataclass(frozen=True)
+class AnalysisExecution:
+    """분석 결과와 실패 트랙 정보를 담는 불변 데이터."""
+
+    status: AnalysisExecutionStatus
+    result: SmishingAnalysisResponse
+    failed_tracks: tuple[str, ...]
+
+
+def classify_execution(
+    result: SmishingAnalysisResponse,
+) -> AnalysisExecution:
+    """AI 분석 응답을 실행 상태와 실패 트랙 목록으로 분류한다."""
+    if result.status == "ERROR":
+        return AnalysisExecution(
+            status=AnalysisExecutionStatus.FAILED,
+            result=result,
+            failed_tracks=("PIPELINE",),
+        )
+
+    failed_tracks: list[str] = []
+
+    text_analysis = result.text_analysis or {}
+    text_result = text_analysis.get("result") or {}
+    stage1_result = text_analysis.get("stage1_naive_bayes") or {}
+
+    text_failed = (
+        text_result.get("grade") == "UNKNOWN"
+        or text_result.get("error_message")
+    )
+    stage1_failed = (
+        stage1_result.get("grade") == "UNKNOWN"
+        or stage1_result.get("error_message")
+    )
+
+    if text_failed:
+        if stage1_result and not stage1_failed:
+            failed_tracks.append("TEXT:GEMINI")
+        else:
+            failed_tracks.append("TEXT")
+    elif stage1_failed:
+        failed_tracks.append("TEXT:NAIVE_BAYES")
+
+    url_analysis = result.url_analysis or {}
+    failed_providers = url_analysis.get("failed_providers") or []
+    url_available = url_analysis.get("available", True)
+
+    if result.url_analysis is not None and not url_available:
+        failed_tracks.append("URL")
+    else:
+        failed_tracks.extend(
+            f"URL:{provider}"
+            for provider in failed_providers
+        )
+
+    rule_analysis = result.rule_analysis or {}
+    if rule_analysis.get("error_message"):
+        failed_tracks.append("RULES")
+
+    status = (
+        AnalysisExecutionStatus.PARTIAL
+        if failed_tracks
+        else AnalysisExecutionStatus.COMPLETED
+    )
+
+    return AnalysisExecution(
+        status=status,
+        result=result,
+        failed_tracks=tuple(failed_tracks),
+    )

app/analysis/schemas.py

@@ -39,9 +39,9 @@ class RiskGrade(str, Enum):
 class ContributionBreakdown(BaseModel):
     # URL이 있으면 LLM 50% / 규칙 20%, URL이 없으면 하이브리드 URL 트랙(30%)이 LLM/규칙으로
     # 재배분되어 LLM 65% / 규칙 35%가 되므로 상한이 두 시나리오 중 더 큰 쪽 기준으로 설정됨
-    llm: int = Field(..., description="LLM 문맥 분석 기여 점수 (URL 있음: 0~50, URL 없음: 0~65)", ge=0, le=65)
-    hybrid_url: int = Field(..., description="하이브리드 URL 보안 엔진 기여 점수 (URL 있을 때만 0~30, 없으면 0)", ge=0, le=30)
-    rules: int = Field(..., description="로컬 가드 규칙 기반 기여 점수 (URL 있음: 0~20, URL 없음: 0~35)", ge=0, le=35)
+    llm: int = Field(ge=0, le=100)
+    hybrid_url: int = Field(ge=0, le=100)
+    rules: int = Field(ge=0, le=100)
 
 class SmishingAnalysisResponse(BaseModel):
     status: str = Field(..., description="응답 상태 (SUCCESS / ERROR)")