app/services/embedding_service.py

"""
Embedding service: compute semantic similarity between resume and JD.

Strategy (two-tier):
  1. Try sentence-transformers (all-MiniLM-L6-v2) for quality embeddings.
  2. Fall back to TF-IDF cosine similarity (sklearn) if model unavailable.

The model is loaded lazily so the server starts instantly even if the
first analysis takes a few extra seconds.
"""
from __future__ import annotations

import logging
import threading
from typing import Any

import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

logger = logging.getLogger(__name__)

# ── Lazy sentence-transformers loader ────────────────────────────────────


_model_lock = threading.Lock()
_model: Any = None          # SentenceTransformer or None
_model_loaded = False       # whether we've attempted loading

_ST_MODEL = "all-MiniLM-L6-v2"


def _try_load_st_model() -> Any | None:
    """Attempt to load the SentenceTransformer model once."""
    global _model, _model_loaded
    with _model_lock:
        if _model_loaded:
            return _model
        _model_loaded = True
        try:
            from sentence_transformers import SentenceTransformer  # noqa: PLC0415
            logger.info("Loading SentenceTransformer model %s …", _ST_MODEL)
            _model = SentenceTransformer(_ST_MODEL)
            logger.info("SentenceTransformer model ready.")
        except Exception as exc:
            logger.warning("SentenceTransformer unavailable (%s); using TF-IDF fallback.", exc)
            _model = None
        return _model


# ── Public API ────────────────────────────────────────────────────────────


def compute_similarity(text_a: str, text_b: str) -> float:
    """
    Return a cosine similarity score in [0, 1] between two texts.
    Uses sentence-transformers if available, else TF-IDF.
    """
    if not text_a.strip() or not text_b.strip():
        return 0.0

    model = _try_load_st_model()
    if model is not None:
        return _st_similarity(model, text_a, text_b)
    return _tfidf_similarity(text_a, text_b)


def compute_section_similarities(
    sections: dict[str, str], jd_text: str
) -> dict[str, float]:
    """
    Compute per-section similarity against the JD.
    Returns dict of {section_name: score_0_to_1}.
    """
    results: dict[str, float] = {}
    for name, content in sections.items():
        if content.strip():
            results[name] = compute_similarity(content, jd_text)
    return results


def _st_similarity(model: Any, a: str, b: str) -> float:
    embs = model.encode([a, b], convert_to_numpy=True)
    score = cosine_similarity(embs[0:1], embs[1:2])[0][0]
    return float(np.clip(score, 0.0, 1.0))


def _tfidf_similarity(a: str, b: str) -> float:
    try:
        vec = TfidfVectorizer(
            ngram_range=(1, 2),
            stop_words="english",
            max_features=8000,
        )
        mat = vec.fit_transform([a, b])
        score = cosine_similarity(mat[0:1], mat[1:2])[0][0]
        return float(np.clip(score, 0.0, 1.0))
    except Exception as exc:
        logger.error("TF-IDF similarity failed: %s", exc)
        return 0.0