Update services/sentiment.py

services/sentiment.py

@@ -1,102 +1,118 @@
-from transformers import pipeline
 import os
 
-# 🔥 PATH MODEL (hasil fine-tuning)
+# ── PATH MODEL FINE-TUNING ──
 LOCAL_MODEL_PATH = "model/final_model"
 
-# 🔄 fallback model (pretrained)
+# ── FALLBACK: model pretrained HuggingFace ──
 FALLBACK_MODEL = "w11wo/indonesian-roberta-base-sentiment-classifier"
 
 
-# 🚀 INIT MODEL
 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
+    """
     try:
-        # 👉 cek apakah model fine-tuning ada
-        if os.path.exists(LOCAL_MODEL_PATH):
-            print("✅ Load fine-tuned model")
-            return pipeline("sentiment-analysis", model=LOCAL_MODEL_PATH)
-
-        else:
-            print("⚠️ Load fallback model (RoBERTa)")
-            return pipeline("sentiment-analysis", model=FALLBACK_MODEL)
-
+        # 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"
+
+        clf = pipeline(
+            "sentiment-analysis",
+            model=model_path,
+            device=-1,          # CPU-only (HF Spaces free tier)
+            truncation=True,
+            max_length=512,
+        )
+        print(f"✅ Model loaded: {label}")
+        return clf
+
+    except ImportError:
+        print("⚠️  PyTorch tidak tersedia — menggunakan rule-based fallback")
+        return None
     except Exception as e:
-        print("❌ Gagal load model:", e)
+        print(f"❌ Gagal load model: {e}")
         return None
 
 
-# 🔥 LOAD SEKALI SAJA (BIAR CEPAT)
+# Load sekali saat startup
 classifier = load_model()
 
 
-# 🧠 NORMALISASI LABEL
-def normalize_label(label):
+# ── NORMALISASI LABEL ──
+def normalize_label(label: str) -> str:
     label = label.lower()
-
-    # 👉 untuk model huggingface (positive/negative)
-    if "positive" in label:
+    if "positive" in label or label == "label_2":
         return "Positive"
-    elif "negative" in label:
+    if "negative" in label or label == "label_0":
         return "Negative"
-    elif "neutral" in label:
+    if "neutral"  in label or label == "label_1":
         return "Neutral"
+    return "Neutral"
 
-    # 👉 untuk model fine-tuned (LABEL_0,1,2)
-    if label == "label_0":
-        return "Negative"
-    elif label == "label_1":
-        return "Neutral"
-    elif label == "label_2":
-        return "Positive"
 
+# ── 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"
     return "Neutral"
 
 
-# 🔍 PREDICT UTAMA
-def predict(texts):
-    results = []
+# ── PREDIKSI UTAMA ──
+def predict(texts: list) -> list:
+    if not texts:
+        return []
 
     if classifier is None:
-        print("⚠️ Model tidak tersedia")
-        return ["Neutral"] * len(texts)
+        print("⚠️  Classifier tidak tersedia → rule-based")
+        return [rule_based(t) for t in texts]
 
     try:
-        # 🔥 batched prediction (lebih cepat)
         outputs = classifier(texts, batch_size=8, truncation=True)
-
-        for o in outputs:
-            label = normalize_label(o['label'])
-            results.append(label)
-
+        return [normalize_label(o["label"]) for o in outputs]
     except Exception as e:
-        print("❌ Error saat prediksi:", e)
-        results = ["Neutral"] * len(texts)
-
-    return results
-
-
-# 🔎 PREDICT SINGLE (opsional)
-def predict_single(text):
+        print(f"❌ Error saat prediksi batch: {e}")
+        # per-item fallback
+        results = []
+        for t in texts:
+            try:
+                out = classifier(t[:512], truncation=True)
+                results.append(normalize_label(out[0]["label"]))
+            except Exception:
+                results.append(rule_based(t))
+        return results
+
+
+# ── PREDICT SINGLE ──
+def predict_single(text: str) -> str:
     return predict([text])[0]
 
 
-# 📊 PREDICT + SCORE (opsional untuk analisis lebih lanjut)
-def predict_with_score(texts):
-    results = []
-
+# ── PREDICT WITH SCORE ──
+def predict_with_score(texts: list) -> list:
     if classifier is None:
-        return [{"label": "Neutral", "score": 0}] * len(texts)
-
+        return [{"label": rule_based(t), "score": 0.0} for t in texts]
     try:
         outputs = classifier(texts, batch_size=8, truncation=True)
-
-        for o in outputs:
-            results.append({
-                "label": normalize_label(o['label']),
-                "score": round(o['score'], 4)
-            })
-
+        return [{"label": normalize_label(o["label"]), "score": round(o["score"], 4)}
+                for o in outputs]
     except Exception as e:
-        print("❌ Error:", e)
-
-    return results
+        print(f"❌ Error predict_with_score: {e}")
+        return [{"label": rule_based(t), "score": 0.0} for t in texts]