Update app.py

app.py

@@ -1,107 +1,57 @@
-# app.py
-# 정치 편향 방지 데모 (라벨링은 내부 활용, 화면은 '팩트 기반 중립 재구성' 중심)
-# - 데이터: HF jacobvs/PoliticalTweets
-# - 모델: TF-IDF + SGDClassifier (증분 학습)
-# - 외부: (선택) 네이버 뉴스 API로 기사 검색, (선택) OpenAI로 사실 기반 재구성
+# 정치 편향 방지 · 사실 기반 중립 재구성 (BERT 분류기 버전)
+# - 분류: bucketresearch/politicalBiasBERT (left/center/right)
+# - 재구성: OpenAI로 사실 중심 요약/재작성
+# - 옵션: 네이버 뉴스 검색
 
 import os
-import json
-import time
-from typing import List, Dict, Tuple, Optional
+from typing import List, Dict, Tuple
 
-import requests
 import streamlit as st
-
-# ML
-import joblib
-from sklearn.feature_extraction.text import TfidfVectorizer
-from sklearn.linear_model import SGDClassifier
-from sklearn.metrics import classification_report, accuracy_score
-
-# HF datasets
-from datasets import load_dataset
+import requests
+import torch
+from transformers import AutoTokenizer, AutoModelForSequenceClassification, pipeline
 
 # =========================
-# App Config
+# Config
 # =========================
-APP_TITLE = "정치 편향 방지 · 사실 기반 중립 재구성 (Streamlit)"
-MODEL_PATH = "incremental_model.pkl"
-VECT_PATH  = "tfidf_vectorizer.pkl"
-
+APP_TITLE = "정치 편향 분석(BERT) · 사실 기반 중립 재구성"
 OPENAI_API_KEY = os.getenv("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")              # 선택: 사실 기반 재구성에 사용
 NAVER_ID       = os.getenv("I_8koTJh3R5l4wLurQbG")             # 선택: 네이버 뉴스 검색
 NAVER_SECRET   = os.getenv("W5oWYlAgur")
 
+MODEL_ID = "bucketresearch/politicalBiasBERT"
+LABELS = ["left", "center", "right"]  # 모델 카드 정의
+
 st.set_page_config(page_title=APP_TITLE, page_icon="🧭", layout="wide")
 st.title(APP_TITLE)
-st.caption("TF-IDF + SGDClassifier 증분학습 · 텍스트 성향 분류(내부) · 화면은 '팩트 기반 중립 재구성'만 노출")
+st.caption("PoliticalBiasBERT로 편향(좌/중/우) 분류 → 화면에는 '사실 기반 중립 재구성'만 노출")
 
 # =========================
-# Data & Labels
+# Model (cached)
 # =========================
-@st.cache_resource(show_spinner=False)
-def load_political_tweets():
-    """
-    HuggingFace: jacobvs/PoliticalTweets
-    columns 예: ['index','data','id','username','text','party']
-    label mapping: Democrat=0, Republican=1, NEUTRAL=2
-    """
-    ds = load_dataset("jacobvs/PoliticalTweets")
-
-    def to_example(x):
-        t = x.get("text") or x.get("data") or ""
-        p = x.get("party") or ""
-        return {"text": t, "label": p}
-
-    train = [to_example(r) for r in ds["train"]]
-    test  = [to_example(r) for r in ds["test"]]
-    return train, test
-
-def label_to_id(lbl: str) -> int:
-    lbl = (lbl or "").strip().lower()
-    if "dem" in lbl: return 0
-    if "rep" in lbl: return 1
-    return 2
-
-def id_to_label(i: int) -> str:
-    return ["Democrat","Republican","NEUTRAL"][i]
-
-def split_xy(rows: List[Dict[str,str]]) -> Tuple[List[str], List[int]]:
-    X = [(r["text"] or "") for r in rows]
-    y = [label_to_id(r["label"]) for r in rows]
-    return X, y
+@st.cache_resource(show_spinner=True)
+def load_bias_pipeline():
+    tok = AutoTokenizer.from_pretrained(MODEL_ID)
+    mdl = AutoModelForSequenceClassification.from_pretrained(MODEL_ID)
+    device = 0 if torch.cuda.is_available() else -1
+    clf = pipeline(
+        "text-classification",
+        model=mdl,
+        tokenizer=tok,
+        return_all_scores=True,
+        device=device
+    )
+    return clf
 
-# =========================
-# Model
-# =========================
-def init_or_load_model():
-    if os.path.exists(MODEL_PATH) and os.path.exists(VECT_PATH):
-        model = joblib.load(MODEL_PATH)
-        vect  = joblib.load(VECT_PATH)
-    else:
-        model = SGDClassifier(loss="log_loss")
-        vect  = TfidfVectorizer(max_features=10000, ngram_range=(1,2), stop_words="english")
-    return model, vect
-
-def incremental_fit(model, vect, texts: List[str], labels: List[int]):
-    X = vect.fit_transform(texts)
-    model.partial_fit(X, labels, classes=[0,1,2])
-    joblib.dump(model, MODEL_PATH)
-    joblib.dump(vect, VECT_PATH)
-    return model, vect
-
-def predict(model, vect, text: str) -> Tuple[int, List[float]]:
-    X = vect.transform([text])
-    probs = getattr(model, "predict_proba", None)
-    if probs is None:
-        y = int(model.predict(X)[0])
-        return y, [0.0, 0.0, 0.0]
-    pr = model.predict_proba(X)[0]
-    y  = int(pr.argmax())
-    return y, pr.tolist()
+def classify_bias(clf, text: str) -> Tuple[str, List[float]]:
+    scores = clf(text)[0]  # list of dicts: [{"label":"LABEL_0","score":...}, ...]
+    # 모델이 LABEL_0/1/2를 쓰므로 index 기준으로 정렬되어 들어온다는 가정
+    probs = [s["score"] for s in scores]                     # [left, center, right] 순
+    pred_idx = int(torch.tensor(probs).argmax().item())
+    return LABELS[pred_idx], probs
 
 # =========================
-# External: Naver News (optional)
+# Naver News (optional)
 # =========================
 def fetch_naver_news(query: str, display: int = 10) -> List[Dict[str,str]]:
     if not (NAVER_ID and NAVER_SECRET):
@@ -114,14 +64,11 @@ def fetch_naver_news(query: str, display: int = 10) -> List[Dict[str,str]]:
         if r.status_code != 200:
             return []
         items = r.json().get("items", [])
-        out = []
-        for it in items:
-            out.append({
-                "title": it.get("title",""),
-                "desc": it.get("description",""),
-                "link": it.get("link","")
-            })
-        return out
+        return [{
+            "title": it.get("title",""),
+            "desc":  it.get("description",""),
+            "link":  it.get("link","")
+        } for it in items]
     except Exception:
         return []
 
@@ -129,24 +76,15 @@ def fetch_naver_news(query: str, display: int = 10) -> List[Dict[str,str]]:
 # OpenAI: Fact-based neutral rewrite (optional)
 # =========================
 def generate_fact_based(text: str) -> str:
-    """
-    정치적 해석/평가 제거, 사실 중심 중립 기사로 재구성
-    """
     if not OPENAI_API_KEY:
         return "(OPENAI_API_KEY 미설정: 재구성 생략됨)"
     import openai
     openai.api_key = OPENAI_API_KEY
-
     prompt = (
-        "다음 텍스트를 정치적 해석이나 의견 없이, 사실 중심의 중립 기사로 재구성하세요.\n"
-        "규칙:\n"
-        "1) 누가(행위자)·언제·어디서·무엇을·어떻게·왜 중 사실 정보만 서술\n"
-        "2) 평가/감정/추측 표현 삭제\n"
-        "3) 수치·날짜·인용은 원문에 있는 범위에서만 사용\n"
-        "4) 한국어 5~7문장, 제목 없이 본문만\n\n"
+        "다음 텍스트를 정치적 해석/의견 없이, 사실 중심의 중립 기사로 재구성하세요.\n"
+        "규칙: 1) 누가·언제·어디서·무엇을 중심 2) 평가/추측 삭제 3) 수치/날짜는 원문 범위 4) 한국어 5~7문장\n\n"
         f"[원문]\n{text}\n\n[중립 기사]"
     )
-
     try:
         resp = openai.ChatCompletion.create(
             model="gpt-4o-mini",
@@ -159,89 +97,54 @@ def generate_fact_based(text: str) -> str:
         return f"(재구성 실패: {e})"
 
 # =========================
-# Sidebar: Train / Evaluate
+# Sidebar
 # =========================
 with st.sidebar:
-    st.subheader("데이터 & 학습")
-    if st.button("① 데이터 로드 & 증분학습", use_container_width=True):
-        with st.spinner("데이터셋 로드 및 학습 중…"):
-            train, test = load_political_tweets()
-            Xtr, ytr = split_xy(train)
-            model, vect = init_or_load_model()
-            model, vect = incremental_fit(model, vect, Xtr, ytr)
-        st.success("학습 완료. 모델/벡터 저장됨.")
-
-    if st.button("② 성능 평가", use_container_width=True):
-        try:
-            _, test = load_political_tweets()
-            Xte_texts, yte = split_xy(test)
-            model, vect = init_or_load_model()
-            Xte = vect.transform(Xte_texts)
-            ypred = model.predict(Xte)
-            acc = accuracy_score(yte, ypred)
-            st.write(f"정확도: **{acc:.3f}**")
-            st.code(classification_report(yte, ypred, target_names=["Democrat","Republican","NEUTRAL"]))
-        except Exception as e:
-            st.error(e)
-
-    st.markdown("---")
-    st.caption("주의: 데이터는 미국 정치 트윗(영문). 한국어 기사 일반화엔 한계가 있음.")
+    st.subheader("모델 상태")
+    with st.spinner("BERT 모델 로딩 중… 처음 한 번만 기다리면 됨"):
+        clf = load_bias_pipeline()
+    st.success("PoliticalBiasBERT 로드 완료")
+    st.caption("좌/중/우 분류는 내부 진단용으로만 사용. 화면은 사실 기반 재구성 위주.")
 
 # =========================
-# Main: News fetch → classify (internal) → fact-based rewrite (visible)
+# Main
 # =========================
 st.markdown("### 1) (선택) 네이버 뉴스 검색")
 q = st.text_input("검색어", value="미 대선")
 cnt = st.slider("표시 개수", 1, 20, 10)
 news_items: List[Dict[str,str]] = []
-
-col_btn1, col_btn2 = st.columns([1,1])
-with col_btn1:
-    if st.button("뉴스 불러오기"):
-        with st.spinner("네이버 뉴스 수집 중…"):
-            news_items = fetch_naver_news(q, cnt)
-        if not news_items:
-            st.info("네이버 API 키가 없거나 호출 실패. 아래 자유 입력으로 테스트하세요.")
+if st.button("뉴스 불러오기"):
+    with st.spinner("네이버 뉴스 수집 중…"):
+        news_items = fetch_naver_news(q, cnt)
+    if not news_items:
+        st.info("네이버 API 키가 없거나 호출 실패. 아래 자유 입력으로 테스트하세요.")
 
 st.markdown("### 2) 텍스트 분석 & 사��� 기반 중립 재구성")
-left, right = st.columns([1,1])
-
-with left:
-    sample = ""
-    if news_items:
-        # 첫 아이템을 샘플로 넣어줌 (사용자가 수정 가능)
-        sample = f"{news_items[0]['title']} — {news_items[0]['desc']}"
+c1, c2 = st.columns(2)
+with c1:
+    sample = f"{news_items[0]['title']} — {news_items[0]['desc']}" if news_items else ""
     text = st.text_area("분석할 텍스트(뉴스 제목+요약 등)", value=sample, height=220)
 
-with right:
+with c2:
     if st.button("분석 및 중립 재구성 실행"):
         if not text.strip():
             st.warning("텍스트를 입력하세요.")
         else:
-            # 1) 내부 분류 (화면에 강조하지 않음)
-            model, vect = init_or_load_model()
-            if not os.path.exists(MODEL_PATH) or not os.path.exists(VECT_PATH):
-                st.warning("먼저 사이드바에서 '데이터 로드 & 증분학습'을 실행하세요.")
-            else:
-                y, pr = predict(model, vect, text)
-                pred = id_to_label(y)
-
-                # 2) 중립 기사 재구성 (화면 노출)
-                st.markdown("#### ✅ 사실 기반 중립 재구성 결과")
-                article = generate_fact_based(text)
-                st.write(article)
-
-                # 3) (선택) 진단 정보는 접어서 제공
-                with st.expander("진단 보기(내부 라벨/확률)"):
-                    st.write(f"예측 라벨: **{pred}**")
-                    st.write(f"확률(민·공·중): {pr}")
-
-                if news_items:
-                    with st.expander("원문 링크"):
-                        st.write(news_items[0].get("link","(링크 없음)"))
+            # 내부 분류(진단용)
+            pred, probs = classify_bias(clf, text)
+
+            # 화면 노출: 사실 기반 재구성
+            st.markdown("#### ✅ 사실 기반 중립 재구성 결과")
+            article = generate_fact_based(text)
+            st.write(article)
+
+            # 진단/출처
+            with st.expander("진단 보기(내부 편향 확률)"):
+                st.write(f"예측: **{pred}**")
+                st.write(f"확률 [left, center, right]: {probs}")
+            if news_items:
+                with st.expander("원문 링크"):
+                    st.write(news_items[0].get("link","(링크 없음)"))
 
 st.markdown("---")
-st.caption(
-    "데모 용도. 실제 서비스에서는 출처 자동추출, 사실 검증(예: 인용·숫자 교차검증), "
-    "정치적 표현 필터링, 개인정보/명예훼손 안전장치가 추가되어야 합니다."
-)
+st.caption("데모 용도. 실제 서비스는 출처 추출·사실 검증·정책 필터링을 추가해야 함.")