apps/accounts/ner/skillner.py

"""Layer 3: skillNer — rule-based EMSI-taxonomy matcher.

https://github.com/AnasAito/SkillNER — Nov 2021 last release, pure Python,
works on any spaCy `nlp` object (no version pin). Uses spaCy's PhraseMatcher
internally so CPU-only and fast.

Gracefully unavailable if its transitive deps (nltk, jellyfish) fail to
import — the dispatcher's try/except continues with the next layer.
"""
from __future__ import annotations

import logging

from . import NerLayer


logger = logging.getLogger(__name__)

_extractor = None
_nlp = None
_import_error = None


def _get_extractor():
    """Lazy singleton. Import both spaCy + skillNer on first call."""
    global _extractor, _nlp, _import_error
    if _extractor is not None:
        return _extractor
    if _import_error is not None:
        # Already attempted and failed — surface the original error so the
        # dispatcher can log once instead of re-throwing noisily.
        raise _import_error

    try:
        import spacy
        from skillNer.skill_extractor_class import SkillExtractor
        from skillNer.general_params import SKILL_DB
        from spacy.matcher import PhraseMatcher
    except Exception as exc:  # pragma: no cover — env-specific
        _import_error = exc
        logger.warning("skillner: unavailable (import failed): %s", exc)
        raise

    # Prefer `_lg` (word vectors help edge cases) but fall back to `_sm` if
    # only the smaller model is installed. skillner uses spaCy's PhraseMatcher
    # which is string-based, so both work; `_lg` is strictly better-quality
    # when RAM / disk allow.
    try:
        _nlp = spacy.load("en_core_web_lg")
    except OSError:
        _nlp = spacy.load("en_core_web_sm")
    # Pass the PhraseMatcher *class*, not an instance: skillNer constructs its
    # own matcher internally via `self.phraseMatcher(nlp.vocab, attr="LOWER")`,
    # so handing it an instance raises at construction and disables the layer.
    _extractor = SkillExtractor(_nlp, SKILL_DB, PhraseMatcher)
    logger.info("skillner: loaded (spaCy=%s, SKILL_DB=%d entries)",
                spacy.__version__, len(SKILL_DB))
    return _extractor


class SkillNerLayer(NerLayer):
    name = "skillner"

    def predict(self, text: str) -> dict[str, float]:
        if not text.strip():
            return {}
        extractor = _get_extractor()
        annotations = extractor.annotate(text[:4000])
        out: dict[str, float] = {}
        results = (annotations or {}).get("results") or {}
        # full_matches: perfect catalog hits — high confidence.
        for match in results.get("full_matches") or []:
            term = (match.get("doc_node_value") or "").strip()
            if term:
                out[term] = max(out.get(term, 0.0), 0.95)
        # ngram_scored: fuzzy catalog hits — skillNer includes its own score.
        for match in results.get("ngram_scored") or []:
            term = (match.get("doc_node_value") or "").strip()
            score = float(match.get("score") or 0.0)
            if term:
                out[term] = max(out.get(term, 0.0), score)
        return out

    def available(self) -> bool:
        if _import_error is not None:
            return False
        try:
            import skillNer  # noqa: F401
            import spacy  # noqa: F401
            return True
        except Exception:
            return False


layer = SkillNerLayer()