TIF_E41253385/api_slearn/model_service.py

import json
import importlib.util
from pathlib import Path

import joblib

# ============================================================
# PATH
# ============================================================

BASE_DIR = Path(__file__).resolve().parent
ARTIFACT_DIR = BASE_DIR / "artifacts_slearn"

MODEL_PATH = ARTIFACT_DIR / "model_logistic_regression.joblib"
VECTORIZER_PATH = ARTIFACT_DIR / "tfidf_vectorizer.joblib"
METADATA_PATH = ARTIFACT_DIR / "metadata_model.json"
PREPROCESSING_PATH = ARTIFACT_DIR / "preprocessing_slearn.py"

# ============================================================
# CEK FILE ARTEFAK
# ============================================================

required_files = [
    MODEL_PATH,
    VECTORIZER_PATH,
    METADATA_PATH,
    PREPROCESSING_PATH
]

for file_path in required_files:
    if not file_path.exists():
        raise FileNotFoundError(
            f"File artefak tidak ditemukan: {file_path}"
        )

# ============================================================
# LOAD PREPROCESSING ASLI PENELITIAN
# ============================================================

def load_preprocessing_module():
    spec = importlib.util.spec_from_file_location(
        "preprocessing_slearn_artifact",
        PREPROCESSING_PATH
    )

    if spec is None or spec.loader is None:
        raise ImportError(
            "preprocessing_slearn.py tidak dapat dimuat."
        )

    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    return module


preprocessing = load_preprocessing_module()

# ============================================================
# LOAD MODEL, TF-IDF, METADATA
# ============================================================

model = joblib.load(MODEL_PATH)
vectorizer = joblib.load(VECTORIZER_PATH)

metadata = json.loads(
    METADATA_PATH.read_text(encoding="utf-8")
)

feature_names = vectorizer.get_feature_names_out()

# ============================================================
# VERIFIKASI ARTEFAK
# ============================================================

jumlah_fitur_metadata = int(
    metadata["jumlah_fitur_tfidf"]
)

if len(feature_names) != jumlah_fitur_metadata:
    raise RuntimeError(
        "Jumlah fitur TF-IDF tidak sesuai dengan metadata."
    )

if model.n_features_in_ != len(feature_names):
    raise RuntimeError(
        "Jumlah fitur model tidak sesuai dengan TF-IDF."
    )

if model.classes_.tolist() != [0, 1]:
    raise RuntimeError(
        "Urutan kelas model harus [0, 1]."
    )

# ============================================================
# DETAIL PREPROCESSING
# Menggunakan fungsi yang SAMA dari artefak penelitian
# ============================================================

def preprocessing_detail(teks):
    if not isinstance(teks, str) or not teks.strip():
        raise ValueError(
            "Masukkan ulasan terlebih dahulu."
        )

    original_text = teks.strip()

    # 1. Cleaning
    cleaned_text = preprocessing.cleaning(
        original_text
    )

    # 2. Case Folding
    case_folded_text = preprocessing.case_folding(
        cleaned_text
    )

    # 3. Normalisasi
    normalized_text = preprocessing.normalisasi(
        case_folded_text
    )

    normalized_text = (
        preprocessing.koreksi_normalisasi_validator(
            normalized_text
        )
    )

    # 4. Tokenisasi
    tokenized_text = preprocessing.tokenisasi(
        normalized_text
    )

    # 5. Stopword Removal
    stopword_text = preprocessing.hapus_stopword(
        tokenized_text
    )

    # 6. Stemming
    stemmed_text = preprocessing.stemming(
        stopword_text
    )

    return {
        "original_text": original_text,
        "cleaning": cleaned_text,
        "case_folding": case_folded_text,
        "normalization": normalized_text,
        "tokenizing": tokenized_text,
        "stopword_removal": stopword_text,
        "stemming": stemmed_text
    }

# ============================================================
# TF-IDF TERATAS
# ============================================================

def get_top_tfidf(fitur, top_n=10, names=None):
    names = feature_names if names is None else names
    if fitur.nnz == 0:
        return []

    row = fitur.getrow(0)

    data = []

    for index, bobot in zip(
        row.indices,
        row.data
    ):
        data.append({
            "kata": str(names[index]),
            "bobot": float(bobot)
        })

    data.sort(
        key=lambda item: item["bobot"],
        reverse=True
    )

    return data[:top_n]

# ============================================================
# ANALISIS SATU TEKS
# ============================================================

def analisis_teks(teks, snapshot=None):
    from model_registry import bundle
    model, vectorizer, metadata = snapshot or bundle()
    proses = preprocessing_detail(teks)

    stemmed_text = proses["stemming"]

    if not stemmed_text.strip():
        raise ValueError(
            "Ulasan tidak memiliki kata yang dapat diproses."
        )

    # PENTING:
    # hanya transform(), TIDAK fit ulang
    fitur = vectorizer.transform(
        [stemmed_text]
    )

    # Jika tidak ada fitur penelitian yang dikenali,
    # Logistic Regression TIDAK dijalankan
    if fitur.nnz == 0:
        raise ValueError(
            "Kata dalam ulasan belum dikenali model."
        )

    label = int(
        model.predict(fitur)[0]
    )

    sentiment = metadata[
        "label_mapping"
    ][str(label)].capitalize()

    tfidf_top = get_top_tfidf(
        fitur,
        top_n=10,
        names=vectorizer.get_feature_names_out()
    )

    return {
        "model_version": metadata['version'],
        "teks": proses["original_text"],
        "label": label,
        "sentimen": sentiment,
        "preprocessing": {
            "cleaning": proses["cleaning"],
            "case_folding": proses["case_folding"],
            "normalization": proses["normalization"],
            "tokenizing": proses["tokenizing"],
            "stopword_removal": proses["stopword_removal"],
            "stemming": proses["stemming"]
        },
        "jumlah_fitur_dikenali": int(
            fitur.nnz
        ),
        "tfidf_top": tfidf_top
    }

# ============================================================
# INFORMASI MODEL
# ============================================================

def get_model_info():
    from model_registry import bundle
    _, _, metadata = bundle()
    return {
        "version": metadata['version'],
        "nama_model": metadata["nama_model"],
        "total_data": metadata["total_data"],
        "jumlah_sentimen_negatif":
            metadata["jumlah_sentimen_negatif"],
        "jumlah_sentimen_positif":
            metadata["jumlah_sentimen_positif"],
        "jumlah_data_training":
            metadata["jumlah_data_training"],
        "jumlah_data_testing":
            metadata["jumlah_data_testing"],
        "jumlah_fitur_tfidf":
            metadata["jumlah_fitur_tfidf"],
        "akurasi":
            metadata["akurasi"],
        "label_mapping":
            metadata["label_mapping"],
        "confusion_matrix":
            metadata["confusion_matrix"],
        "classification_report":
            metadata["classification_report"]
    }