vignette/apps/api/app/services/guardrail.py

"""입출력 가드레일 — PII 마스킹 · 위기분류 · 출력 안전레일.

MASTERPLAN §2.2 / §3.3 / R5 / R7 / F-03, MEMORY_DESIGN §B:
  [입력]  Presidio PII 마스킹(미설치 시 정규식 폴백) + 위기분류(실제위기 vs 페르소나 연기)
  [출력]  자살수단/방법 정보 차단, ideation_stage 상한(<=3) 강제.

설계 원칙:
  - 모듈 경계 명확: 입력 가드레일(mask_pii / classify_crisis)과 출력 가드레일(sanitize_client_reply)을
    순수함수에 가깝게 분리. IO·LLM·DB 의존 없음(테스트·재사용 용이).
  - 외부 LLM 경로 진입 전 *하드 게이트*: 마스킹 안 된 원문은 게이트웨이로 절대 안 나간다(F-03).
  - Presidio 는 선택 의존(미설치 환경에서도 import 가능해야 함) → 지연 로드 + 정규식 폴백.

TODO(Phase 2): Presidio MedicalNERRecognizer + 한국어 자살콘텐츠 분류기(JMIR few-shot 5단계, R8)
              로 교체. 현재 정규식/키워드 폴백은 1차 안전망(재현율 우선).
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional

# ── 출력 가드레일 상한 (R5) ──────────────────────────────
IDEATION_STAGE_CAP = 3  # 내담자 발화/상태가 넘을 수 없는 자살사고 단계 상한


# ════════════════════════════════════════════════════════════════════════════
# 1. PII 마스킹 (입력 — 저장·외부전송 전 하드 게이트, F-03)
# ════════════════════════════════════════════════════════════════════════════
# 정규식 폴백 패턴 (Presidio 미설치 시). 한국 맥락 우선.
# TODO: Presidio + MedicalNERRecognizer 로 정밀화(이름/주소/기관 NER).
_PII_PATTERNS: list[tuple[str, re.Pattern[str]]] = [
    # 주민등록번호 (6자리-7자리)
    ("RRN", re.compile(r"\b\d{6}[-\s]?\d{7}\b")),
    # 휴대폰 (010-1234-5678 등)
    ("PHONE", re.compile(r"\b01[016789][-\s]?\d{3,4}[-\s]?\d{4}\b")),
    # 일반 전화
    ("PHONE", re.compile(r"\b0\d{1,2}[-\s]?\d{3,4}[-\s]?\d{4}\b")),
    # 이메일
    ("EMAIL", re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b")),
    # 카드/계좌 유사 긴 숫자열 (12자리 이상)
    ("NUMID", re.compile(r"\b\d{12,}\b")),
]

# Presidio 지연 로드 캐시 (-1=미시도, None=미설치, 객체=설치됨)
_PRESIDIO_ANALYZER: object = -1
_PRESIDIO_ANONYMIZER: object = -1


def _try_load_presidio():
    """Presidio (analyzer, anonymizer) 지연 로드. 미설치면 (None, None)."""
    global _PRESIDIO_ANALYZER, _PRESIDIO_ANONYMIZER
    if _PRESIDIO_ANALYZER != -1:
        return _PRESIDIO_ANALYZER, _PRESIDIO_ANONYMIZER
    try:
        from presidio_analyzer import AnalyzerEngine  # type: ignore
        from presidio_anonymizer import AnonymizerEngine  # type: ignore

        _PRESIDIO_ANALYZER = AnalyzerEngine()
        _PRESIDIO_ANONYMIZER = AnonymizerEngine()
    except Exception:
        _PRESIDIO_ANALYZER = None
        _PRESIDIO_ANONYMIZER = None
    return _PRESIDIO_ANALYZER, _PRESIDIO_ANONYMIZER


@dataclass(slots=True)
class MaskResult:
    text_masked: str
    entities: list[str] = field(default_factory=list)  # 탐지된 엔티티 타입들
    used_presidio: bool = False


def mask_pii(text: str) -> MaskResult:
    """PII 마스킹. Presidio 가용 시 우선, 아니면 정규식 폴백.

    반환 text_masked 만 저장(turns.text_masked)·외부 LLM 전송에 사용한다(F-03).
    """
    if not text:
        return MaskResult(text_masked=text, entities=[], used_presidio=False)

    analyzer, anonymizer = _try_load_presidio()
    if analyzer is not None and anonymizer is not None:
        try:
            results = analyzer.analyze(text=text, language="en")  # TODO: ko 모델 등록 시 language="ko"
            ents = sorted({r.entity_type for r in results})
            anonymized = anonymizer.anonymize(text=text, analyzer_results=results)
            return MaskResult(text_masked=anonymized.text, entities=ents, used_presidio=True)
        except Exception:
            pass  # 폴백으로

    # 정규식 폴백
    masked = text
    found: list[str] = []
    for label, pat in _PII_PATTERNS:
        if pat.search(masked):
            found.append(label)
            masked = pat.sub(f"[{label}]", masked)
    return MaskResult(text_masked=masked, entities=sorted(set(found)), used_presidio=False)


# ════════════════════════════════════════════════════════════════════════════
# 2. 위기 분류 (입력 — 실제위기 vs 페르소나 연기 구분, R8)
# ════════════════════════════════════════════════════════════════════════════
class CrisisKind(str, Enum):
    NONE = "none"
    PERSONA_PLAY = "persona_play"     # 페르소나 연기 맥락의 위기 표현(시뮬레이션 정상)
    LEARNER_REAL = "learner_real"     # 수련생 본인의 실제 위기 신호(에스컬레이션 대상)


@dataclass(slots=True)
class CrisisResult:
    kind: CrisisKind
    risk_level: int = 0               # 0~5 (한국어 자살콘텐츠 5단계 자리; 현재 휴리스틱)
    matched: list[str] = field(default_factory=list)
    escalate: bool = False            # safety_events 적재 + 교수자 알림 트리거 여부


# 위기 표현 키워드(한국어 우선). TODO: JMIR 한국어 벤치 few-shot 분류기로 교체(R8).
_CRISIS_TERMS = [
    "죽고 싶", "죽고싶", "자살", "목숨", "사라지고 싶", "없어지고 싶",
    "자해", "끝내고 싶", "살기 싫", "살아서 뭐", "죽어야",
]
# 실제 위기로 가중되는 1인칭 현재 단서(수련생 본인 신호일 가능성)
_FIRST_PERSON_NOW = ["지금 나", "나 진짜", "제가 지금", "저 지금", "real", "도와주세요"]


def classify_crisis(text: str, *, speaker_is_persona_context: bool = True) -> CrisisResult:
    """위기 분류.

    Args:
      speaker_is_persona_context: True 면 상담 시뮬레이션 발화(수련생→가상내담자) 맥락.
        이 경우 위기 표현은 기본 PERSONA_PLAY 로 본다(연기). 단 1인칭 실제 단서가 강하면
        LEARNER_REAL 로 승격해 에스컬레이션(보수적, 재현율 우선).

    NOTE: 가상내담자(P1)의 자살사고 연기는 시뮬레이션 핵심이라 차단 대상이 아니다.
          진짜 위험은 *수련생 본인*의 실제 위기 신호다.
    """
    if not text:
        return CrisisResult(kind=CrisisKind.NONE)
    matched = [t for t in _CRISIS_TERMS if t in text]
    if not matched:
        return CrisisResult(kind=CrisisKind.NONE)

    risk = min(5, 2 + len(matched))  # 거친 단계화(휴리스틱)
    real_cues = [c for c in _FIRST_PERSON_NOW if c in text]

    if not speaker_is_persona_context or real_cues:
        # 실제 위기 가능 → 에스컬레이션(보수적)
        return CrisisResult(
            kind=CrisisKind.LEARNER_REAL,
            risk_level=max(risk, 4),
            matched=matched + real_cues,
            escalate=True,
        )
    # 시뮬레이션 연기 맥락
    return CrisisResult(
        kind=CrisisKind.PERSONA_PLAY,
        risk_level=risk,
        matched=matched,
        escalate=False,
    )


# ════════════════════════════════════════════════════════════════════════════
# 3. 출력 가드레일 (내담자 AI 응답 — 자살수단 차단, ideation 상한 R5)
# ════════════════════════════════════════════════════════════════════════════
# 자살/자해 '수단·방법' 정보 패턴 (절대 출력 금지). 정서 표현은 허용.
_MEANS_TERMS = [
    "약을 얼마나", "몇 알", "몇 정", "투신", "목을 매", "목 매", "번개탄", "농약",
    "손목을", "칼로", "뛰어내리", "수면제 ", "독", "방법은", "어떻게 죽",
]


@dataclass(slots=True)
class OutputGuardResult:
    text: str                         # 정제된(또는 원본) 응답
    blocked: bool = False             # 차단/치환 발생 여부
    needs_regeneration: bool = False  # 재생성 필요(수단정보 누출 시)
    reasons: list[str] = field(default_factory=list)


def sanitize_client_reply(text: str, *, ideation_stage: int) -> OutputGuardResult:
    """내담자 AI 응답 출력 가드레일.

    1. 자살/자해 수단·방법 정보가 있으면 → needs_regeneration=True(차단, 재생성 신호).
    2. ideation_stage > IDEATION_STAGE_CAP(3) 이면 안전 상한 위반으로 기록(상태머신이 상한 강제하나
       이중방어). 응답 자체 텍스트는 정서 표현이면 유지.
    """
    reasons: list[str] = []
    blocked = False
    needs_regen = False

    if text:
        for term in _MEANS_TERMS:
            if term in text:
                reasons.append(f"means_info:{term}")
                blocked = True
                needs_regen = True
                break

    if ideation_stage > IDEATION_STAGE_CAP:
        reasons.append(f"ideation_over_cap:{ideation_stage}>{IDEATION_STAGE_CAP}")
        blocked = True

    return OutputGuardResult(
        text=text,
        blocked=blocked,
        needs_regeneration=needs_regen,
        reasons=reasons,
    )


def clamp_ideation(stage: int) -> int:
    """ideation_stage 를 안전 상한(3)으로 클램프 (R5)."""
    return max(1, min(IDEATION_STAGE_CAP, stage))


__all__ = [
    "IDEATION_STAGE_CAP",
    "MaskResult",
    "mask_pii",
    "CrisisKind",
    "CrisisResult",
    "classify_crisis",
    "OutputGuardResult",
    "sanitize_client_reply",
    "clamp_ideation",
]