zioinfo-mail/itsm/core/nlu.py

"""
GUARDiA NLU Engine — 순수 규칙 기반 자연어 이해 (외부 API 없음).

처리 흐름:
  1. 정규식 패턴 매칭으로 인텐트(의도) 분류
  2. 키워드/패턴으로 엔티티 추출
  3. (sr_id, engineer, status, sr_type, inst_code, priority, keyword)
"""
import re
from dataclasses import dataclass, field
from typing import Optional


# ── 인텐트 정의 ───────────────────────────────────────────────────────────────

INTENT_PATTERNS: list[tuple[str, list[str]]] = [
    # 우선순위 높은 인텐트부터

    # SR 작업 시뮬레이션 실행
    ("SIMULATE", [
        r"(시뮬레이션|simulate|작업\s*실행|자동\s*실행|AI\s*실행)",
        r"(실행|run).*(SR|작업)",
    ]),

    # 담당자 배정·변경
    ("ASSIGN_ENGINEER", [
        r"담당자.*(변경|바꿔|배정|교체|지정)",
        r"(배정|assign).*(engineer|엔지니어)",
        r"engineer\d+.*(맡겨|담당|배정|지정)",
        r"(자동\s*배정|auto\s*assign)",
    ]),

    # 승인
    ("APPROVE_SR", [
        r"(승인|approve)\s*(해줘|처리|완료|부탁)",
        r"결재\s*(해줘|처리)",
    ]),

    # 반려
    ("REJECT_SR", [
        r"(반려|거절|reject)\s*(해줘|처리|할게)",
    ]),

    # KB 검색
    ("SEARCH_KB", [
        r"(KB|기술\s*문서|지식|매뉴얼).*(검색|찾아|조회)",
        r"(검색|찾아줘|알려줘).*(오류|error|fault|장애|문제)",
        r"(해결\s*방법|원인|솔루션).*(알려줘|찾아줘)",
        r"(oom|outofmemory|connection\s*pool|ssl|nginx|502|undo|gc\s*overhead|디스크)",
    ]),

    # 워크로드 조회
    ("QUERY_WORKLOAD", [
        r"(워크로드|workload|작업\s*현황|담당\s*현황)",
        r"engineer.*(몇\s*건|현황|상황|바빠)",
        r"(엔지니어|engineer).*(상황|현황|보여줘)",
    ]),

    # SR 상세 조회
    ("QUERY_SR_DETAIL", [
        r"SR-\d{8}-[A-Z0-9]{6}.*(상태|정보|내용|어떻게|확인)",
        r"(상태|정보|현황).*(알려줘|보여줘|확인).*(SR|서비스)",
    ]),

    # SR 목록 조회
    ("QUERY_SR_LIST", [
        r"(목록|리스트|list|조회|보여줘).*(SR|서비스|요청)",
        r"(SR|서비스\s*요청).*(목록|있어|있나|몇\s*건|현황|리스트)",
        r"(승인\s*대기|진행\s*중|완료|긴급|접수).*(SR|서비스|있어|있나|보여)",
        r"(어떤|무슨)\s*SR",
    ]),

    # 통계/요약
    ("QUERY_STATS", [
        r"(통계|요약|summary|stats|현황\s*요약|전체\s*현황)",
        r"(총|전체).*(SR|건수|몇\s*건)",
        r"(오늘|이번\s*주).*(SR|처리|완료)",
    ]),
]


# ── 엔티티 추출 ───────────────────────────────────────────────────────────────

_SR_ID_RE       = re.compile(r'SR-\d{8}-[A-Z0-9]{6}', re.IGNORECASE)
_ENG_USER_RE    = re.compile(r'engineer\d+', re.IGNORECASE)
_PRIORITY_MAP   = {"긴급": "CRITICAL", "높음": "HIGH", "보통": "MEDIUM", "낮음": "LOW",
                   "critical": "CRITICAL", "high": "HIGH", "medium": "MEDIUM", "low": "LOW"}
_STATUS_MAP     = {
    "승인": "APPROVED", "반려": "REJECTED", "완료": "COMPLETED",
    "진행": "IN_PROGRESS", "접수": "RECEIVED", "대기": "PENDING_APPROVAL",
}
_TYPE_MAP       = {
    "배포": "DEPLOY", "deploy": "DEPLOY",
    "재기동": "RESTART", "restart": "RESTART",
    "로그": "LOG", "log": "LOG",
    "문의": "INQUIRY", "inquiry": "INQUIRY",
}
_INST_CODES     = ["MOF", "MOIS", "MSS"]
_ENG_ALIAS      = {"김엔지니어": "engineer1", "이엔지니어": "engineer2"}


@dataclass
class ParseResult:
    intent:    str              = "UNKNOWN"
    sr_id:     Optional[str]   = None
    engineer:  Optional[str]   = None
    status:    Optional[str]   = None
    sr_type:   Optional[str]   = None
    inst_code: Optional[str]   = None
    priority:  Optional[str]   = None
    keyword:   Optional[str]   = None
    raw_text:  str              = ""


def parse(text: str) -> ParseResult:
    """자연어 텍스트를 분석해 인텐트 + 엔티티를 반환."""
    r     = ParseResult(raw_text=text)
    lower = text.lower()

    # ── 인텐트 분류 ──────────────────────────────────────
    for intent_name, patterns in INTENT_PATTERNS:
        for pat in patterns:
            if re.search(pat, text, re.IGNORECASE):
                r.intent = intent_name
                break
        if r.intent != "UNKNOWN":
            break

    # ── 엔티티 추출 ──────────────────────────────────────
    # SR ID
    m = _SR_ID_RE.search(text)
    if m:
        r.sr_id = m.group().upper()

    # 엔지니어 (username 또는 한글 별칭)
    m = _ENG_USER_RE.search(text)
    if m:
        r.engineer = m.group().lower()
    else:
        for alias, uname in _ENG_ALIAS.items():
            if alias in text:
                r.engineer = uname
                break

    # 상태
    for k, v in _STATUS_MAP.items():
        if k in lower:
            r.status = v
            break

    # SR 유형
    for k, v in _TYPE_MAP.items():
        if k in lower:
            r.sr_type = v
            break

    # 기관 코드
    for inst in _INST_CODES:
        if inst in text.upper():
            r.inst_code = inst
            break

    # 우선순위
    for k, v in _PRIORITY_MAP.items():
        if k in lower:
            r.priority = v
            break

    # 키워드 (KB 검색용) — 따옴표 안 또는 '검색:' 뒤
    qm = re.search(r'["\'](.+?)["\']', text)
    if qm:
        r.keyword = qm.group(1)
    else:
        # fallback: 큰 단어 추출
        kb_kw = re.search(
            r'(oom|outofmemory|connection\s*pool|ssl|nginx|502|undo|gc overhead'
            r'|디스크|메모리|커넥션|인증서|tomcat|oracle|ora-\d+)',
            text, re.IGNORECASE
        )
        if kb_kw:
            r.keyword = kb_kw.group(1)

    return r