classifier.py

"""문서 보안등급 분류기 (Rule-based / Zero-shot)
====================================================

SPEC_파일분류_PoC.md §2.2 / §2.3 / §10 를 기반으로 한 **간단한** 1차 분류 모델.

설계 의도
---------
- SPEC 의 최종 목표는 KoELECTRA + mDeBERTa 앙상블 + 사용자 피드백 루프지만,
  그 모델은 **사용자 레이블이 누적되어야** 학습이 가능하다 (콜드스타트 문제).
- 본 모듈은 **부트스트랩 1단계 — zero-shot 자동 레이블 생성기**.
  (SPEC §2.3 ① "룰 + 키워드 매칭만으로 자동 레이블")
- 점수(score) 기반 결정 → 임계값(threshold) 으로 C / S / O 분류.
- "왜 그 등급인가" 를 함께 돌려준다 (SPEC 기능 4 — XAI).

C / S / O 정의 (SPEC §2.2)
--------------------------
- **C** Critical / 위험   : 직접식별자(주민번호·여권·카드 등) 또는 강한 등급 마커
- **S** Sensitive / 민감  : API 키, 사업자번호, 내부 프로젝트명, VIP 등
- **O** Open / 공개       : 그 외

수식
----
    score = Σ (w_e · count_e)  +  Σ (w_k · min(count_k, 3))
            (entity)              (grade keyword)

    grade = C  if score >= C_THRESHOLD
            S  if score >= S_THRESHOLD
            O  otherwise

신뢰도(confidence) = 0.55 + 0.4 · tanh(margin / 2)
  margin = 임계값에서 떨어진 거리 (밴드 내부 폭은 양쪽 경계까지의 최소거리)
"""
from __future__ import annotations

import math
import re
from typing import Iterable

# 등급 → 정수 (SPEC §10.1, gap 계산용)
GRADE_RANK = {"O": 0, "S": 1, "C": 2}

# entity_type 별 단일 매치당 가산점
ENTITY_WEIGHTS: dict[str, float] = {
    # --- Critical 후보 (직접식별자) ---
    "KR_RRN":            5.0,   # 한국 주민등록번호
    "KR_PASSPORT":       4.5,
    "CREDIT_CARD":       4.5,
    "US_SSN":            4.5,
    "IBAN_CODE":         3.0,
    # --- Sensitive 후보 (계정·비즈·내부) ---
    "AWS_ACCESS_KEY":    3.5,
    "GENERIC_API_KEY":   1.0,   # 오탐 잦음 (32자 임의 토큰) → 낮춤
    "KR_BIZ_NO":         2.5,
    "VIP_NAMES":         2.0,
    "INTERNAL_PROJECTS": 2.0,
    "KR_ADDRESS":        1.5,
    # --- 약한 식별자 ---
    "IP_ADDRESS":        0.4,
    "KR_PHONE":          0.5,
    "PHONE_NUMBER":      0.5,
    "EMAIL_ADDRESS":     0.4,
    "PERSON":            0.3,
    "LOCATION":          0.15,  # 일반 지명 흔함 — 낮춤
    "ORGANIZATION":      0.15,  # 일반 회사명 흔함 — 낮춤
    "URL":               0.0,   # 일반 URL — PII 아님
    "DATE_TIME":         0.0,   # 일반 날짜 / 버전 번호 (1.9.16 등) — PII 아님
    # --- 비-PII (NER 모델 noise) — fallback DEFAULT 가 0.3 으로 잡혀 학습 dominate 방지 ---
    "NRP":               0.0,   # NORP = Nationalities/Religious/Political (spaCy 영어 NER)
    "NORP":              0.0,   # spaCy 원본 라벨 (Presidio 가 NRP 로 축약 표기)
    # --- 한국 환경 무관 외국 PII (Presidio default 정규식, 한국어 텍스트엔 오탐) ---
    "US_DRIVER_LICENSE": 0.0,   # s3, y3, X0 같은 2글자 오탐 다수
    "US_ITIN":           0.0,
    "US_PASSPORT":       0.0,
    "IN_PAN":            0.0,
}
DEFAULT_ENTITY_WEIGHT = 0.05   # 미명시 entity 가 학습을 dominate 하지 않도록 (기존 0.3 → 0.05)

# 본문 등급 키워드 — 명시적 라벨이 박혀있을 때 강하게 반영
# (kw_lower, weight, display_label)
GRADE_KEYWORDS: list[tuple[str, float, str]] = [
    ("극비",         4.0, "극비"),
    ("top secret",   4.0, "Top Secret"),
    ("대외비",       3.0, "대외비"),
    ("기밀",         3.0, "기밀"),
    ("confidential", 3.0, "Confidential"),
    ("secret",       2.5, "Secret"),
    ("내부용",       1.5, "내부용"),
    ("사내한정",     1.5, "사내한정"),
    ("internal use", 1.5, "Internal Use"),
    ("restricted",   1.5, "Restricted"),
    ("private",      1.0, "Private"),
    ("개인정보",     1.0, "개인정보 라벨"),
]

# 동일 키워드 누적 상한 — "비밀 비밀 비밀..." 어뷰징 방지
KW_COUNT_CAP = 3

# 임계값 — 손튜닝값. 데이터 누적 후 SPEC §10.4 의 Platt scaling 으로 보정 예정
C_THRESHOLD = 5.0
S_THRESHOLD = 2.0

CLASSIFIER_VERSION = "rule-v1"           # 모듈 출하 시 베이스 버전 (불변)
_active_version    = CLASSIFIER_VERSION  # 학습/핫스왑 후 갱신되는 활성 버전


def active_version() -> str:
    """현재 활성 모델 버전. /api/analyze 응답 등에서 사용."""
    return _active_version


def set_active_version(v: str) -> None:
    """train.apply_new_weights() 가 핫스왑 후 호출 — 버전 라벨 업데이트."""
    global _active_version
    _active_version = v


# ---------------------------------------------------------------------------
# extra grade keywords (사용자 추가) — 부팅 시 GRADE_KEYWORDS 에 병합 / 영속화
# ---------------------------------------------------------------------------
import json as _json
from pathlib import Path as _Path

_EXTRA_KW_FILE = _Path(__file__).resolve().parent / "extra_grade_keywords.json"


def _load_extra_kw() -> list[dict]:
    if not _EXTRA_KW_FILE.exists():
        return []
    try:
        return _json.loads(_EXTRA_KW_FILE.read_text(encoding="utf-8"))
    except Exception:
        return []


def _save_extra_kw(items: list[dict]) -> None:
    _EXTRA_KW_FILE.write_text(
        _json.dumps(items, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )


def _is_builtin_kw(label: str) -> bool:
    """초기 GRADE_KEYWORDS 에 등록된 키워드인지 — 사용자 삭제 불가."""
    return label in {l for (_kw, _w, l) in _BUILTIN_GRADE_KEYWORDS}


# 빌트인 스냅샷 (load_extras 가 여러 번 호출돼도 중복 추가 안 되도록)
_BUILTIN_GRADE_KEYWORDS: list[tuple[str, float, str]] = list(GRADE_KEYWORDS)


def reload_extra_grade_keywords() -> int:
    """파일에서 다시 읽어 GRADE_KEYWORDS 를 빌트인 + 사용자 정의로 재구성. 반환: 추가된 항목 수."""
    extras = _load_extra_kw()
    GRADE_KEYWORDS.clear()
    GRADE_KEYWORDS.extend(_BUILTIN_GRADE_KEYWORDS)
    for x in extras:
        kw = (x.get("keyword") or "").lower()
        w = float(x.get("weight", 1.0))
        lbl = x.get("label") or x.get("keyword") or ""
        if kw:
            GRADE_KEYWORDS.append((kw, w, lbl))
    return len(extras)


def add_extra_grade_keyword(keyword: str, weight: float, label: str) -> dict:
    """사용자 정의 등급 키워드 1개 추가 (이미 있으면 가중치/라벨 갱신).

    Returns: {keyword, weight, label, action: 'added'|'updated'}
    """
    keyword = (keyword or "").strip().lower()
    if not keyword:
        raise ValueError("keyword is empty")
    label = (label or keyword).strip()
    items = _load_extra_kw()
    action = "added"
    for it in items:
        if it.get("keyword", "").lower() == keyword:
            it["weight"] = float(weight)
            it["label"]  = label
            action = "updated"
            break
    else:
        items.append({"keyword": keyword, "weight": float(weight), "label": label})
    _save_extra_kw(items)
    reload_extra_grade_keywords()
    return {"keyword": keyword, "weight": float(weight), "label": label, "action": action}


def remove_extra_grade_keyword(keyword: str) -> dict:
    """사용자 추가 키워드 삭제. 빌트인은 삭제 불가."""
    keyword = (keyword or "").strip().lower()
    items = _load_extra_kw()
    before = len(items)
    items = [it for it in items if it.get("keyword", "").lower() != keyword]
    if len(items) == before:
        return {"keyword": keyword, "removed": False, "reason": "not in extras (built-in or unknown)"}
    _save_extra_kw(items)
    reload_extra_grade_keywords()
    return {"keyword": keyword, "removed": True}


def list_extra_grade_keywords() -> list[dict]:
    return _load_extra_kw()


# 모듈 로드 시 자동 병합
reload_extra_grade_keywords()


def _grade_for_score(score: float) -> tuple[str, str, float]:
    """score → (grade, label, margin). margin 은 가장 가까운 임계값까지의 거리."""
    if score >= C_THRESHOLD:
        return "C", "위험 (Critical)", score - C_THRESHOLD
    if score >= S_THRESHOLD:
        return "S", "민감 (Sensitive)", min(score - S_THRESHOLD, C_THRESHOLD - score)
    return "O", "공개 (Open)", S_THRESHOLD - score


def _confidence(margin: float) -> float:
    """margin → 0.55..0.95 범위의 신뢰도. 임계값에 가까울수록 0.55."""
    if margin < 0:
        margin = 0.0
    return round(0.55 + 0.4 * math.tanh(margin / 2.0), 3)


def _scan_keywords(text: str) -> list[dict]:
    """본문에서 등급 키워드 매칭 → 기여도 dict 리스트."""
    if not text:
        return []
    lower = text.lower()
    out: list[dict] = []
    for kw, w, label in GRADE_KEYWORDS:
        cnt = lower.count(kw)
        if cnt <= 0:
            continue
        capped = min(cnt, KW_COUNT_CAP)
        out.append({
            "kind": "keyword",
            "label": label,
            "weight": round(w, 2),
            "count": cnt,
            "counted": capped,
            "contribution": round(w * capped, 2),
        })
    return out


def _scan_entities(findings: Iterable[dict]) -> list[dict]:
    """findings → entity_type 별 기여도 dict 리스트."""
    counts: dict[str, int] = {}
    for f in findings or []:
        et = f.get("entity_type")
        if not et:
            continue
        counts[et] = counts.get(et, 0) + 1
    out: list[dict] = []
    for et, cnt in counts.items():
        w = ENTITY_WEIGHTS.get(et, DEFAULT_ENTITY_WEIGHT)
        out.append({
            "kind": "entity",
            "label": et,
            "weight": round(w, 2),
            "count": cnt,
            "contribution": round(w * cnt, 2),
        })
    return out


def classify(findings: Iterable[dict] | None, text: str | None) -> dict:
    """findings + 본문 → 분류 결과.

    Args:
        findings: /api/analyze 가 만든 findings 리스트 (entity_type 키 필수).
        text:     본문 — 등급 키워드 스캔에 사용. None/"" 이면 키워드 스캔 생략.

    Returns: {
        grade:        "C" | "S" | "O"
        grade_label:  "위험 (Critical)" 등
        score:        누적 점수
        confidence:   0.55..0.95
        thresholds:   {"C": 5.0, "S": 2.0}
        reasons:      기여 큰 순 정렬된 entity/keyword 항목 (최대 20)
        version:      "rule-v1"
        text_chars:   스캔 대상 본문 길이
    }
    """
    entity_reasons = _scan_entities(findings or [])
    keyword_reasons = _scan_keywords(text or "")
    reasons = entity_reasons + keyword_reasons
    reasons.sort(key=lambda r: -r["contribution"])

    score = sum(r["contribution"] for r in reasons)
    grade, grade_label, margin = _grade_for_score(score)

    return {
        "grade":       grade,
        "grade_label": grade_label,
        "score":       round(score, 2),
        "confidence":  _confidence(margin),
        "thresholds":  {"C": C_THRESHOLD, "S": S_THRESHOLD},
        "reasons":     reasons[:20],
        "version":     CLASSIFIER_VERSION,
        "text_chars":  len(text or ""),
    }


def gap(ai_grade: str, user_grade: str) -> int:
    """SPEC §10.1 — |aiRank - userRank|. 0=일치, 1=인접 불일치, 2=반대."""
    return abs(GRADE_RANK.get(ai_grade, 0) - GRADE_RANK.get(user_grade, 0))


def sample_weight(g: int) -> float:
    """SPEC §10.2 — 학습 시 가중치. 1 + 1.5 * gap."""
    return 1.0 + 1.5 * g