api/service.py

"""Model loading, prediction and stored-metrics access for the API.
Only one model is served: whatever is saved as ``artifacts/best_model.joblib``.
"""

import json
from functools import lru_cache

from config.constants import MIN_WORDS_PER_SENTENCE, NEGATION_WORDS
from config.emoji_map import emojis as EMOJI_TO_ARABIC
from config.paths import ARTIFACTS_DIR, DEFAULT_MODEL_PATH
from preprocessing.pipeline import clean_text
from training.model_io import load_model
from utils.chart_style import PLOTS_SUBDIR

#: Written by ``run_pipeline.py``
METRICS_PATH = ARTIFACTS_DIR / "metrics.json"

#: Where ``--save-plots`` puts the PNGs.
PLOTS_DIR = ARTIFACTS_DIR / PLOTS_SUBDIR

#: Stable keys the web app asks for -> the file names the pipeline writes.
PLOT_FILES = {
    "sentiment_distribution_raw": "01_sentiment_distribution_raw.png",
    "sentiment_distribution_balanced": "02_sentiment_distribution_balanced.png",
    "review_length": "03_review_length.png",
    "top_tokens": "04_top_tokens.png",
    "model_comparison": "05_model_comparison.png",
    "confusion_matrix": "06_confusion_matrix.png",
    "per_class_metrics": "07_per_class_metrics.png",
}


class EmptyAfterCleaning(ValueError):
    """Raised when cleaning leaves nothing for the model to classify.
    Happens when the input has no Arabic content at all - the ``Removing_non_arabic`` step strips it to an empty string.
    """


@lru_cache(maxsize=1)
def get_model():
    """Load the saved pipeline once and reuse it for every request."""
    return load_model(DEFAULT_MODEL_PATH)


def describe_model():
    """Report what the served pipeline actually is."""
    pipeline = get_model()
    classifier = pipeline.named_steps["clf"]
    vectorizer = pipeline.named_steps["vect"]

    return {
        "name": type(classifier).__name__,
        "classes": [str(label) for label in classifier.classes_],
        "vocabulary_size": len(vectorizer.vocabulary_),
        "supports_probabilities": hasattr(pipeline, "predict_proba"),
        "supports_margins": hasattr(pipeline, "decision_function"),
    }


def _class_scores(pipeline, cleaned):
    """Per-class scores for one cleaned string, or ``None`` if unavailable.

    Two different things can come back, and the caller must not conflate them:

    ``probabilities``
        Real calibrated probabilities summing to 1, from ``predict_proba``.
        Available for MultinomialNB / LogisticRegression / RandomForest.

    ``margins``
        Signed distances from ``decision_function``. ``SVC`` has these but no
        probabilities. They rank the classes but are *not* percentages, so the
        UI labels them as margins and never renders them as a confidence.
    """
    if hasattr(pipeline, "predict_proba"):
        scores = pipeline.predict_proba([cleaned])[0]
        return "probabilities", {
            str(label): float(score)
            for label, score in zip(pipeline.classes_, scores)
        }

    if hasattr(pipeline, "decision_function"):
        scores = pipeline.decision_function([cleaned])[0]
        return "margins", {
            str(label): float(score)
            for label, score in zip(pipeline.classes_, scores)
        }

    return None, None


def _find_negations(text):
    """Negation particles present in the raw input.

    ``KEEP_NEGATIONS`` is on, so these survive stopword removal - which is the
    whole point, since dropping them would turn "لا احب" into "احب".
    """
    tokens = set(text.split())
    return sorted(tokens & NEGATION_WORDS)


def _find_emojis(text):
    """Emoji in the raw input paired with the Arabic word they are replaced by."""
    found = []
    seen = set()
    for character in text:
        if character in EMOJI_TO_ARABIC and character not in seen:
            seen.add(character)
            found.append({"emoji": character, "arabic": EMOJI_TO_ARABIC[character]})
    return found


def analyze(text, convert_emojis=True):
    """Clean ``text``, classify it, and report everything worth showing.

    Args:
        text: raw user input.
        convert_emojis: run the emoji substitution steps first. ``True`` matches how the training corpus was built.
    """
    pipeline = get_model()
    original = text.strip()
    cleaned = clean_text(original, convert_emojis=convert_emojis).strip()

    if not cleaned:
        raise EmptyAfterCleaning(
            "Nothing left after cleaning - the input has no Arabic content."
        )

    label = str(pipeline.predict([cleaned])[0])
    score_kind, scores = _class_scores(pipeline, cleaned)

    word_count = len(original.split())
    cleaned_word_count = len(cleaned.split())

    notes = []
    if word_count < MIN_WORDS_PER_SENTENCE:
        notes.append(
            f"Only {word_count} word(s). Reviews shorter than "
            f"{MIN_WORDS_PER_SENTENCE} words were dropped from the training "
            f"corpus, so this is outside what the model learned from."
        )
    if cleaned_word_count < MIN_WORDS_PER_SENTENCE <= word_count:
        notes.append(
            f"Cleaning reduced this to {cleaned_word_count} word(s). Most of the "
            f"input was stopwords, punctuation or non-Arabic characters."
        )

    return {
        "label": label,
        "model": type(pipeline.named_steps["clf"]).__name__,
        "original_text": original,
        "cleaned_text": cleaned,
        "word_count": word_count,
        "cleaned_word_count": cleaned_word_count,
        "score_kind": score_kind,
        "scores": scores,
        "negations_found": _find_negations(original),
        "emojis_found": _find_emojis(original),
        "notes": notes,
    }


def load_metrics():
    """Read ``artifacts/metrics.json``, or return an unavailable placeholder.
    The file is written by ``run_pipeline.py``.
    """
    available_plots = {
        key: name
        for key, name in PLOT_FILES.items()
        if (PLOTS_DIR / name).is_file()
    }

    if not METRICS_PATH.is_file():
        return {
            "available": False,
            "best_model": None,
            "selected_by": None,
            "generated_at": None,
            "dataset": None,
            "models": [],
            "plots": available_plots,
        }

    stored = json.loads(METRICS_PATH.read_text(encoding="utf-8"))
    stored["available"] = True
    stored.setdefault("plots", {})
    # Trust the filesystem over the manifest: charts can be deleted or regenerated without rerunning training.
    stored["plots"] = available_plots or stored["plots"]
    return stored


def plot_path(name):
    """Absolute path of a chart PNG, or ``None`` if it is not a known chart.
    Only names listed in :data:`PLOT_FILES` resolve, so a request cannot walk
    out of the plots directory.
    """
    if name not in PLOT_FILES.values():
        return None
    candidate = PLOTS_DIR / name
    return candidate if candidate.is_file() else None