CornerStone11th/Intent_Classification

1# app_gradio.py
2import json
3from pathlib import Path
4from datetime import datetime
5import gradio as gr
6import pandas as pd
7import matplotlib.pyplot as plt
8
9from llm_helper import (
10    extract_json, prune_extra_keys, ensure_required_fields,
11    normalize_result, rebias_intent, validate_json
12)
13import platform
14from matplotlib import font_manager as fm
15import matplotlib
16
17def setup_korean_font():
18    """시스템별로 사용 가능한 한글 폰트를 자동 선택."""
19    try:
20        system = platform.system()
21        candidates = []
22        if system == "Windows":
23            candidates = ["Malgun Gothic", "맑은 고딕"]
24        elif system == "Darwin":  # macOS
25            candidates = ["AppleGothic"]
26        else:  # Linux 등
27            candidates = ["NanumGothic", "Noto Sans CJK KR", "Noto Sans KR"]
28
29        available = {f.name for f in fm.fontManager.ttflist}
30        for name in candidates:
31            if name in available:
32                matplotlib.rcParams["font.family"] = name
33                break
34        # 마이너스 기호 깨짐 방지
35        matplotlib.rcParams["axes.unicode_minus"] = False
36    except Exception:
37        # 폰트가 하나도 없으면 기본 폰트(영문)로 진행
38        pass
39
40# <<< 파일 상단 어딘가에서 1회 호출 >>>
41setup_korean_font()
42# -------------------------------------------------------------------
43# 공용 경로/유틸
44# -------------------------------------------------------------------
45LOG_PATH = Path("data/logs.jsonl")           # ← Path로 고정
46LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
47
48def now_iso() -> str:
49    return datetime.now().astimezone().isoformat()
50
51# -------------------------------------------------------------------
52# 로그 → DF 로딩 & 차트 함수
53# -------------------------------------------------------------------
54def load_logs_df() -> pd.DataFrame:
55    if not LOG_PATH.exists():
56        return pd.DataFrame(columns=["ts","intent","confidence","product","quantity","valid","date"])
57    rows = []
58    with LOG_PATH.open(encoding="utf-8") as f:
59        for line in f:
60            line = line.strip()
61            if not line:
62                continue
63            d = json.loads(line)
64            res = d.get("result", {}) or {}
65            fields = res.get("fields", {}) or {}
66            rows.append({
67                "ts": d.get("ts"),
68                "intent": res.get("intent"),
69                "confidence": res.get("confidence"),
70                "product": fields.get("product"),
71                "quantity": fields.get("quantity"),
72                "valid": d.get("valid", True),
73            })
74    df = pd.DataFrame(rows)
75    if df.empty:
76        df["date"] = pd.to_datetime([]).date
77        return df
78    df["date"] = pd.to_datetime(df["ts"]).dt.date
79    # 수량은 숫자로 강제 (None/문자 → NaN → 0)
80    df["quantity"] = pd.to_numeric(df["quantity"], errors="coerce").fillna(0).astype(int)
81    return df
82
83def plot_intent_pie(df: pd.DataFrame):
84    fig = plt.figure()
85    if df.empty:
86        plt.title("데이터가 아직 없어요")
87        return fig
88    counts = df["intent"].value_counts()
89    # 색상 지정하지 않음(기본 팔레트 사용)
90    plt.pie(counts.values, labels=counts.index, autopct="%1.1f%%", startangle=90)
91    plt.title("의도 분포 (원형)")
92    plt.tight_layout()
93    return fig
94
95def plot_daily_trend(df: pd.DataFrame, window_days: int | None = None):
96    fig = plt.figure()
97    if df.empty:
98        plt.title("데이터가 아직 없어요")
99        return fig
100    dff = df.copy()
101    if window_days:
102        cutoff = pd.Timestamp.today().date() - pd.Timedelta(days=window_days-1)
103        dff = dff[dff["date"] >= cutoff]
104    if dff.empty:
105        plt.title("선택 기간에 데이터가 없어요")
106        return fig
107    piv = dff.pivot_table(index="date", columns="intent", values="ts", aggfunc="count").fillna(0.0)
108    piv.sort_index(inplace=True)
109    for col in piv.columns:
110        plt.plot(piv.index.astype(str), piv[col], marker="o", label=col)
111    plt.title("일자별 의도 추세")
112    plt.xlabel("날짜"); plt.ylabel("건수")
113    plt.xticks(rotation=45); plt.legend()
114    plt.tight_layout()
115    return fig
116
117def plot_top_products(df: pd.DataFrame, topn: int = 10):
118    fig = plt.figure()
119    if df.empty:
120        plt.title("데이터가 아직 없어요")
121        return fig
122    dfo = df[(df["intent"] == "order") & df["product"].notna()]
123    if dfo.empty:
124        plt.title("주문 데이터가 아직 없어요")
125        return fig
126    s = dfo.groupby("product")["quantity"].sum().sort_values(ascending=False).head(topn)
127    plt.bar(s.index, s.values)
128    plt.title("주문 수량 TOP N")
129    plt.xticks(rotation=45, ha="right")
130    plt.tight_layout()
131    return fig
132
133def refresh_dashboard(period: str):
134    df = load_logs_df()
135    window = None if period == "전체" else int(period)
136    return (plot_intent_pie(df),
137            plot_daily_trend(df, window),
138            plot_top_products(df))
139
140# -------------------------------------------------------------------
141# 분류 + 로그 저장
142# -------------------------------------------------------------------
143def classify_and_log(text: str, save_log: bool):
144    if not text.strip():
145        return "입력 없음", "", False, ""
146    j = extract_json(text)
147    if not j:
148        return "LLM JSON 실패", "", False, ""
149    j = prune_extra_keys(j)
150    j = ensure_required_fields(j)
151    j = normalize_result(j)
152    j = rebias_intent(j, text)
153    ok, errs = validate_json(j)
154    if save_log:
155        LOG_PATH.write_text("", encoding="utf-8") if not LOG_PATH.exists() else None
156        with LOG_PATH.open("a", encoding="utf-8") as f:
157            f.write(json.dumps({"ts": now_iso(), "input": text, "result": j,
158                                "valid": ok, "errors": errs}, ensure_ascii=False) + "\n")
159    label = f"intent={j['intent']} | conf={j['confidence']:.2f} | valid={ok}"
160    details = json.dumps(j, ensure_ascii=False, indent=2)
161    errtxt = "\n".join(errs) if not ok else ""
162    return label, details, ok, errtxt
163
164# -------------------------------------------------------------------
165# Gradio UI
166# -------------------------------------------------------------------
167with gr.Blocks(title="의견·의도 분류기") as demo:
168    gr.Markdown("### 의견 → 의도 분류 + JSON 추출 (로컬 LLM)")
169
170    with gr.Tabs():
171        # --- 분류 탭 ---
172        with gr.Tab("분류"):
173            with gr.Row():
174                inp = gr.Textbox(label="의견 입력", lines=3, placeholder="예) 매일우유 1L 다섯 개 내일 부탁드립니다.")
175            with gr.Row():
176                save = gr.Checkbox(label="로그 저장(data/logs.jsonl)", value=True)
177                btn = gr.Button("분류 실행")
178            out_label = gr.Label(label="요약")
179            out_json = gr.Code(label="JSON", language="json")
180            out_ok = gr.Checkbox(label="VALID (스키마 통과)", interactive=False)
181            out_errs = gr.Textbox(label="검증 에러", lines=3)
182            btn.click(classify_and_log, [inp, save], [out_label, out_json, out_ok, out_errs])
183
184        # --- 대시보드 탭 ---
185        with gr.Tab("대시보드"):
186            gr.Markdown("#### 로그 기반 시각화 (data/logs.jsonl)")
187            with gr.Row():
188                period = gr.Dropdown(choices=["7","30","전체"], value="7", label="기간(일)")
189                btn_refresh = gr.Button("새로고침")
190            plot1 = gr.Plot(label="의도 분포")
191            plot2 = gr.Plot(label="일자별 의도 추세")
192            plot3 = gr.Plot(label="주문 수량 TOP N")
193            btn_refresh.click(refresh_dashboard, [period], [plot1, plot2, plot3])
194
195            # --- 자동 갱신(5초마다) ---
196            try:
197                timer = gr.Timer(5.0)  # 5초 간격
198                timer.tick(refresh_dashboard, inputs=[period], outputs=[plot1, plot2, plot3])
199            except Exception:
200                # Gradio 버전에 Timer가 없으면, 페이지 로드시 주기적 갱신으로 대체
201                demo.load(refresh_dashboard, inputs=[period], outputs=[plot1, plot2, plot3], every=5)
202
203if __name__ == "__main__":
204    demo.queue()
205    demo.launch(server_port=7860, inbrowser=True)
206