[Feat] 페이지 스냅샷 추가

app/core/config.py

@@ -194,6 +194,15 @@ def alembic_database_url(self) -> str:
     content_render_settle_ms: int = 500
     content_render_concurrency: int = 2
 
+    # 사용자 표시용 페이지 스냅샷. 파이프라인 응답 SLA 보호를 위해 동기 경로에서는
+    # 기본 1초까지만 기다리고, 지연/실패는 verdict 와 분리해 상태로만 내려준다.
+    page_snapshot_enabled: bool = True
+    page_snapshot_timeout_seconds: float = 1.0
+    page_snapshot_storage_dir: str = "/private/tmp/linclean-page-snapshots"
+    page_snapshot_viewport_width: int = 1365
+    page_snapshot_viewport_height: int = 768
+    page_snapshot_full_page: bool = False
+
     # 콘텐츠 분석 점수
     score_weight_brand_impersonation: int = 50
     score_weight_logo_alt_impersonation: int = 10

app/schemas/__init__.py

@@ -4,10 +4,13 @@
     DbIndependentPipelineStages,
     DbIndependentPipelineSuccess,
 )
+from app.schemas.page_snapshot import PageSnapshotResult, PageSnapshotStatus
 
 __all__ = [
     "DbIndependentPipelineFailure",
     "DbIndependentPipelineResult",
     "DbIndependentPipelineStages",
     "DbIndependentPipelineSuccess",
+    "PageSnapshotResult",
+    "PageSnapshotStatus",
 ]

app/schemas/analysis.py

@@ -16,6 +16,7 @@
 )
 from app.schemas.domain_heuristic import DomainHeuristicResult, DomainHeuristicSignal, RdapInfo
 from app.schemas.normalize import NormalizeResult
+from app.schemas.page_snapshot import PageSnapshotResult, PageSnapshotStatus
 from app.schemas.pipeline import (
     PipelineFailure,
     PipelineResult,
@@ -47,6 +48,8 @@
     "GSBResult",
     "HopRecord",
     "NormalizeResult",
+    "PageSnapshotResult",
+    "PageSnapshotStatus",
     "PipelineFailure",
     "PipelineResult",
     "PipelineStage",

app/schemas/db_independent_pipeline.py

@@ -5,6 +5,7 @@
 from app.schemas.content_analysis import ContentAnalysisResult
 from app.schemas.domain_heuristic import DomainHeuristicResult
 from app.schemas.normalize import NormalizeResult
+from app.schemas.page_snapshot import PageSnapshotResult
 from app.schemas.pipeline import PipelineStage, PipelineTimings, Verdict
 from app.schemas.unchain import UnchainResult
 
@@ -27,6 +28,7 @@ class DbIndependentPipelineSuccess(BaseModel):
     final_url: str
     verdict: Verdict
     score: int = Field(ge=0, le=100)
+    snapshot: PageSnapshotResult | None = None
     timings: PipelineTimings | None = None
     stages: DbIndependentPipelineStages
 

app/schemas/page_snapshot.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+from enum import StrEnum
+
+from pydantic import BaseModel, Field
+
+
+class PageSnapshotStatus(StrEnum):
+    AVAILABLE = "available"
+    SKIPPED = "skipped"
+    TIMEOUT = "timeout"
+    FAILED = "failed"
+
+
+class PageSnapshotResult(BaseModel):
+    status: PageSnapshotStatus
+    final_url: str
+    storage_key: str | None = None
+    elapsed_seconds: float | None = Field(default=None, ge=0)
+    error: str | None = None

app/schemas/pipeline.py

@@ -6,6 +6,7 @@
 from app.schemas.content_analysis import ContentAnalysisResult
 from app.schemas.domain_heuristic import DomainHeuristicResult
 from app.schemas.normalize import NormalizeResult
+from app.schemas.page_snapshot import PageSnapshotResult
 from app.schemas.threat_db import ThreatDbResult
 from app.schemas.unchain import UnchainResult
 
@@ -69,6 +70,7 @@ class PipelineSuccess(BaseModel):
     verdict: Verdict
     score: int = Field(ge=0, le=100)
     summary: str | None = None
+    snapshot: PageSnapshotResult | None = None
     timings: PipelineTimings | None = None
     stages: PipelineStages
 

app/services/page_snapshot.py

@@ -0,0 +1,130 @@
+"""사용자 표시용 페이지 스냅샷 생성."""
+
+from __future__ import annotations
+
+import asyncio
+import re
+import time
+from collections.abc import Callable
+from pathlib import Path
+from typing import Any
+
+from app.core.config import settings
+from app.core.logging import get_logger
+from app.schemas.page_snapshot import PageSnapshotResult, PageSnapshotStatus
+from app.services.content_analyzer.fetch import _pick_user_agent
+from app.services.content_analyzer.render import _target_blocked
+
+logger = get_logger(__name__)
+
+_SAFE_KEY = re.compile(r"[^A-Za-z0-9_.-]+")
+
+
+def _elapsed_seconds(started: float) -> float:
+    return round(time.perf_counter() - started, 6)
+
+
+def skipped_page_snapshot(final_url: str, reason: str) -> PageSnapshotResult:
+    return PageSnapshotResult(
+        status=PageSnapshotStatus.SKIPPED,
+        final_url=final_url,
+        error=reason,
+    )
+
+
+def timed_out_page_snapshot(final_url: str, started: float | None = None) -> PageSnapshotResult:
+    return PageSnapshotResult(
+        status=PageSnapshotStatus.TIMEOUT,
+        final_url=final_url,
+        elapsed_seconds=_elapsed_seconds(started) if started is not None else None,
+        error="timeout",
+    )
+
+
+def failed_page_snapshot(
+    final_url: str,
+    *,
+    error: str,
+    started: float | None = None,
+) -> PageSnapshotResult:
+    return PageSnapshotResult(
+        status=PageSnapshotStatus.FAILED,
+        final_url=final_url,
+        elapsed_seconds=_elapsed_seconds(started) if started is not None else None,
+        error=error,
+    )
+
+
+def _storage_path(analysis_id: str) -> tuple[Path, str]:
+    safe_id = _SAFE_KEY.sub("-", analysis_id).strip("-") or "snapshot"
+    storage_dir = Path(settings.page_snapshot_storage_dir)
+    filename = f"{safe_id}.png"
+    return storage_dir / filename, filename
+
+
+def _load_playwright() -> tuple[type[BaseException], Callable[[], Any]]:
+    from playwright.async_api import TimeoutError as PlaywrightTimeoutError
+    from playwright.async_api import async_playwright
+
+    return PlaywrightTimeoutError, async_playwright
+
+
+async def capture_page_snapshot(analysis_id: str, final_url: str) -> PageSnapshotResult:
+    """Playwright 로 최종 URL 첫 화면 PNG 를 저장한다.
+
+    보안 차단은 콘텐츠 렌더링과 같은 대상 host 검사를 재사용한다. Playwright 가 없거나
+    브라우저 실행이 실패하면 파이프라인을 깨지 않고 failed 상태를 반환한다.
+    """
+    started = time.perf_counter()
+
+    if not settings.page_snapshot_enabled:
+        return skipped_page_snapshot(final_url, "snapshot_disabled")
+
+    if await _target_blocked(final_url):
+        logger.info("page_snapshot.blocked_host", url=final_url)
+        return failed_page_snapshot(final_url, error="blocked_host", started=started)
+
+    try:
+        playwright_timeout_error, async_playwright = _load_playwright()
+    except ImportError:
+        return failed_page_snapshot(final_url, error="playwright_unavailable", started=started)
+
+    path, storage_key = _storage_path(analysis_id)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    timeout_ms = int(settings.page_snapshot_timeout_seconds * 1000)
+
+    try:
+        async with async_playwright() as p:
+            browser = await p.chromium.launch(headless=True)
+            try:
+                page = await browser.new_page(
+                    user_agent=_pick_user_agent(),
+                    locale="ko-KR",
+                    viewport={
+                        "width": settings.page_snapshot_viewport_width,
+                        "height": settings.page_snapshot_viewport_height,
+                    },
+                )
+                await page.goto(final_url, wait_until="domcontentloaded", timeout=timeout_ms)
+                await page.screenshot(path=str(path), full_page=settings.page_snapshot_full_page)
+            finally:
+                await browser.close()
+    except playwright_timeout_error:
+        return timed_out_page_snapshot(final_url, started)
+    except asyncio.CancelledError:
+        raise
+    except Exception as exc:
+        logger.warning(
+            "page_snapshot.failed",
+            url=final_url,
+            error=str(exc),
+            error_type=type(exc).__name__,
+        )
+        return failed_page_snapshot(final_url, error="snapshot_failed", started=started)
+
+    return PageSnapshotResult(
+        status=PageSnapshotStatus.AVAILABLE,
+        final_url=final_url,
+        storage_key=storage_key,
+        elapsed_seconds=_elapsed_seconds(started),
+    )

app/services/pipeline.py

@@ -21,6 +21,7 @@
     DomainHeuristicSkippedReason,
 )
 from app.schemas.normalize import NormalizeResult
+from app.schemas.page_snapshot import PageSnapshotResult
 from app.schemas.pipeline import (
     PipelineFailure,
     PipelineStage,
@@ -36,6 +37,12 @@
 from app.services.content_analyzer import analyze_content, skipped_already_danger
 from app.services.domain_heuristic import check_domain_heuristic
 from app.services.normalizer import normalize_url
+from app.services.page_snapshot import (
+    capture_page_snapshot,
+    failed_page_snapshot,
+    skipped_page_snapshot,
+    timed_out_page_snapshot,
+)
 from app.services.page_unavailability import (
     PAGE_UNAVAILABLE_CODE,
     content_page_unavailable,
@@ -213,6 +220,40 @@ async def _stage_content_analysis(
     return result
 
 
+async def _capture_snapshot_for_pipeline(
+    log: structlog.stdlib.BoundLogger,
+    analysis_id: str,
+    final_url: str,
+) -> PageSnapshotResult:
+    started = time.perf_counter()
+    try:
+        result = await asyncio.wait_for(
+            capture_page_snapshot(analysis_id, final_url),
+            timeout=settings.page_snapshot_timeout_seconds,
+        )
+    except TimeoutError:
+        result = timed_out_page_snapshot(final_url, started)
+    except asyncio.CancelledError:
+        raise
+    except Exception as exc:
+        log.warning(
+            "pipeline.page_snapshot.failed",
+            final_url=final_url,
+            error=str(exc),
+            error_type=type(exc).__name__,
+        )
+        result = failed_page_snapshot(final_url, error="snapshot_failed", started=started)
+
+    log.info(
+        "pipeline.page_snapshot.done",
+        status=result.status.value,
+        storage_key=result.storage_key,
+        elapsed_seconds=result.elapsed_seconds,
+        error=result.error,
+    )
+    return result
+
+
 def _preceding_score(threat: ThreatDbResult, heuristic: DomainHeuristicResult) -> int:
     """2~3단계 누적 점수. 4단계 건너뛸지 판단하는 기준."""
     score = heuristic.score
@@ -307,6 +348,7 @@ def _pipeline_success(
     threat: ThreatDbResult,
     heuristic: DomainHeuristicResult,
     content: ContentAnalysisResult,
+    snapshot: PageSnapshotResult | None = None,
 ) -> PipelineSuccess:
     return PipelineSuccess(
         analysis_id=analysis_id,
@@ -320,6 +362,7 @@ def _pipeline_success(
             heuristic=heuristic,
             content=content,
         ),
+        snapshot=snapshot,
         timings=_build_timings(started, stage_timings),
         stages=PipelineStages(
             normalize=normalize,
@@ -499,6 +542,7 @@ async def run_pipeline(
         score = _total_score(threat, heuristic, content)
         if threat.is_malicious or score >= settings.score_caution_threshold:
             verdict = _decide_verdict(score, threat)
+            page_unavailable_snapshot = skipped_page_snapshot(unchain.final_url, "page_unavailable")
             return _pipeline_success(
                 analysis_id=analysis_id,
                 original_url=original_url,
@@ -512,6 +556,7 @@ async def run_pipeline(
                 threat=threat,
                 heuristic=heuristic,
                 content=content,
+                snapshot=page_unavailable_snapshot,
             )
         return _page_unavailable_failure(
             analysis_id=analysis_id,
@@ -557,6 +602,8 @@ async def run_pipeline(
         short_circuited = False
 
     heuristic = _augment_heuristic_with_redirect_signals(heuristic, unchain)
+    snapshot: PageSnapshotResult | None = None
+    snapshot_task: asyncio.Task[PageSnapshotResult] | None = None
 
     if short_circuited:
         log.info(
@@ -568,6 +615,7 @@ async def run_pipeline(
         stage_started = time.perf_counter()
         content = skipped_already_danger(unchain.final_url)
         _set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, stage_started)
+        snapshot = skipped_page_snapshot(unchain.final_url, "skipped_already_danger")
     else:
         # known malicious 는 verdict 가 이미 외부 DB 로 확정됐으므로 페이지를 받아보지 않는다.
         # 휴리스틱 danger 는 페이지가 존재하지 않을 수 있으므로 content fetch 로 가용성을 확인한다.
@@ -581,8 +629,12 @@ async def run_pipeline(
             stage_started = time.perf_counter()
             content = skipped_already_danger(unchain.final_url)
             _set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, stage_started)
+            snapshot = skipped_page_snapshot(unchain.final_url, "skipped_already_danger")
         else:
             upstream = _collect_upstream_signals(threat, heuristic)
+            snapshot_task = asyncio.create_task(
+                _capture_snapshot_for_pipeline(log, analysis_id, unchain.final_url)
+            )
             try:
                 content = await deadline.run(
                     PipelineStage.CONTENT_ANALYSIS.value,
@@ -610,6 +662,8 @@ async def run_pipeline(
                 )
                 content = timed_out_content_result(unchain.final_url)
 
+            snapshot = await snapshot_task
+
     if unavailable := content_page_unavailable(content):
         message, status_code = unavailable
         log.info(
@@ -635,6 +689,7 @@ async def run_pipeline(
                 threat=threat,
                 heuristic=heuristic,
                 content=content,
+                snapshot=snapshot,
             )
         return _page_unavailable_failure(
             analysis_id=analysis_id,
@@ -668,4 +723,5 @@ async def run_pipeline(
         threat=threat,
         heuristic=heuristic,
         content=content,
+        snapshot=snapshot,
     )