Upload 2 files

utils/emotion_utils.py

@@ -0,0 +1,82 @@
+# utils/emotion_utils.py
+import re
+import math
+import numpy as np
+from collections import defaultdict
+
+LABELS = ['불평/불만', '환영/호의', '감동/감탄', '지긋지긋', '고마움', '슬픔',
+          '화남/분노', '존경', '기대감', '우쭐댐/무시함', '안타까움/실망', '비장함',
+          '의심/불신', '뿌듯함', '편안/쾌적', '신기함/관심', '아껴주는', '부끄러움',
+          '공포/무서움', '절망', '한심함', '역겨움/징그러움', '짜증', '어이없음',
+          '없음', '패배/자기혐오', '귀찮음', '힘듦/지침', '즐거움/신남', '깨달음',
+          '죄책감', '증오/혐오', '흐뭇함(귀여움/예쁨)', '당황/난처', '경악',
+          '부담/안_내킴', '서러움', '재미없음', '불쌍함/연민', '놀람', '행복',
+          '불안/걱정', '기쁨', '안심/신뢰']
+
+NEGATIVE_EMOTIONS = [
+    '불평/불만', '지긋지긋', '슬픔', '화남/분노', '의심/불신', '공포/무서움', '절망', '한심함',
+    '역겨움/징그러움', '짜증', '어이없음', '패배/자기혐오', '귀찮음', '힘듦/지침', '죄책감',
+    '증오/혐오', '당황/난처', '부담/안_내킴', '서러움', '재미없음'
+]
+
+def parse_dialogue(text: str):
+    """
+    "화자:문장" 형식의 줄단위 대화를 파싱
+    """
+    lines = [ln.strip() for ln in text.strip().split("\n") if ln.strip()]
+    pairs = []
+    for line in lines:
+        m = re.match(r"([^:]+):(.+)", line)
+        if m:
+            pairs.append((m.group(1).strip(), m.group(2).strip()))
+    return pairs
+
+def adjusted_score(raw_prob: float, k: float = 5.0) -> float:
+    """
+    [0,1] 확률을 사람이 직관적으로 이해할 수 있도록 0~100 점수로 스케일링
+    (logistic stretching)
+    """
+    return 100.0 / (1.0 + math.exp(-k * (raw_prob - 0.5)))
+
+def apply_ema(series, alpha=0.4):
+    if not series:
+        return []
+    smoothed = [series[0]]
+    for s in series[1:]:
+        smoothed.append(alpha * s + (1 - alpha) * smoothed[-1])
+    return smoothed
+
+def detect_emotion_spikes(emotion_series, z_thresh=1.8, min_len=5):
+    """
+    Z-score 기반 급변 감정 탐지
+    - 통계적 배경: 표준점수 z = (x - μ) / σ
+    - 권장 z 임계: 1.8~2.5 (데이터 변동성에 따라 조정)
+    """
+    if len(emotion_series) < min_len:
+        return []
+    mean = float(np.mean(emotion_series))
+    std = float(np.std(emotion_series)) + 1e-6
+    spikes = []
+    for i, v in enumerate(emotion_series):
+        z = (v - mean) / std
+        if z >= z_thresh:
+            spikes.append((i, v, round(z, 2)))
+    return spikes
+
+def infer_conflict_initiator(dialogue, spikes_by_speaker):
+    """
+    간단한 규칙기반 유발자 추정:
+    - B의 감정이 급변한 시점 idx의 직전 발화자가 A라면, A가 유발 count +1
+    - 가장 많은 급변을 유발한 화자를 '추정 유발자'로 반환
+    """
+    blame = {}
+    for speaker, spikes in spikes_by_speaker.items():
+        for (idx, v, z) in spikes:
+            if idx == 0:
+                continue
+            prev_speaker = dialogue[idx - 1][0]
+            if prev_speaker != speaker:
+                blame[prev_speaker] = blame.get(prev_speaker, 0) + 1
+    if not blame:
+        return None
+    return sorted(blame.items(), key=lambda x: -x[1])[0][0]

utils/prediction_utils.py

@@ -0,0 +1,83 @@
+# utils/prediction_utils.py
+import warnings
+warnings.filterwarnings("ignore")
+
+import numpy as np
+
+# Prophet 시도, 실패 시 Holt(지수 평활) 폴백
+try:
+    from prophet import Prophet
+    _HAS_PROPHET = True
+except Exception:
+    _HAS_PROPHET = False
+
+try:
+    from statsmodels.tsa.holtwinters import ExponentialSmoothing
+    _HAS_HOLT = True
+except Exception:
+    _HAS_HOLT = False
+
+import pandas as pd
+
+def prophet_emotion_forecast(scores, steps=3):
+    """
+    Prophet 기반 예측. 입력: 점수 리스트(0~100)
+    반환: 예측 점수 리스트 길이=steps (float)
+    참고: Taylor & Letham (2018) 'Forecasting at Scale'
+    """
+    if not _HAS_PROPHET or len(scores) < 5:
+        return None
+
+    df = pd.DataFrame({
+        'ds': pd.date_range(start='2023-01-01', periods=len(scores), freq='D'),
+        'y': scores
+    })
+    model = Prophet(
+        daily_seasonality=False, weekly_seasonality=False, yearly_seasonality=False,
+        changepoint_prior_scale=0.2  # 변화점 민감도
+    )
+    model.fit(df)
+    future = model.make_future_dataframe(periods=steps)
+    forecast = model.predict(future)
+    tail = forecast[['yhat']].tail(steps)['yhat'].to_list()
+    # 안전 클리핑
+    return [float(np.clip(x, 0, 100)) for x in tail]
+
+def holt_fallback_forecast(scores, steps=3):
+    """
+    Holt 지수평활(추세형) 폴백 예측.
+    Prophet 사용 불가 환경에서 간단하고 안정적.
+    """
+    if not _HAS_HOLT or len(scores) < 4:
+        return None
+    try:
+        model = ExponentialSmoothing(
+            np.array(scores, dtype=float),
+            trend='add', seasonal=None
+        ).fit()
+        fcast = model.forecast(steps)
+        return [float(np.clip(x, 0, 100)) for x in fcast.tolist()]
+    except Exception:
+        return None
+
+def forecast_scores(scores, steps=3):
+    """
+    통합 예측 엔진:
+    1) Prophet 성공 시 Prophet 사용
+    2) 아니면 Holt 폴백
+    3) 그래도 안 되면, 마지막 기울기 선형 외삽(극단 폴백)
+    """
+    if len(scores) < 2:
+        return [scores[-1]] * steps if scores else [0.0]*steps
+
+    yhat = prophet_emotion_forecast(scores, steps=steps)
+    if yhat is not None:
+        return yhat
+
+    yhat = holt_fallback_forecast(scores, steps=steps)
+    if yhat is not None:
+        return yhat
+
+    # 마지막 두 점의 기울기로 외삽
+    slope = scores[-1] - scores[-2]
+    return [float(np.clip(scores[-1] + slope*(i+1), 0, 100)) for i in range(steps)]