model/classification/src/classifier_simple.py

"""
베이스라인 분류기: TF-IDF + Logistic Regression
=================================================
담당: 경이
역할: 라벨 데이터(notice_sample_v3.csv + notices_labeled_v2.csv)로 학습한
      6-class 텍스트 분류기.
      GPU 불필요. CPU에서 수십 ms 이내 추론 가능.
      KcELECTRA 파인튜닝과 성능 비교할 베이스라인.

사용법:
    python classifier_simple.py          # 학습 + 저장
    python classifier_simple.py --eval   # 저장된 모델 평가
"""

import pickle
import argparse
from pathlib import Path

import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.metrics import classification_report, confusion_matrix

# 경로 기준: model/classification/
_BASE = Path(__file__).parent.parent
DATA_CSV    = _BASE / "data" / "notice_sample_v3.csv"
LEGACY_CSV  = _BASE / "data" / "notices_labeled_v2.csv"   # 기존 라벨 데이터
SPLIT_CSV   = _BASE / "data" / "split_v1.csv"
MODEL_PKL   = _BASE / "checkpoints" / "simple_tfidf_logreg.pkl"

LABELS = ["일정", "준비물", "제출", "비용", "건강·안전", "기타"]

# 기존 CSV는 컬럼명이 'original_text'이므로 통일 처리
_TEXT_COLS = ["text", "original_text", "sentence"]


def _normalize_df(df: pd.DataFrame) -> pd.DataFrame:
    """컬럼명이 다른 여러 데이터 소스를 text/category 형식으로 통일."""
    for col in _TEXT_COLS:
        if col in df.columns and "text" not in df.columns:
            df = df.rename(columns={col: "text"})
            break
    return df


# ─────────────────────────────────────────────────────────────
# 데이터 로드
# ─────────────────────────────────────────────────────────────
def load_data(split: str = "train") -> tuple[list[str], list[str]]:
    """split_v1.csv → 없으면 원본 CSV (+ legacy CSV 병합)에서 텍스트·라벨 로드.

    split: "train" | "val" | "test" | "all"
    """
    if SPLIT_CSV.exists():
        df = pd.read_csv(SPLIT_CSV)
        df = _normalize_df(df)
        if split != "all":
            df = df[df["split"] == split]
    else:
        frames = [pd.read_csv(DATA_CSV)]
        if LEGACY_CSV.exists():
            legacy = pd.read_csv(LEGACY_CSV)
            legacy = _normalize_df(legacy)
            frames.append(legacy)
        df = pd.concat(frames, ignore_index=True)
        df = _normalize_df(df)

    df = df.dropna(subset=["text", "category"])
    df = df[df["category"].isin(LABELS)]
    return df["text"].tolist(), df["category"].tolist()


# ─────────────────────────────────────────────────────────────
# 파이프라인 정의
# ─────────────────────────────────────────────────────────────
def build_pipeline() -> Pipeline:
    """TF-IDF + Logistic Regression 파이프라인.

    TF-IDF 파라미터:
      - analyzer="char_wb": 한국어는 글자 단위 n-gram이 어절 단위보다 OOV((Out-of-Vocabulary)에 강함 : 단어 분리 없이 글자 단위 n-gram 사용 → 형태소 분석기 불필요, 미등록어(OOV)에 강함
      - 글자 n-gram은 조사/어미가 달라도 어근 조각이 겹치기 때문에 자연스럽게 대응
      - 글자 조각(n-gram)으로 쪼개면 처음 보는 표현도 익숙한 조각들로 분해되어 의미를 파악할 수 있다.
      - ngram_range=(2, 4): 2~4글자 조합으로 어미/조사 같은 형태소 정보 간접 포착
      - max_features=30000: 메모리·속도 균형
      - sublinear_tf=True: 빈도를 log(1+tf)로 변환해 특정 단어의 과도한 영향 억제
    LogReg 파라미터:
      - C=1.0: 기본 정규화 (오버피팅 방지) / C=1.0 — 정규화 강도
      - 정규화의 필요성: 모델이 학습 데이터에 너무 딱 맞게 학습되면 새로운 데이터에서 성능이 떨어집니다 (오버피팅).
      - C는 정규화의 반대 개념입니다.
      - C 값이 작을수록  →  정규화 강함  →  가중치를 강하게 억제  →  단순한 모델 
      - C 값이 클수록   →  정규화 약함  →  가중치를 자유롭게 키움  →  복잡한 모델
      - max_iter=1000: 수렴 보장
      - 초기 가중치 → 예측 → 오차 계산 → 가중치 조정 → 예측 → 오차 계산 → ...
        (1번째 iter)                        (2번째 iter)
      - class_weight="balanced": 클래스 불균형 대응
      - solver="lbfgs": 다중 클래스에 적합한 최적화 알고리즘
    """
    return Pipeline([
        ("tfidf", TfidfVectorizer(
            analyzer="char_wb",
            ngram_range=(2, 4),
            max_features=30_000,
            sublinear_tf=True,
        )),
        ("clf", LogisticRegression(
            C=1.0,
            max_iter=1000,
            class_weight="balanced",
            random_state=42,
            solver="lbfgs",
        )),
    ])


# ─────────────────────────────────────────────────────────────
# 학습
# ─────────────────────────────────────────────────────────────
def train() -> Pipeline:
    texts, labels = load_data("train")
    if not texts:
        texts, labels = load_data("all")

    pipe = build_pipeline()
    pipe.fit(texts, labels)

    MODEL_PKL.parent.mkdir(parents=True, exist_ok=True)
    with open(MODEL_PKL, "wb") as f:
        pickle.dump(pipe, f)
    print(f"[simple] 모델 저장 완료: {MODEL_PKL}")
    print(f"[simple] 학습 데이터 수: {len(texts)}개")
    return pipe


# ─────────────────────────────────────────────────────────────
# 로드 (캐시)
# ─────────────────────────────────────────────────────────────
_pipeline: Pipeline | None = None


def load_pipeline() -> Pipeline:
    global _pipeline
    if _pipeline is not None:
        return _pipeline

    if not MODEL_PKL.exists():
        print("[simple] 저장된 모델 없음 → 학습 시작")
        _pipeline = train()
    else:
        with open(MODEL_PKL, "rb") as f:
            _pipeline = pickle.load(f)
    return _pipeline


# ─────────────────────────────────────────────────────────────
# 추론
# ─────────────────────────────────────────────────────────────
def predict_simple(text: str) -> dict:
    """텍스트 → 카테고리 + 신뢰도(각 클래스 확률).

    반환:
        {
            "category": str,          # 예측 라벨
            "confidence": float,      # 예측 클래스 확률
            "probs": dict[str, float] # 전체 클래스 확률 (explain 용)
        }
    """
    pipe = load_pipeline()
    proba = pipe.predict_proba([text])[0]
    classes = pipe.classes_

    idx = proba.argmax()
    return {
        "category":   classes[idx],
        "confidence": float(proba[idx]),
        "probs":      {c: float(p) for c, p in zip(classes, proba)},
    }


# ─────────────────────────────────────────────────────────────
# 평가
# ─────────────────────────────────────────────────────────────
def evaluate(split: str = "test") -> dict:
    texts, true_labels = load_data(split)
    if not texts:
        texts, true_labels = load_data("all")

    pipe = load_pipeline()
    pred_labels = pipe.predict(texts)

    report = classification_report(
        true_labels, pred_labels,
        labels=LABELS,
        output_dict=True,
        zero_division=0,
    )
    cm = confusion_matrix(true_labels, pred_labels, labels=LABELS)

    print("\n[simple] 분류 리포트")
    print(classification_report(true_labels, pred_labels, labels=LABELS, zero_division=0))
    print("[simple] Confusion Matrix")
    print(cm)

    return {"report": report, "confusion_matrix": cm.tolist(), "model": "simple"}


# ─────────────────────────────────────────────────────────────
# CLI
# ─────────────────────────────────────────────────────────────
if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--eval", action="store_true", help="저장된 모델 평가")
    args = parser.parse_args()

    if args.eval:
        evaluate()
    else:
        train()
        evaluate("test")