190 lines
6.3 KiB
Python
190 lines
6.3 KiB
Python
import os
|
|
import re
|
|
import string
|
|
import joblib
|
|
import requests
|
|
import nltk
|
|
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
from nltk.tokenize import word_tokenize
|
|
from Sastrawi.Stemmer.StemmerFactory import StemmerFactory
|
|
|
|
# ---------------------------------------------------------
|
|
# 1. KONFIGURASI ENVIRONMENT & VERCEL BLOB
|
|
# ---------------------------------------------------------
|
|
BLOB_TOKEN = os.environ.get("BLOB_READ_WRITE_TOKEN", "vercel_blob_rw_rtoKzdsaGXwvNykn_6VpBFIALjlA8NPCMtwJmUVTOzisxfe")
|
|
MODEL_FILENAME = "model_sentiment.pkl"
|
|
MODEL_PATH = f"/tmp/{MODEL_FILENAME}"
|
|
|
|
# ---------------------------------------------------------
|
|
# 2. KONFIGURASI NLTK UNTUK VERCEL (SERVERLESS)
|
|
# ---------------------------------------------------------
|
|
nltk_data_path = '/tmp/nltk_data'
|
|
os.makedirs(nltk_data_path, exist_ok=True)
|
|
nltk.data.path.append(nltk_data_path)
|
|
|
|
try:
|
|
nltk.data.find('tokenizers/punkt')
|
|
except LookupError:
|
|
nltk.download('punkt', download_dir=nltk_data_path)
|
|
|
|
try:
|
|
nltk.data.find('tokenizers/punkt_tab')
|
|
except LookupError:
|
|
nltk.download('punkt_tab', download_dir=nltk_data_path)
|
|
|
|
# ---------------------------------------------------------
|
|
# 3. INISIALISASI SASTRAWI GLOBAL (PENCEGAH TIMEOUT VERCEL)
|
|
# ---------------------------------------------------------
|
|
factory = StemmerFactory()
|
|
global_stemmer = factory.create_stemmer()
|
|
|
|
app = FastAPI(title="Sentiment Analysis API")
|
|
|
|
# ---------------------------------------------------------
|
|
# 4. STATE MODEL GLOBAL
|
|
# ---------------------------------------------------------
|
|
model_state = {
|
|
"is_loaded": False,
|
|
"vectorizer": None,
|
|
"classifier": None,
|
|
"metrics": None,
|
|
"slangwords": {},
|
|
"stopwords": set(),
|
|
"data": None
|
|
}
|
|
|
|
class PredictRequest(BaseModel):
|
|
texts: list[str]
|
|
|
|
# ---------------------------------------------------------
|
|
# 5. FUNGSI UNDUH & MUAT MODEL DARI VERCEL BLOB
|
|
# ---------------------------------------------------------
|
|
def load_model_from_blob():
|
|
try:
|
|
headers = {"Authorization": f"Bearer {BLOB_TOKEN}"}
|
|
list_url = "https://blob.vercel-storage.com"
|
|
response = requests.get(list_url, headers=headers)
|
|
response.raise_for_status()
|
|
|
|
blobs = response.json().get('blobs', [])
|
|
pkl_url = next((b['url'] for b in blobs if b['pathname'].endswith(MODEL_FILENAME)), None)
|
|
|
|
if not pkl_url:
|
|
raise Exception(f"File {MODEL_FILENAME} tidak ditemukan di Vercel Blob.")
|
|
|
|
pkl_response = requests.get(pkl_url)
|
|
pkl_response.raise_for_status()
|
|
with open(MODEL_PATH, 'wb') as f:
|
|
f.write(pkl_response.content)
|
|
|
|
loaded_data = joblib.load(MODEL_PATH)
|
|
|
|
model_state["vectorizer"] = loaded_data['vectorizer']
|
|
model_state["classifier"] = loaded_data['classifier']
|
|
model_state["metrics"] = loaded_data['metrics']
|
|
model_state["slangwords"] = loaded_data['preprocessing_assets']['slangwords']
|
|
model_state["stopwords"] = loaded_data['preprocessing_assets']['stopwords']
|
|
model_state["data"] = loaded_data.get('data')
|
|
model_state["is_loaded"] = True
|
|
print("Model berhasil dimuat dari Vercel Blob!")
|
|
|
|
except Exception as e:
|
|
print(f"Gagal memuat model: {e}")
|
|
raise e
|
|
|
|
# ---------------------------------------------------------
|
|
# 6. PIPELINE PRAPEMROSESAN
|
|
# ---------------------------------------------------------
|
|
def cleaningText(text):
|
|
text = re.sub(r'@[A-Za-z0-9]+', ' ', text)
|
|
text = re.sub(r'#[A-Za-z0-9]+', ' ', text)
|
|
text = re.sub(r'RT[\s]', ' ', text)
|
|
text = re.sub(r"http\S+", ' ', text)
|
|
text = re.sub(r'[0-9]+', ' ', text)
|
|
text = re.sub(r'[^\w\s]', ' ', text)
|
|
text = text.replace('\n', ' ')
|
|
text = text.translate(str.maketrans('', '', string.punctuation))
|
|
return text.strip(' ')
|
|
|
|
def casefoldingText(text):
|
|
return text.lower()
|
|
|
|
def fast_fix_slangwords(text):
|
|
pattern = re.compile(r'\b\w+\b')
|
|
return pattern.sub(lambda x: model_state['slangwords'].get(x.group(), x.group()), text)
|
|
|
|
def fast_filteringText(text):
|
|
return [txt for txt in text if txt not in model_state['stopwords']]
|
|
|
|
def stemmingText(text_list):
|
|
return global_stemmer.stem(' '.join(text_list))
|
|
|
|
def fast_preprocess_pipeline(text):
|
|
text = cleaningText(text)
|
|
text = casefoldingText(text)
|
|
text = fast_fix_slangwords(text)
|
|
text = word_tokenize(text)
|
|
text = fast_filteringText(text)
|
|
text = stemmingText(text)
|
|
return text
|
|
|
|
# ---------------------------------------------------------
|
|
# 7. ENDPOINTS
|
|
# ---------------------------------------------------------
|
|
@app.get("/")
|
|
def read_root():
|
|
return {
|
|
"status": "Online",
|
|
"message": "API Sentimen Fast API siap digunakan.",
|
|
"model_loaded": model_state["is_loaded"]
|
|
}
|
|
|
|
@app.get("/info")
|
|
def get_info():
|
|
if not model_state["is_loaded"]:
|
|
raise HTTPException(status_code=503, detail="Model belum dimuat. Silakan akses POST /reload-model.")
|
|
return {
|
|
"status": "success",
|
|
"data": model_state["data"]
|
|
}
|
|
|
|
@app.get("/metrics")
|
|
def get_metrics():
|
|
if not model_state["is_loaded"]:
|
|
raise HTTPException(status_code=503, detail="Model belum dimuat. Silakan akses POST /reload-model.")
|
|
return {
|
|
"status": "success",
|
|
"metrics": model_state["metrics"]
|
|
}
|
|
|
|
@app.post("/reload-model")
|
|
def reload_model():
|
|
try:
|
|
load_model_from_blob()
|
|
return {"status": "success", "message": "Model berhasil diunduh ulang dan dimuat."}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"Gagal memuat model: {str(e)}")
|
|
|
|
@app.post("/predict")
|
|
def predict(req: PredictRequest):
|
|
if not model_state["is_loaded"]:
|
|
raise HTTPException(status_code=503, detail="Model belum dimuat. Silakan akses POST /reload-model.")
|
|
|
|
results = []
|
|
|
|
try:
|
|
clean_texts = [fast_preprocess_pipeline(t) for t in req.texts]
|
|
|
|
vectorized_texts = model_state["vectorizer"].transform(clean_texts)
|
|
predictions = model_state["classifier"].predict(vectorized_texts)
|
|
|
|
for pred in predictions:
|
|
results.append({
|
|
"sentiment": pred.item()
|
|
})
|
|
|
|
return results
|
|
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"Error saat prediksi: {str(e)}") |