"""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()