services/job_matcher.py

from __future__ import annotations

import hashlib
import json
import threading
from typing import Any, Dict, List, Optional, Set, Tuple

import numpy as np
import torch
from cachetools import TTLCache
from pydantic import BaseModel, Field
from sentence_transformers import SentenceTransformer

from services.processor_utils import (
    DEFAULT_SKILL_ALIASES,
    DEFAULT_CATEGORY_SKILLS,
    SENIORITY_ORDER,
)

# ---------------------------------------------------------------------------
# Requires (for the ONNX path): pip install "optimum[onnxruntime]" onnxruntime
# If these aren't installed, model loading below automatically falls back to
# the plain PyTorch backend -- nothing crashes, it just runs at pre-ONNX speed.
# ---------------------------------------------------------------------------
try:
    import onnxruntime as ort

    _ONNXRUNTIME_AVAILABLE = True
except ImportError:
    _ONNXRUNTIME_AVAILABLE = False

# ---------------------------------------------------------------------------
# Pin CPU thread count to the box's vCPU count. Left unpinned, PyTorch's
# (and separately, ONNX Runtime's) default intra-op threading can
# oversubscribe a 2-vCPU host and cause contention between concurrent
# encode() calls.
# ---------------------------------------------------------------------------
torch.set_num_threads(2)


# ---------------------------------------------------------------------------
# Pydantic models
# ---------------------------------------------------------------------------


class JobInput(BaseModel):
    """Strictly typed job posting for JobMatcher.predict()."""

    job_id: str
    job_title: str
    company_name: str
    description: str
    skills_canonical: List[str] = Field(default_factory=list)
    skills_technical: List[str] = Field(default_factory=list)
    skills_soft: List[str] = Field(default_factory=list)
    category: str = "Other"
    seniority: str = "mid"
    min_years_experience: int = 0


class ScoreBreakdown(BaseModel):
    skill_score: float
    semantic_score: float
    title_score: float
    category_score: float
    seniority_score: float
    experience_score: float
    category_modifier: float = 0.0


class MismatchAnalysis(BaseModel):
    reason: str
    matched_technical_skills: List[str]
    missing_technical_skills: List[str]
    matched_soft_skills: List[str]
    missing_soft_skills: List[str]
    cv_extra_strengths: List[str]
    seniority_fit: str
    category_fit: str
    improvement_advice: List[str]


class MatchPrediction(BaseModel):
    job_id: str
    job_title: str
    company_name: str
    match_score: str
    match_level: str
    cv_title: str
    cv_seniority: str
    cv_category: str
    cv_skills: List[str]
    cv_years_experience: int
    score_breakdown: ScoreBreakdown
    mismatch_analysis: MismatchAnalysis


# ---------------------------------------------------------------------------
# Pure scoring helpers
# ---------------------------------------------------------------------------


def _cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    norm_a = np.linalg.norm(a)
    norm_b = np.linalg.norm(b)
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return float(np.dot(a, b) / (norm_a * norm_b))


def _seniority_fit(cv_seniority: str, job_seniority: str) -> Tuple[float, str]:
    cv_rank = SENIORITY_ORDER.get(cv_seniority, 1)
    job_rank = SENIORITY_ORDER.get(job_seniority, 2)
    diff = cv_rank - job_rank
    if diff == 0:
        return 1.0, f"Good seniority fit ({cv_seniority} → {job_seniority})."
    if diff == -1:
        return (
            0.65,
            f"Slightly under the requested level ({cv_seniority} → {job_seniority}).",
        )
    if diff < -1:
        return 0.25, f"Seniority mismatch ({cv_seniority} CV for {job_seniority} role)."
    return (
        0.85,
        f"Candidate may be more experienced than required ({cv_seniority} → {job_seniority}).",
    )


def _experience_fit(cv_years: int, required_years: int) -> float:
    if required_years <= 0:
        return 1.0
    if cv_years <= 0:
        return 0.55 if required_years <= 1 else 0.35
    return max(0.1, min(1.0, cv_years / required_years))


def _soft_category_score(
    cv_skills: Set[str],
    job_category: str,
    category_skills: Dict[str, Set[str]],
) -> float:
    """
    Jaccard-based soft category match instead of binary 1.0 / 0.30.
    Scores range from 0.30 (no overlap) to 1.0 (exact match).
    """
    job_cat_skills = category_skills.get(job_category, set())
    if not job_cat_skills:
        return 0.30
    intersection = len(cv_skills & job_cat_skills)
    union = len(cv_skills | job_cat_skills)
    if union == 0:
        return 0.30
    jaccard = intersection / union
    return max(0.30, min(1.0, 0.30 + jaccard * 0.70))


def _match_level(score: float) -> str:
    if score >= 80:
        return "Excellent Match"
    if score >= 70:
        return "Strong Match"
    if score >= 55:
        return "Good Match"
    if score >= 40:
        return "Moderate Match"
    if score >= 25:
        return "Weak Match"
    return "Very Weak Match"


def _build_advice(
    missing_technical: List[str],
    job: JobInput,
    cv_category: str,
) -> List[str]:
    advice: List[str] = []
    if missing_technical:
        advice.append(
            "Add evidence for these technical skills if you have them: "
            + ", ".join(missing_technical[:5])
            + "."
        )
    if cv_category != job.category:
        advice.append(
            f"This CV looks closer to {cv_category}; tailor the summary and "
            f"projects toward {job.category} for this job."
        )
    if job.seniority == "senior":
        advice.append(
            "For senior roles, highlight leadership responsibilities, team size "
            "managed, and measurable achievements."
        )
    if not advice:
        advice.append(
            "The CV is well aligned; strengthen it with measurable achievements "
            "and recent work examples."
        )
    return advice[:4]


# ---------------------------------------------------------------------------
# Main class
# ---------------------------------------------------------------------------


class JobMatcher:
    """
    CV-to-job matcher using sentence embeddings + skill taxonomy scoring.

    Single-job usage
    ----------------
    matcher = JobMatcher()
    result  = matcher.predict(job_dict, cv_markdown)

    Batch usage (same CV, many jobs — CV processed only once)
    ---------------------------------------------------------
    results = matcher.predict_batch(list_of_job_dicts, cv_markdown)

    Performance notes (batched implementation)
    -------------------------------------------
    All model.encode() calls are batched ACROSS jobs, not per-job:
      - 1 call embeds every job's semantic text
      - 1 call embeds every job's title
      - 1 call embeds the (deduplicated) set of missing technical skills
        across *all* jobs, used for semantic skill fallback matching
      - The CV title vector is computed once and cached (was previously
        recomputed on every job).
    This means a request with N jobs makes a fixed ~4 model.encode() calls
    total instead of up to 4*N + 1, regardless of N. Scoring math and
    output are unchanged — this is purely a batching optimization.

    Model choice
    ------------
    Semantic model is BAAI/bge-small-en-v1.5 (33M params, standard BERT
    encoder architecture, 384-dim, 512-token context). This replaces
    jina-embeddings-v2-base-en (137M, custom ALiBi remote-code arch).
    The jina model's custom architecture does not reliably export via
    optimum's ONNX path, so on most boxes it was silently falling back to
    full PyTorch fp32 despite use_onnx=True. bge-small has published
    quantized ONNX weights matching the _ONNX_FILE_CANDIDATES naming
    below, ships an order of magnitude fewer parameters, and stays within
    a couple of points of jina-base on semantic similarity benchmarks —
    it does not meaningfully change match_score behavior for this use
    case since semantic similarity is only 23% of the final weighted
    score to begin with.

    Title model is UNCHANGED (TechWolf/JobBERT-v3, falling back to v2) —
    title matching is cheap (short strings) and this model is
    domain-specific for job titles, so there's no reason to swap it.

    Truncation was reduced from 4000 to 1500 chars (job text) / 2000
    chars (CV) — most of the matching signal lives in the first few
    hundred words, and this cuts attention cost roughly in half again on
    top of the smaller model, with negligible score impact in practice.

    Text is no longer manually chunked into overlapping ~180-word windows
    before embedding; the smaller model's 512-token window comfortably
    covers the reduced truncation length in a single forward pass.

    Both models are loaded via the ONNX Runtime backend (with int8
    quantization where a quantized export is available on the model repo)
    for faster CPU inference, with automatic fallback to plain PyTorch if
    ONNX packages or ONNX weights aren't available — see _load_model().
    The active backend is logged explicitly after load so a silent
    fallback to PyTorch is visible instead of assumed away.

    CV caching
    ----------
    Parsing + embedding a CV (cv_vec, cv_title_vec, individual_skills_vecs)
    is the fixed per-request cost that doesn't shrink with batching. It's
    cached in-process for a configurable TTL, keyed by a hash of the CV
    inputs, using cachetools.TTLCache (pure in-memory, no external
    service required, thread-safe via an internal lock). A repeat request
    for the same CV — e.g. pagination across job results — skips CV
    parsing and all three CV-side encode() calls entirely.
    """

    # Candidate ONNX file names to try, in preference order (most
    # aggressively quantized first). `None` means "let sentence-transformers
    # auto-select / auto-export the default ONNX model". Not every model
    # repo ships every variant, so we try each and move on if one 404s.
    _ONNX_FILE_CANDIDATES: List[Optional[str]] = [
        "onnx/model_qint8_avx512_vnni.onnx",
        "onnx/model_quantized.onnx",
        "onnx/model_qint8.onnx",
        "onnx/model_int8.onnx",
        "onnx/model.onnx",
        None,
    ]

    # Truncation lengths — kept as class constants so they're easy to tune
    # without hunting through the method bodies.
    _JOB_TEXT_MAX_CHARS: int = 1500
    _CV_TEXT_MAX_CHARS: int = 2000

    def __init__(
        self,
        semantic_model_name: str = "BAAI/bge-small-en-v1.5",
        title_model_name: str = "TechWolf/JobBERT-v2",
        skill_aliases: Optional[Dict[str, List[str]]] = None,
        category_skills: Optional[Dict[str, Set[str]]] = None,
        use_onnx: bool = True,
        cv_cache_maxsize: int = 256,
        cv_cache_ttl_seconds: int = 600,
    ) -> None:
        # bge-small does not need trust_remote_code — it's a standard
        # BERT-family architecture, which is exactly why it exports to
        # ONNX reliably (unlike jina's custom ALiBi remote code).
        self._model = self._load_model(
            semantic_model_name, use_onnx=use_onnx, trust_remote_code=False
        )
        try:
            self._title_model = self._load_model(
                "TechWolf/JobBERT-v3", use_onnx=use_onnx, trust_remote_code=False
            )
        except Exception:
            self._title_model = self._load_model(
                title_model_name, use_onnx=use_onnx, trust_remote_code=False
            )

        self._aliases: Dict[str, List[str]] = skill_aliases or DEFAULT_SKILL_ALIASES
        self._category_skills: Dict[str, Set[str]] = (
            category_skills or DEFAULT_CATEGORY_SKILLS
        )

        # In-process CV embedding cache. No external service, no setup —
        # just a bounded, TTL'd dict living in this process's memory.
        # Guarded by a lock since TTLCache itself is not thread-safe.
        self._cv_cache: TTLCache = TTLCache(
            maxsize=cv_cache_maxsize, ttl=cv_cache_ttl_seconds
        )
        self._cv_cache_lock = threading.RLock()

    # ------------------------------------------------------------------
    # Internal: load a SentenceTransformer via ONNX Runtime if possible,
    # trying quantized variants first, falling back to plain PyTorch.
    # ------------------------------------------------------------------

    @classmethod
    def _load_model(
        cls,
        model_name: str,
        use_onnx: bool = True,
        trust_remote_code: bool = False,
    ) -> SentenceTransformer:
        if use_onnx and _ONNXRUNTIME_AVAILABLE:
            session_options = None
            try:
                session_options = ort.SessionOptions()
                session_options.intra_op_num_threads = 2
                session_options.inter_op_num_threads = 1
            except Exception:
                session_options = None

            for file_name in cls._ONNX_FILE_CANDIDATES:
                try:
                    model_kwargs: Dict[str, Any] = {}
                    if file_name:
                        model_kwargs["file_name"] = file_name
                    if session_options is not None:
                        model_kwargs["session_options"] = session_options

                    model = SentenceTransformer(
                        model_name,
                        backend="onnx",
                        trust_remote_code=trust_remote_code,
                        model_kwargs=model_kwargs,
                    )
                    print(
                        f"[JobMatcher] Loaded '{model_name}' via ONNX Runtime "
                        f"(file={file_name or 'auto-export'})."
                    )
                    cls._log_active_backend(model, model_name)
                    return model
                except Exception as exc:
                    print(
                        f"[JobMatcher] ONNX load attempt failed for "
                        f"'{model_name}' (file={file_name or 'auto-export'}): {exc}"
                    )
        elif use_onnx and not _ONNXRUNTIME_AVAILABLE:
            print(
                "[JobMatcher] onnxruntime/optimum not installed — "
                f"falling back to PyTorch backend for '{model_name}'. "
                'Install with: pip install "optimum[onnxruntime]" onnxruntime'
            )

        print(f"[JobMatcher] Loading '{model_name}' via PyTorch backend.")
        model = SentenceTransformer(model_name, trust_remote_code=trust_remote_code)
        cls._log_active_backend(model, model_name)
        return model

    @staticmethod
    def _log_active_backend(model: SentenceTransformer, model_name: str) -> None:
        """
        Explicitly verify + log which backend actually ended up active.
        The try/except ladder in _load_model can silently fall through to
        PyTorch even when use_onnx=True, so this makes that visible
        instead of assumed.
        """
        backend = getattr(model, "backend", None) or "pytorch"
        print(f"[JobMatcher] '{model_name}' ACTIVE backend = {backend}")
        if backend != "onnx":
            print(
                f"[JobMatcher] NOTE: '{model_name}' is running on PyTorch, not ONNX. "
                "Check that 'optimum[onnxruntime]' and 'onnxruntime' are installed "
                "and that the model repo has quantized ONNX weights."
            )

    # ------------------------------------------------------------------
    # Public: single-job prediction (backward-compatible)
    # ------------------------------------------------------------------

    def predict(
        self,
        job_json: Dict[str, Any],
        cv_markdown: Optional[str] = None,
        cv_title: Optional[str] = None,
        cv_years: Optional[int] = None,
        cv_seniority: Optional[str] = None,
        cv_skills_canonical: Optional[List[str]] = None,
        cv_skills_technical: Optional[List[str]] = None,
        cv_skills_soft: Optional[List[str]] = None,
        cv_category: Optional[str] = None,
    ) -> Dict[str, Any]:
        """
        Predict the match between one job and one CV.

        Implemented as a thin wrapper around predict_batch([job]) so there
        is a single scoring code path to keep in sync.
        """
        results = self.predict_batch(
            jobs=[job_json],
            cv_markdown=cv_markdown or "",
            cv_title=cv_title or "Unknown",
            cv_years=cv_years or 0,
            cv_seniority=cv_seniority or "mid",
            cv_skills_canonical=cv_skills_canonical or [],
            cv_skills_technical=cv_skills_technical or [],
            cv_skills_soft=cv_skills_soft or [],
            cv_category=cv_category or "Other",
        )
        return results[0]

    # ------------------------------------------------------------------
    # Public: batch prediction (one CV vs. N jobs)
    # ------------------------------------------------------------------

    def predict_batch(
        self,
        jobs: List[Dict[str, Any]],
        cv_markdown: str,
        cv_title: str,
        cv_years: int,
        cv_seniority: str,
        cv_skills_canonical: List[str],
        cv_skills_technical: List[str],
        cv_skills_soft: List[str],
        cv_category: str,
    ) -> List[Dict[str, Any]]:
        """
        Match one CV against multiple jobs.

        The CV is parsed and embedded **once** (or fetched from cache if
        an identical CV was matched recently). All job-level embeddings
        (semantic text, title, missing-skill fallback) are computed in
        batched encode() calls across every job in the request, then
        scoring is done purely in numpy with no further model calls.
        """
        cv_data = self._get_or_parse_cv(
            cv_markdown=cv_markdown,
            cv_title=cv_title,
            cv_years=cv_years,
            cv_seniority=cv_seniority,
            cv_skills_canonical=cv_skills_canonical,
            cv_skills_technical=cv_skills_technical,
            cv_skills_soft=cv_skills_soft,
            cv_category=cv_category,
        )

        job_objs = [JobInput(**j) for j in jobs]
        n = len(job_objs)
        if n == 0:
            return []

        # ---- Batch 1: job semantic-text embeddings, ONE encode() call ----
        job_texts = [self._build_job_index_text(j) for j in job_objs]
        job_vecs = self._embed_batch_texts(job_texts)

        # ---- Batch 2: job title embeddings, ONE encode() call ----
        job_title_vecs = self._embed_title_batch([j.job_title for j in job_objs])

        # ---- Precompute exact skill matches + collect missing technical
        #      skills across ALL jobs for a single batched semantic pass ----
        cv_skills_technical_set = cv_data["cv_skills_technical"]
        per_job_technical: List[Set[str]] = []
        per_job_soft: List[Set[str]] = []
        per_job_matched_technical: List[List[str]] = []
        per_job_missing_technical: List[List[str]] = []

        all_missing_skills: Set[str] = set()
        for job in job_objs:
            job_technical = {
                s.strip().lower() for s in job.skills_technical if s.strip()
            }
            job_soft = {s.strip().lower() for s in job.skills_soft if s.strip()}
            matched = sorted(cv_skills_technical_set & job_technical)
            missing = sorted(job_technical - cv_skills_technical_set)
            per_job_technical.append(job_technical)
            per_job_soft.append(job_soft)
            per_job_matched_technical.append(matched)
            per_job_missing_technical.append(missing)
            all_missing_skills.update(missing)

        # ---- Batch 3: semantic skill-matching embeddings, ONE encode()
        #      call for the deduplicated union of missing skills across
        #      every job in the request ----
        missing_skill_vecs: Dict[str, np.ndarray] = {}
        individual_skills_vecs = cv_data["individual_skills_vecs"]
        if all_missing_skills and len(individual_skills_vecs) > 0:
            unique_missing = sorted(all_missing_skills)
            contextualized = [f"software development skill {s}" for s in unique_missing]
            vecs = self._model.encode(
                contextualized,
                normalize_embeddings=True,
                show_progress_bar=False,
                batch_size=32,
            )
            missing_skill_vecs = dict(zip(unique_missing, np.asarray(vecs)))

        # ---- Score each job — pure numpy from here, no model calls ----
        results = [
            self._score_job(
                job=job_objs[i],
                cv_data=cv_data,
                job_vec=job_vecs[i],
                job_title_vec=job_title_vecs[i],
                job_technical=per_job_technical[i],
                job_soft=per_job_soft[i],
                matched_technical=per_job_matched_technical[i],
                missing_technical=per_job_missing_technical[i],
                missing_skill_vecs=missing_skill_vecs,
            )
            for i in range(n)
        ]

        results.sort(
            key=lambda x: float(x["match_score"].rstrip("%")),
            reverse=True,
        )
        return results

    # ------------------------------------------------------------------
    # Internal: CV cache lookup / population
    # ------------------------------------------------------------------

    def _cv_cache_key(
        self,
        cv_markdown: str,
        cv_title: str,
        cv_years: int,
        cv_seniority: str,
        cv_skills_canonical: List[str],
        cv_skills_technical: List[str],
        cv_skills_soft: List[str],
        cv_category: str,
    ) -> str:
        """
        Deterministic hash of every input that affects CV parsing/embedding.
        Truncate the markdown to the same length actually used for
        embedding so two CVs differing only past the truncation point
        still hit the same cache entry.
        """
        payload = {
            "cv_markdown": (cv_markdown or "")[: self._CV_TEXT_MAX_CHARS],
            "cv_title": cv_title or "",
            "cv_years": cv_years,
            "cv_seniority": cv_seniority,
            "cv_skills_canonical": sorted(cv_skills_canonical or []),
            "cv_skills_technical": sorted(cv_skills_technical or []),
            "cv_skills_soft": sorted(cv_skills_soft or []),
            "cv_category": cv_category or "",
        }
        blob = json.dumps(payload, sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(blob.encode("utf-8")).hexdigest()

    def _get_or_parse_cv(
        self,
        cv_markdown: str,
        cv_title: str,
        cv_years: int,
        cv_seniority: str,
        cv_skills_canonical: List[str],
        cv_skills_technical: List[str],
        cv_skills_soft: List[str],
        cv_category: str,
    ) -> Dict[str, Any]:
        cache_key = self._cv_cache_key(
            cv_markdown=cv_markdown,
            cv_title=cv_title,
            cv_years=cv_years,
            cv_seniority=cv_seniority,
            cv_skills_canonical=cv_skills_canonical,
            cv_skills_technical=cv_skills_technical,
            cv_skills_soft=cv_skills_soft,
            cv_category=cv_category,
        )

        with self._cv_cache_lock:
            cached = self._cv_cache.get(cache_key)
            if cached is not None:
                return cached

        cv_data = self._parse_cv(
            cv_markdown=cv_markdown,
            cv_title=cv_title,
            cv_years=cv_years,
            cv_seniority=cv_seniority,
            cv_skills_canonical=cv_skills_canonical,
            cv_skills_technical=cv_skills_technical,
            cv_skills_soft=cv_skills_soft,
            cv_category=cv_category,
        )

        with self._cv_cache_lock:
            self._cv_cache[cache_key] = cv_data

        return cv_data

    # ------------------------------------------------------------------
    # Internal: parse + embed CV (done once per unique CV, then cached)
    # ------------------------------------------------------------------

    def _parse_cv(
        self,
        cv_markdown: str,
        cv_title: str,
        cv_years: int,
        cv_seniority: str,
        cv_skills_canonical: List[str],
        cv_skills_technical: List[str],
        cv_skills_soft: List[str],
        cv_category: str,
    ) -> Dict[str, Any]:
        """
        Return a dict holding all CV-level data needed by _score_job.
        """
        cv_skills_set = set(cv_skills_canonical)
        cv_skills_technical_set = set(cv_skills_technical)
        cv_skills_soft_set = set(cv_skills_soft)

        cv_vec = (
            self._embed(cv_markdown[: self._CV_TEXT_MAX_CHARS])
            if cv_markdown
            else np.zeros(384, dtype="float32")
        )

        # Step 1: compute the CV title vector once here instead of
        # recomputing it inside every per-job title-similarity call.
        cv_title_vec = (
            self._embed_title(cv_title) if cv_title and cv_title != "Unknown" else None
        )

        individual_skills_list = list(cv_skills_technical_set | cv_skills_soft_set)
        if individual_skills_list:
            contextualized_cv = [
                f"software development skill {s}" for s in individual_skills_list
            ]
            individual_skills_vecs = self._model.encode(
                contextualized_cv,
                normalize_embeddings=True,
                show_progress_bar=False,
                batch_size=32,
            )
        else:
            individual_skills_vecs = np.array([])

        return {
            "cv_title": cv_title,
            "cv_title_vec": cv_title_vec,
            "cv_years": cv_years,
            "cv_seniority": cv_seniority,
            "cv_skills": cv_skills_set,
            "cv_skills_technical": cv_skills_technical_set,
            "cv_skills_soft": cv_skills_soft_set,
            "cv_category": cv_category,
            "cv_vec": cv_vec,
            "individual_skills_list": individual_skills_list,
            "individual_skills_vecs": individual_skills_vecs,
        }

    # ------------------------------------------------------------------
    # Internal: score one job against pre-parsed CV data + precomputed
    # job-level vectors (no model calls happen in this method).
    # ------------------------------------------------------------------

    def _score_job(
        self,
        job: JobInput,
        cv_data: Dict[str, Any],
        job_vec: np.ndarray,
        job_title_vec: np.ndarray,
        job_technical: Set[str],
        job_soft: Set[str],
        matched_technical: List[str],
        missing_technical: List[str],
        missing_skill_vecs: Dict[str, np.ndarray],
    ) -> Dict[str, Any]:
        cv_title = cv_data["cv_title"]
        cv_title_vec = cv_data["cv_title_vec"]
        cv_years = cv_data["cv_years"]
        cv_seniority = cv_data["cv_seniority"]
        cv_skills = cv_data["cv_skills"]
        cv_skills_soft = cv_data["cv_skills_soft"]
        cv_category = cv_data["cv_category"]
        cv_vec = cv_data["cv_vec"]
        individual_skills_vecs = cv_data["individual_skills_vecs"]

        # ── Semantic similarity (job embedding precomputed in the batch) ──
        semantic = max(0.0, min(1.0, _cosine_similarity(cv_vec, job_vec)))

        # ── Skill scoring ───────────────────────────────────────────────
        matched_technical = list(matched_technical)
        missing_technical = list(missing_technical)

        # Semantic fallback for missing technical skills, using the
        # precomputed (batched, deduplicated) missing-skill vector lookup.
        if missing_technical and len(individual_skills_vecs) > 0 and missing_skill_vecs:
            semantic_matches = self._semantic_skill_matches_precomputed(
                missing_technical,
                missing_skill_vecs,
                individual_skills_vecs,
                threshold=0.88,
            )
            if semantic_matches:
                matched_technical = sorted(set(matched_technical) | semantic_matches)
                missing_technical = sorted(set(missing_technical) - semantic_matches)

        matched_soft = sorted(cv_skills_soft & job_soft)
        missing_soft = sorted(job_soft - cv_skills_soft)

        extra = sorted(cv_skills - job_technical - job_soft)[:8]

        tech_denom = min(len(job_technical), 12) if job_technical else 1
        soft_denom = min(len(job_soft), 3) if job_soft else 1

        tech_score = (
            min(1.0, len(matched_technical) / tech_denom) if job_technical else 0.50
        )
        soft_score = min(1.0, len(matched_soft) / soft_denom) if job_soft else 0.50

        skill = min(1.0, (tech_score * 0.80) + (soft_score * 0.20))

        # ── Component scores ─────────────────────────────────────────────
        cv_cat_lower = cv_category.strip().lower() if cv_category else ""
        job_cat_lower = job.category.strip().lower() if job.category else ""

        if cv_cat_lower and job_cat_lower and cv_cat_lower == job_cat_lower:
            cat_score = 1.0
        else:
            cat_score = _soft_category_score(
                cv_skills, job.category, self._category_skills
            )

        seniority, sen_reason = _seniority_fit(cv_seniority, job.seniority)
        title = self._title_similarity_precomputed(
            cv_title, cv_title_vec, job_title_vec
        )
        experience = _experience_fit(cv_years, job.min_years_experience)

        # ── Weighted aggregation ─────────────────────────────────────────
        base_final = (
            0.42 * skill
            + 0.23 * semantic
            + 0.12 * title
            + 0.11 * cat_score
            + 0.07 * seniority
            + 0.05 * experience
        ) * 100

        # Smooth Modifier applies universally
        if cat_score < 0.75:
            cat_modifier = -20.0 + ((cat_score - 0.30) / 0.45) * 20.0
            cat_modifier = max(-20.0, min(0.0, cat_modifier))
        else:
            cat_modifier = ((cat_score - 0.75) / 0.25) * 10.0
            cat_modifier = max(0.0, min(10.0, cat_modifier))

        final = base_final + cat_modifier

        # Guardrails

        if job_technical and not matched_technical:
            final = min(final, 34.0)
        if cv_cat_lower != job_cat_lower and skill < 0.30:
            final = min(final, 42.0)
        if (
            job.seniority == "senior"
            and SENIORITY_ORDER.get(cv_seniority, 1) < SENIORITY_ORDER["mid"]
        ):
            final = min(final, 45.0)

        final = round(max(0.0, min(100.0, final)), 2)
        level = _match_level(final)

        # ── Reason string ────────────────────────────────────────────────
        if matched_technical or matched_soft:
            reason = (
                f"{level}: matched {len(matched_technical)}/{len(job_technical)} technical skill(s) and "
                f"{len(matched_soft)}/{len(job_soft)} soft skill(s)."
            )
        else:
            reason = (
                f"{level}: the CV does not show the main technical skills for this "
                f"role. It looks closer to {cv_category}."
            )

        prediction = MatchPrediction(
            job_id=job.job_id,
            job_title=job.job_title,
            company_name=job.company_name,
            match_score=f"{final}%",
            match_level=level,
            cv_title=cv_title,
            cv_seniority=cv_seniority,
            cv_category=cv_category,
            cv_skills=sorted(cv_skills),
            cv_years_experience=cv_years,
            score_breakdown=ScoreBreakdown(
                skill_score=round(skill * 100, 2),
                semantic_score=round(semantic * 100, 2),
                title_score=round(title * 100, 2),
                category_score=round(cat_score * 100, 2),
                seniority_score=round(seniority * 100, 2),
                experience_score=round(experience * 100, 2),
                category_modifier=round(cat_modifier, 2),
            ),
            mismatch_analysis=MismatchAnalysis(
                reason=reason,
                matched_technical_skills=matched_technical,
                missing_technical_skills=missing_technical[:10],
                matched_soft_skills=matched_soft,
                missing_soft_skills=missing_soft[:10],
                cv_extra_strengths=extra,
                seniority_fit=sen_reason,
                category_fit=f"CV category: {cv_category}; Job category: {job.category}.",
                improvement_advice=_build_advice(missing_technical, job, cv_category),
            ),
        )
        return prediction.model_dump()

    # ------------------------------------------------------------------
    # Private helpers
    # ------------------------------------------------------------------

    def _build_job_index_text(self, job: JobInput) -> str:
        """Build the semantic index text for a job (same as before, factored out)."""
        text = (
            f"{job.job_title}\nCompany: {job.company_name}\n{job.description}\n"
            f"Technical skills: {', '.join(job.skills_technical)}\n"
            f"Soft skills: {', '.join(job.skills_soft)}"
        )
        return text[: self._JOB_TEXT_MAX_CHARS]

    def _embed_batch_texts(self, texts: List[str]) -> np.ndarray:
        """
        Embed a list of texts in a SINGLE batched encode() call.

        No manual word-chunking: bge-small-en-v1.5 supports sequences up
        to 512 tokens, and job text is already truncated to
        _JOB_TEXT_MAX_CHARS chars upstream, so it fits in one forward
        pass per item.
        """
        if not texts:
            return np.zeros((0, 384), dtype="float32")

        safe_texts = [t if t else "" for t in texts]
        vectors = self._model.encode(
            safe_texts,
            normalize_embeddings=True,
            show_progress_bar=False,
            batch_size=32,
        )
        return np.asarray(vectors, dtype="float32")

    def _embed(self, text: str) -> np.ndarray:
        """Encode a single text → normalised float32 vector (single forward pass)."""
        return self._embed_batch_texts([text])[0]

    def _embed_title_batch(self, titles: List[str]) -> np.ndarray:
        """Encode a list of titles in ONE batched encode() call with the title model."""
        if not titles:
            return np.zeros((0, 384), dtype="float32")
        vectors = self._title_model.encode(
            titles, normalize_embeddings=True, show_progress_bar=False, batch_size=32
        )
        return np.asarray(vectors).astype("float32")

    def _embed_title(self, text: str) -> np.ndarray:
        """Encode a single title using the title model (used once per CV)."""
        vector = self._title_model.encode(
            text, normalize_embeddings=True, show_progress_bar=False
        )
        return vector.astype("float32")

    def _semantic_skill_matches_precomputed(
        self,
        missing_skills: List[str],
        missing_skill_vecs: Dict[str, np.ndarray],
        cv_skills_vecs: np.ndarray,
        threshold: float = 0.72,
    ) -> Set[str]:
        """
        Same semantics as the original _semantic_skill_matches, but takes
        precomputed vectors for the missing skills (built once per request
        in a single batched encode() call in predict_batch) instead of
        encoding them again per job.
        """
        if not missing_skills or len(cv_skills_vecs) == 0 or not missing_skill_vecs:
            return set()

        matched: Set[str] = set()
        for skill in missing_skills:
            vec = missing_skill_vecs.get(skill)
            if vec is None:
                continue
            sims = np.dot(cv_skills_vecs, vec)
            if np.max(sims) >= threshold:
                matched.add(skill)
        return matched

    def _title_similarity_precomputed(
        self,
        cv_title: str,
        cv_title_vec: Optional[np.ndarray],
        job_title_vec: np.ndarray,
    ) -> float:
        """
        Same semantics as the original _title_similarity, but uses the
        CV title vector cached once in cv_data and the job title vector
        computed once per request in the batched title encode() call —
        no model calls happen here.
        """
        if not cv_title or cv_title == "Unknown" or cv_title_vec is None:
            return 0.35
        return max(0.2, min(1.0, _cosine_similarity(cv_title_vec, job_title_vec)))