Update services/sentiment.py

services/sentiment.py

@@ -1,118 +1,127 @@
+"""
+services/sentiment.py
+Model sentimen berbasis IndoBERT / RoBERTa-ID.
+Torch di-import secara lazy agar tidak crash saat package belum siap.
+"""
+
 import os
 
-# ── PATH MODEL FINE-TUNING ──
 LOCAL_MODEL_PATH = "model/final_model"
+FALLBACK_MODEL   = "w11wo/indonesian-roberta-base-sentiment-classifier"
 
-# ── FALLBACK: model pretrained HuggingFace ──
-FALLBACK_MODEL = "w11wo/indonesian-roberta-base-sentiment-classifier"
+# ── RULE-BASED FALLBACK ──
+_POS_KW = [
+    "bagus","baik","senang","suka","mantap","keren","hebat","oke","setuju",
+    "benar","sukses","berhasil","love","good","great","nice","best","amazing",
+    "excellent","wonderful","happy","glad","positif","mendukung","bangga",
+    "luar biasa","terima kasih","apresiasi","semangat","maju","berkembang",
+]
+_NEG_KW = [
+    "buruk","jelek","benci","kecewa","gagal","salah","rugi","marah","bohong",
+    "hoax","fitnah","jahat","tidak setuju","parah","malu","takut","bad",
+    "worst","terrible","hate","fail","wrong","poor","awful","negatif","tolak",
+    "menolak","turun","jatuh","hancur","krisis","masalah","bahaya","ancam",
+]
+
+def _rule_based(text: str) -> str:
+    lower = text.lower()
+    pos   = sum(1 for k in _POS_KW if k in lower)
+    neg   = sum(1 for k in _NEG_KW if k in lower)
+    if pos > neg:   return "Positive"
+    if neg > pos:   return "Negative"
+    return "Neutral"
 
 
-def load_model():
-    """
-    Load pipeline sentimen. Urutan prioritas:
-    1. Model fine-tuned lokal (jika ada)
-    2. Model pretrained dari HuggingFace Hub
-    3. None → fallback ke rule-based
-    """
+# ── MODEL LOADING ──
+def _load_model():
     try:
-        # import di dalam fungsi agar tidak crash saat torch tidak tersedia
         import torch
         from transformers import pipeline
 
-        model_path = LOCAL_MODEL_PATH if os.path.exists(LOCAL_MODEL_PATH) else FALLBACK_MODEL
-        label = "fine-tuned" if os.path.exists(LOCAL_MODEL_PATH) else "fallback RoBERTa"
+        path  = LOCAL_MODEL_PATH if os.path.exists(LOCAL_MODEL_PATH) else FALLBACK_MODEL
+        label = "fine-tuned" if os.path.exists(LOCAL_MODEL_PATH) else "fallback RoBERTa-ID"
 
         clf = pipeline(
             "sentiment-analysis",
-            model=model_path,
-            device=-1,          # CPU-only (HF Spaces free tier)
+            model=path,
+            device=-1,
             truncation=True,
             max_length=512,
         )
-        print(f"✅ Model loaded: {label}")
+        print(f"✅ Sentiment model loaded: {label}")
         return clf
 
     except ImportError:
-        print("⚠️  PyTorch tidak tersedia — menggunakan rule-based fallback")
+        print("⚠️  PyTorch tidak tersedia — rule-based fallback aktif")
         return None
     except Exception as e:
-        print(f"❌ Gagal load model: {e}")
+        print(f"❌ Gagal load sentiment model: {e}")
         return None
 
 
-# Load sekali saat startup
-classifier = load_model()
+classifier = _load_model()
 
 
-# ── NORMALISASI LABEL ──
-def normalize_label(label: str) -> str:
+# ── LABEL NORMALIZATION ──
+def _normalize(label: str) -> str:
     label = label.lower()
-    if "positive" in label or label == "label_2":
-        return "Positive"
-    if "negative" in label or label == "label_0":
-        return "Negative"
-    if "neutral"  in label or label == "label_1":
-        return "Neutral"
-    return "Neutral"
-
-
-# ── RULE-BASED FALLBACK ──
-POS_KW = ["bagus","baik","senang","suka","mantap","keren","hebat","oke","setuju",
-          "benar","bagus","sukses","berhasil","love","good","great","nice","best",
-          "amazing","excellent","wonderful","happy","glad"]
-NEG_KW = ["buruk","jelek","benci","kecewa","gagal","salah","rugi","marah","bohong",
-          "hoax","fitnah","jahat","tidak setuju","parah","malu","takut",
-          "bad","worst","terrible","hate","fail","wrong","poor","awful"]
-
-def rule_based(text: str) -> str:
-    lower = text.lower()
-    pos = sum(1 for k in POS_KW if k in lower)
-    neg = sum(1 for k in NEG_KW if k in lower)
-    if pos > neg:
-        return "Positive"
-    if neg > pos:
-        return "Negative"
+    if "positive" in label or label == "label_2": return "Positive"
+    if "negative" in label or label == "label_0": return "Negative"
+    if "neutral"  in label or label == "label_1": return "Neutral"
     return "Neutral"
 
 
-# ── PREDIKSI UTAMA ──
+# ── PUBLIC API ──
 def predict(texts: list) -> list:
-    if not texts:
-        return []
-
+    """Return list of label strings."""
+    if not texts: return []
     if classifier is None:
-        print("⚠️  Classifier tidak tersedia → rule-based")
-        return [rule_based(t) for t in texts]
-
+        return [_rule_based(t) for t in texts]
     try:
         outputs = classifier(texts, batch_size=8, truncation=True)
-        return [normalize_label(o["label"]) for o in outputs]
+        return [_normalize(o["label"]) for o in outputs]
     except Exception as e:
-        print(f"❌ Error saat prediksi batch: {e}")
-        # per-item fallback
+        print(f"❌ predict() batch error: {e} — per-item fallback")
         results = []
         for t in texts:
             try:
                 out = classifier(t[:512], truncation=True)
-                results.append(normalize_label(out[0]["label"]))
+                results.append(_normalize(out[0]["label"]))
             except Exception:
-                results.append(rule_based(t))
+                results.append(_rule_based(t))
         return results
 
 
-# ── PREDICT SINGLE ──
 def predict_single(text: str) -> str:
     return predict([text])[0]
 
 
-# ── PREDICT WITH SCORE ──
 def predict_with_score(texts: list) -> list:
+    """
+    Return list of dicts: {label, score}
+    score = confidence dari model (0–1).
+    """
+    if not texts: return []
+
     if classifier is None:
-        return [{"label": rule_based(t), "score": 0.0} for t in texts]
+        return [{"label": _rule_based(t), "score": 0.5} for t in texts]
+
     try:
         outputs = classifier(texts, batch_size=8, truncation=True)
-        return [{"label": normalize_label(o["label"]), "score": round(o["score"], 4)}
-                for o in outputs]
+        return [
+            {"label": _normalize(o["label"]), "score": round(float(o["score"]), 4)}
+            for o in outputs
+        ]
     except Exception as e:
-        print(f"❌ Error predict_with_score: {e}")
-        return [{"label": rule_based(t), "score": 0.0} for t in texts]
+        print(f"❌ predict_with_score() error: {e} — per-item fallback")
+        results = []
+        for t in texts:
+            try:
+                out = classifier(t[:512], truncation=True)
+                results.append({
+                    "label": _normalize(out[0]["label"]),
+                    "score": round(float(out[0]["score"]), 4)
+                })
+            except Exception:
+                results.append({"label": _rule_based(t), "score": 0.5})
+        return results