C.E.L_Slide_test2/tests/pipeline/code/keyword_normalizer.py

"""공통 키워드 정규화 모듈.

Step 4, 5, 7 등 여러 파이프라인이 공유하는 정규화 로직을 한 파일로 집중.
이 모듈은 **AI 호출을 수행하지 않음** — 순수 결정론적 규칙만.

제공 기능:
  - PRE_COLLAPSE regex (디지털 전환(DX) 계열 collapse)
  - 괄호 확장 (A(B)C → AC + B)
  - phrase_variants 치환
  - Kiwi user_dict 토큰화
  - 허용 태그 필터 (NNG/NNP/SL/SN + 1글자/순수숫자 제외)

사용 pipeline:
  - pipeline_04_normalize.py (정규화 + 저장)
  - pipeline_07_auto_anchor_candidates.py (source_text 기반 후보)
  (미래) 기타 step

설계 제약:
  - 상수/함수는 재진입 가능해야 함 (counter 를 None 허용)
  - Kiwi 객체는 build_kiwi() 로 생성 (global state 금지)
  - 모든 함수는 deterministic (random seed 없음)
"""
import re
from collections import Counter
from pathlib import Path
import yaml
from kiwipiepy import Kiwi

# ============================================================
# 상수
# ============================================================

# Kiwi user_dict — compound (SL)
USER_DICT_SL = ['S/W', 'H/W', '2D', '3D', 'DX', 'BIM', 'As-is', 'To-Be']
# Kiwi user_dict — phrase canonical (NNG)
USER_DICT_NNG = ['결과혁신', '과정혁신', '필수조건', '의사소통', '시행착오']

# Kiwi 허용 태그
ALLOWED_TAGS = {'NNG', 'NNP', 'SL', 'SN'}

# PRE_COLLAPSE: "디지털 전환(DX...)" / "DX(디지털전환)" / "DX(Digital Transformation)" / "DX(DX...)" → DX
# 순서 중요: 긴 → 짧은, 원본 → 치환 부산물.
PRE_COLLAPSE = [
    (
        re.compile(
            r'디지털\s*전환\s*\(\s*DX(?:\s*[,/]\s*Digital\s+Transformation)?\s*\)',
            re.IGNORECASE,
        ),
        'DX',
        'digital_transition_with_dx',
    ),
    (
        re.compile(r'DX\s*\(\s*디지털\s*전환\s*\)', re.IGNORECASE),
        'DX',
        'dx_with_digital_transition',
    ),
    (
        re.compile(r'DX\s*\(\s*Digital\s+Transformation\s*\)', re.IGNORECASE),
        'DX',
        'dx_with_english_fullname',
    ),
    (
        re.compile(
            r'DX\s*\(\s*DX(?:\s*[,/]\s*Digital\s+Transformation)?\s*\)',
            re.IGNORECASE,
        ),
        'DX',
        'dx_duplicate',
    ),
    # 5. SW (단독 word) → S/W — 정규화 누락 보정 (word boundary 중요: SWOT 등 보호)
    (
        re.compile(r'(?<![A-Za-z])SW(?![A-Za-z])'),
        'S/W',
        'sw_to_sw_slash',
    ),
]

# 괄호 확장
MAX_PAREN_LEN = 40
PAREN_PATTERN = re.compile(r'\(([^()]{1,' + str(MAX_PAREN_LEN) + r'})\)')
# 숫자/기호-only 괄호 내용 배제 (예: (1), (2), (※))
PAREN_SKIP_PATTERN = re.compile(r'[\d\s\-.,:;※]+')
PAREN_SAMPLE_MAX = 20

# ============================================================
# 필터 규칙 (사용자 확정 — 제거 후보 3그룹 + 제품 whitelist)
# ============================================================

# 의문사 (대소문자 무관 제거)
INTERROGATIVES = {
    'how', 'what', 'when', 'why', 'who', 'where', 'which',
}

# 완전 소문자 일반 기능어 제거
SOFT_STOPWORDS = {
    'or', 'for', 'and', 'the', 'of', 'to', 'at', 'by', 'in', 'on', 'up',
    'as', 'a', 'an', 'with', 'from', 'into',
    'is', 'be', 'are', 'was', 'were', 'been', 'being',
    'it', 'we', 'he', 'she', 'they', 'i', 'you',
    'also', 'only', 'all', 'but', 'not', 'so', 'too', 'very',
    'this', 'that', 'these', 'those', 'etc',
}

FILTER_NUMBER_COMMA_PATTERN = re.compile(r'^[\d,]+$')
FILTER_SHORT_LOWERCASE_PATTERN = re.compile(r'^[a-z]{1,3}$')
FILTER_UPPER_ABBREV_PATTERN = re.compile(r'^[A-Z][A-Z0-9]+$')

# 제품명/고유명사 수동 whitelist (사용자 승인, 자동 규칙 대신 수동)
PRODUCT_WHITELIST = {
    # 상용 S/W 제품명
    'Revit', 'AutoCAD', 'Autocad', 'AutoCad', 'Auto CAD',
    'SketchUp', 'Sketchup',
    'Navisworks', 'Naviswork',
    'Infraworks',
    'Rhino',
    'ArchiCAD',
    'ArcGIS', 'QGIS',
    'Blender',
    'WatchBIM',
    'Domainer',
    'BCMF',
    # 회사/벤더명
    'Bentley', 'Bently',
    'Autodesk', 'AutoDesk',
    'Nemetschek',
    'ESRI',
    # 국제기구 약어 (대문자 약어 규칙으로 자동 보호되지만 안전하게 명시)
    'ADB', 'IBRD',
}

# 화이트리스트 (user_dict + 제품명) — 필터 규칙에 걸려도 절대 제거 안 함
FILTER_WHITELIST = set(USER_DICT_SL) | set(USER_DICT_NNG) | PRODUCT_WHITELIST


def should_filter_token(token):
    """필터 규칙 판정. (제거_여부, 이유) 반환.

    우선순위:
      0. 화이트리스트 → 절대 제거 안 함
      1. 전부 대문자 약어 (IT/AI/ERP 등) → 보존
      2. 의문사 (대소문자 무관) → 제거
      3. 숫자+쉼표 → 제거
      4. 완전 소문자 stopword → 제거
      5. 소문자 1-3자 영어 → 제거
    """
    # 0. 화이트리스트 우선
    if token in FILTER_WHITELIST:
        return False, None
    # 1. 의문사 (대소문자 무관) — 대문자 약어 검사보다 먼저 (HOW/WHEN 등)
    if token.lower() in INTERROGATIVES:
        return True, 'interrogative'
    # 2. 대문자 약어 보존 (IT, AI, ERP, CCTV 등)
    if FILTER_UPPER_ABBREV_PATTERN.fullmatch(token):
        return False, None
    # 3. 숫자+쉼표 패턴
    if FILTER_NUMBER_COMMA_PATTERN.fullmatch(token):
        return True, 'number_pattern'
    # 4. 완전 소문자 stopword
    if token in SOFT_STOPWORDS:
        return True, 'stopword'
    # 5. 소문자 1-3자 영어
    if FILTER_SHORT_LOWERCASE_PATTERN.fullmatch(token):
        return True, 'short_lowercase'
    return False, None


# ============================================================
# 함수
# ============================================================

def load_phrase_variants(synonyms_path):
    """synonyms.yaml 에서 phrase_variants 섹션 로드."""
    data = yaml.safe_load(Path(synonyms_path).read_text(encoding='utf-8'))
    pv = data.get('phrase_variants')
    if not pv:
        raise RuntimeError(f"{synonyms_path} 에 phrase_variants 섹션이 없음.")
    return pv


def build_substitutions(phrase_variants):
    """(variant, canonical) 리스트. canonical 자기 자신 재치환 금지. 긴 variant 먼저."""
    subs = []
    seen = set()
    for canonical, variants in phrase_variants.items():
        for v in (variants or []):
            if v == canonical:
                continue
            if (v, canonical) in seen:
                continue
            seen.add((v, canonical))
            subs.append((v, canonical))
    subs.sort(key=lambda x: (-len(x[0]), x[0]))
    return subs


def expand_parentheses(text, paren_counter=None, paren_samples=None):
    """A(B)C → 'AC B' 로 펼침.

    - 괄호 밖: 빈 문자열로 제거 (조사 결합 유지)
    - 괄호 안: 뒤에 공백 연결
    - 길이 > MAX_PAREN_LEN 은 펼치지 않음
    - 숫자/기호-only 괄호 내용 제외 ((1), (※) 등)

    paren_counter / paren_samples 는 None 허용 — 카운팅 안 할 때는 None.
    """
    contents = PAREN_PATTERN.findall(text)
    if not contents:
        return text
    meaningful = []
    for c in contents:
        stripped = c.strip()
        if not stripped:
            continue
        if PAREN_SKIP_PATTERN.fullmatch(stripped):
            continue
        meaningful.append(stripped)
    if not meaningful:
        return text
    before = text
    text_out = PAREN_PATTERN.sub('', text)
    text_out = re.sub(r'\s+', ' ', text_out).strip()
    for c in meaningful:
        text_out = text_out + ' ' + c
        if paren_counter is not None:
            paren_counter['expansions'] += 1
    if paren_samples is not None and before != text_out and len(paren_samples) < PAREN_SAMPLE_MAX:
        paren_samples.append({'before': before, 'after': text_out})
    return text_out


def apply_substitutions(text, subs, replacement_counter, pre_collapse_counter,
                        paren_counter=None, paren_samples=None):
    """PRE_COLLAPSE → 괄호 확장 → phrase_variants 순서 적용."""
    # 1) PRE_COLLAPSE (regex)
    for pattern, replacement, rule_id in PRE_COLLAPSE:
        new_text, n = pattern.subn(replacement, text)
        if n > 0:
            if pre_collapse_counter is not None:
                pre_collapse_counter[rule_id] += n
            text = new_text
    # 1.5) 괄호 확장
    text = expand_parentheses(text, paren_counter, paren_samples)
    # 2) phrase_variants (literal)
    for variant, canonical in subs:
        new_text, n = re.subn(re.escape(variant), canonical, text)
        if n > 0:
            if replacement_counter is not None:
                replacement_counter[canonical] += n
            text = new_text
    return text


def build_kiwi():
    """user_dict 포함 Kiwi 객체 생성."""
    kiwi = Kiwi()
    for w in USER_DICT_SL:
        kiwi.add_user_word(w, 'SL', 9.0)
    for w in USER_DICT_NNG:
        kiwi.add_user_word(w, 'NNG', 9.0)
    return kiwi


def extract_tokens(kiwi, text, removal_log=None):
    """Kiwi tokenize + 허용 태그/1글자/순수숫자 필터 + 사용자 확정 필터 규칙.

    removal_log: dict[str, list[str]] 또는 None — 제거된 토큰 + 이유 기록 (추적용).
      예: {'interrogative': ['How', 'WHEN'], 'stopword': ['the', 'to'], ...}
    """
    out = []
    for tok in kiwi.tokenize(text):
        if tok.tag not in ALLOWED_TAGS:
            continue
        f = tok.form
        if len(f) < 2:
            continue
        if re.fullmatch(r'\d+', f):
            continue
        # 사용자 확정 필터 (stopword / number / short_lowercase / interrogative)
        remove, reason = should_filter_token(f)
        if remove:
            if removal_log is not None:
                removal_log.setdefault(reason, []).append(f)
            continue
        out.append(f)
    return out


def normalize_and_tokenize(text, subs, kiwi,
                           replacement_counter=None, pre_collapse_counter=None,
                           paren_counter=None, paren_samples=None):
    """text → (normalized_text, tokens). 복합 파이프라인 단일 진입점."""
    normalized = apply_substitutions(
        text, subs, replacement_counter, pre_collapse_counter,
        paren_counter, paren_samples,
    )
    tokens = extract_tokens(kiwi, normalized)
    return normalized, tokens