[FEATURE] AI 비의존 API 추가

app/api/v1/endpoints/analyze.py

@@ -100,6 +100,30 @@ async def analyze_sync(
     )
 
 
+@router.post(
+    "/analyze/no-ai/sync",
+    response_model=PipelineSuccess | PipelineFailure,
+    summary="전체 파이프라인 — AI 비사용 동기 결과 반환",
+    description=(
+        "run_pipeline 을 AI 추론 없이 동기로 실행합니다. normalize, unchain, threat DB, "
+        "도메인 휴리스틱, 콘텐츠 정적 분석은 수행하지만 OpenAI provider 를 호출하지 않고 "
+        "AI verdict/score 가산 없이 규칙 기반 verdict 와 summary 를 반환합니다."
+    ),
+)
+async def analyze_no_ai_sync(
+    body: AnalyzeSyncRequest,
+    session: DBSession,
+    _: InternalApiKey,
+) -> PipelineSuccess | PipelineFailure:
+    analysis_id = str(uuid.uuid4())
+    return await run_pipeline(
+        analysis_id=analysis_id,
+        original_url=body.url,
+        session=session,
+        use_ai=False,
+    )
+
+
 @router.post(
     "/analyze/db-independent/sync",
     response_model=DbIndependentPipelineSuccess | DbIndependentPipelineFailure,

app/schemas/pipeline.py

@@ -68,6 +68,7 @@ class PipelineSuccess(BaseModel):
     final_url: str
     verdict: Verdict
     score: int = Field(ge=0, le=100)
+    summary: str | None = None
     timings: PipelineTimings | None = None
     stages: PipelineStages
 

app/services/analysis_callback.py

@@ -9,13 +9,13 @@
 
 from app.core.config import settings
 from app.core.logging import get_logger
-from app.schemas.pipeline import PipelineFailure, PipelineStage, PipelineSuccess, Verdict
+from app.schemas.pipeline import PipelineFailure, PipelineStage, PipelineSuccess
+from app.services.analysis_summary import build_analysis_summary
 
 logger = get_logger(__name__)
 
 _CALLBACK_PATH = "/internal/analysis-result"
 _MAX_ATTEMPTS = 3
-_KNOWN_MALICIOUS_SUMMARY = "악성으로 알려진 페이지 입니다."
 
 _STAGE_NUMBERS: dict[PipelineStage, int] = {
     PipelineStage.NORMALIZE: 1,
@@ -191,8 +191,7 @@ def _callback_reasons(result: PipelineSuccess) -> list[dict[str, Any]]:
                 stage=4,
                 weight=settings.score_weight_ai_phishing,
                 message=(
-                    stages.content_analysis.ai_reason
-                    or "AI가 피싱 가능성이 높다고 판단했습니다."
+                    stages.content_analysis.ai_reason or "AI가 피싱 가능성이 높다고 판단했습니다."
                 ),
             )
         elif verdict == "suspicious":
@@ -202,36 +201,27 @@ def _callback_reasons(result: PipelineSuccess) -> list[dict[str, Any]]:
                 stage=4,
                 weight=settings.score_weight_ai_suspicious,
                 message=(
-                    stages.content_analysis.ai_reason
-                    or "AI가 의심스러운 페이지로 판단했습니다."
+                    stages.content_analysis.ai_reason or "AI가 의심스러운 페이지로 판단했습니다."
                 ),
             )
 
     return reasons
 
 
 def _callback_summary(result: PipelineSuccess) -> str:
-    if result.stages.threat_db.is_malicious:
-        return _KNOWN_MALICIOUS_SUMMARY
-    content = result.stages.content_analysis
-    if content.ai_reason:
-        return content.ai_reason
-    if content.reason:
-        return content.reason
-    if result.verdict == Verdict.DANGER:
-        return "위험성이 높은 페이지입니다. 접속과 정보 입력을 피하세요."
-    if result.verdict == Verdict.CAUTION:
-        return "주의가 필요한 페이지입니다. 링크와 입력 정보를 한 번 더 확인하세요."
-    return "현재 분석 기준에서 뚜렷한 위험 신호가 확인되지 않았습니다."
+    if result.summary:
+        return result.summary
+    return build_analysis_summary(
+        verdict=result.verdict,
+        threat=result.stages.threat_db,
+        heuristic=result.stages.domain_heuristic,
+        content=result.stages.content_analysis,
+    )
 
 
 def _spring_stages(result: PipelineSuccess) -> dict[str, Any]:
     stages = result.stages
-    gsb_matches = [
-        match.threat_type
-        for match in stages.threat_db.gsb.matches
-        if match.threat_type
-    ]
+    gsb_matches = [match.threat_type for match in stages.threat_db.gsb.matches if match.threat_type]
 
     chain = [hop.url for hop in stages.unchain.hops]
     if not chain:

app/services/analysis_summary.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+from typing import Any
+
+from app.schemas.content_analysis import ContentAnalysisResult
+from app.schemas.domain_heuristic import DomainHeuristicResult
+from app.schemas.threat_db import ThreatDbResult
+
+KNOWN_MALICIOUS_SUMMARY = "악성으로 알려진 페이지 입니다."
+SAFE_SUMMARY = "현재 분석 기준에서 뚜렷한 위험 신호가 확인되지 않았습니다."
+CAUTION_SUMMARY = "주의가 필요한 페이지입니다. 링크와 입력 정보를 한 번 더 확인하세요."
+DANGER_SUMMARY = "위험성이 높은 페이지입니다. 접속과 정보 입력을 피하세요."
+
+
+def _code(value: Any) -> str:
+    return value.value if hasattr(value, "value") else str(value)
+
+
+def build_analysis_summary(
+    *,
+    verdict: Any,
+    threat: ThreatDbResult,
+    heuristic: DomainHeuristicResult,
+    content: ContentAnalysisResult,
+) -> str:
+    if threat.is_malicious:
+        return KNOWN_MALICIOUS_SUMMARY
+    if content.ai_reason:
+        return content.ai_reason
+
+    domain_signals = {_code(signal) for signal in heuristic.signals}
+    content_signals = {_code(signal) for signal in content.signals}
+
+    if "FETCH_FAILED" in content_signals:
+        return content.reason or "페이지를 확인할 수 없어 안전 여부를 판단하기 어렵습니다."
+    if "SKIPPED_ALREADY_DANGER" in content_signals:
+        return content.reason or DANGER_SUMMARY
+
+    if "BRAND_IMPERSONATION_FORM" in content_signals and (
+        content.has_password_field or "SENSITIVE_PATH" in domain_signals
+    ):
+        return "브랜드 사칭과 로그인 정보 입력 유도 정황이 있어 비밀번호를 입력하지 마세요."
+    if "CREDENTIAL_FORM_EXTERNAL" in content_signals:
+        return "로그인 정보가 외부 주소로 전송될 수 있어 계정 정보를 입력하지 마세요."
+    if "BRAND_IMPERSONATION_FORM" in content_signals or "LOGO_ALT_IMPERSONATION" in content_signals:
+        return "브랜드 사칭 정황이 있어 링크와 입력 정보를 신중히 확인하세요."
+    if {"PII_COLLECTION_FORM", "SENSITIVE_ID_FIELD", "FINANCIAL_FIELD"} & content_signals:
+        return "민감정보 입력을 유도하는 정황이 있어 개인정보나 결제 정보를 입력하지 마세요."
+    if "RISKY_DOWNLOAD_LINK" in content_signals:
+        return "위험한 파일 다운로드를 유도할 수 있어 파일을 열지 마세요."
+    if {"PUBLIC_AGENCY_LURE", "KOREAN_LURE_TEXT"} & content_signals:
+        return "국내 사용자 또는 공공기관을 사칭하는 유인 문구가 있어 주의하세요."
+
+    if "TYPO_DOMAIN" in domain_signals:
+        return "유명 서비스와 유사한 도메인이므로 로그인이나 결제 정보를 입력하지 마세요."
+    if {"URL_USERINFO", "OPEN_REDIRECT_PARAM", "REDIRECT_CROSS_ORIGIN"} & domain_signals:
+        return "URL이 다른 사이트 이동이나 위장에 악용될 수 있어 접속 전 주소를 확인하세요."
+    if {"NEW_DOMAIN", "SUSPICIOUS_TLD", "DGA_LIKE"} & domain_signals:
+        return "신뢰도가 낮은 도메인 특성이 있어 링크와 입력 정보를 한 번 더 확인하세요."
+
+    if content.reason:
+        return content.reason
+    verdict_code = _code(verdict)
+    if verdict_code == "danger":
+        return DANGER_SUMMARY
+    if verdict_code == "caution":
+        return CAUTION_SUMMARY
+    return SAFE_SUMMARY

app/services/content_analyzer/ai.py

@@ -55,6 +55,12 @@ class AIProvider(Protocol):
     async def infer(self, ctx: AIPromptContext) -> AIInference | None: ...
 
 
+class AIProviderError(Exception):
+    def __init__(self, code: str, message: str) -> None:
+        super().__init__(message)
+        self.code = code
+
+
 class NullAIProvider:
     """모델이 설정되지 않았을 때의 기본 구현. 추론을 시도하지 않고 None 반환.
 

app/services/content_analyzer/analyze.py

@@ -19,7 +19,12 @@
     ContentSignal,
     TokenUsage,
 )
-from app.services.content_analyzer.ai import AIPromptContext, AIProvider, get_ai_provider
+from app.services.content_analyzer.ai import (
+    AIPromptContext,
+    AIProvider,
+    AIProviderError,
+    get_ai_provider,
+)
 from app.services.content_analyzer.extract import ExtractedFeatures, extract_features_async
 from app.services.content_analyzer.fetch import fetch_page
 from app.services.content_analyzer.render import render_page
@@ -114,8 +119,7 @@ def _merge_features(static: ExtractedFeatures, rendered: ExtractedFeatures) -> E
         title=rendered.title or static.title,
         has_password_field=static.has_password_field or rendered.has_password_field,
         has_password_form_external_action=(
-            static.has_password_form_external_action
-            or rendered.has_password_form_external_action
+            static.has_password_form_external_action or rendered.has_password_form_external_action
         ),
         has_meta_refresh=static.has_meta_refresh or rendered.has_meta_refresh,
         has_external_meta_refresh=(
@@ -246,6 +250,7 @@ async def analyze_content(
     *,
     provider: AIProvider | None = None,
     upstream_signals: Iterable[str] | Awaitable[Iterable[str]] = (),
+    use_ai: bool = True,
 ) -> ContentAnalysisResult:
     """콘텐츠 정적 분석 진입점.
 
@@ -289,27 +294,38 @@ async def analyze_content(
     ai_model: str | None = None
     ai_token_usage: TokenUsage | None = None
 
-    active_provider = provider if provider is not None else get_ai_provider()
     if inspect.isawaitable(upstream_signals):
         upstream_tuple = tuple(await upstream_signals)
     else:
         upstream_tuple = tuple(upstream_signals)
-    try:
-        inference = await active_provider.infer(
-            _build_prompt_context(final_url, features, upstream_tuple)
-        )
-    except asyncio.CancelledError:
-        # shutdown / 요청 타임아웃 신호는 degraded 결과로 흡수하지 않는다
-        raise
-    except Exception as exc:
-        logger.warning(
-            "content_analysis.ai_error",
-            url=final_url,
-            error=str(exc),
-            error_type=type(exc).__name__,
-        )
-        ai_error = "ai_unavailable"
-        inference = None
+    active_provider: AIProvider | None = None
+    inference = None
+    if use_ai:
+        active_provider = provider if provider is not None else get_ai_provider()
+        try:
+            inference = await active_provider.infer(
+                _build_prompt_context(final_url, features, upstream_tuple)
+            )
+        except asyncio.CancelledError:
+            # shutdown / 요청 타임아웃 신호는 degraded 결과로 흡수하지 않는다
+            raise
+        except AIProviderError as exc:
+            logger.warning(
+                "content_analysis.ai_error",
+                url=final_url,
+                error=str(exc),
+                error_code=exc.code,
+                error_type=type(exc).__name__,
+            )
+            ai_error = exc.code
+        except Exception as exc:
+            logger.warning(
+                "content_analysis.ai_error",
+                url=final_url,
+                error=str(exc),
+                error_type=type(exc).__name__,
+            )
+            ai_error = "ai_unavailable"
 
     score = scoring.score
     if inference is not None:
@@ -323,7 +339,7 @@ async def analyze_content(
             upstream_tuple,
         ):
             score += _ai_score_weight(inference.verdict)
-    elif ai_error is None:
+    elif use_ai and ai_error is None and active_provider is not None:
         # 추론이 None 인데 호출 단계 예외도 없었다면 NullAIProvider 동작.
         # 부팅 시 misconfiguration 으로 폴백된 NullProvider 면 fallback_reason 을 응답에 노출.
         # 정상 NullProvider 는 fallback_reason=None 이라 ai_error 가 None 으로 유지된다.