diff --git a/presidio-analyzer/install_nlp_models.py b/presidio-analyzer/install_nlp_models.py index 268d4c9cb7..767f5a8b8e 100644 --- a/presidio-analyzer/install_nlp_models.py +++ b/presidio-analyzer/install_nlp_models.py @@ -98,6 +98,8 @@ def _download_model(engine_name: str, model_name: Union[str, Dict[str, str]]) -> _install_transformers_spacy_models(model_name) else: raise ImportError("transformers is not installed") + elif engine_name == "no_op": + logger.info("No NLP model installation required for no_op engine") else: raise ValueError(f"Unsupported nlp engine: {engine_name}") diff --git a/presidio-analyzer/presidio_analyzer/analyzer_engine.py b/presidio-analyzer/presidio_analyzer/analyzer_engine.py index b68d24d6ab..d05a060236 100644 --- a/presidio-analyzer/presidio_analyzer/analyzer_engine.py +++ b/presidio-analyzer/presidio_analyzer/analyzer_engine.py @@ -1,6 +1,7 @@ import json import logging import os +import warnings from collections import Counter from typing import List, Optional @@ -15,7 +16,12 @@ ContextAwareEnhancer, LemmaContextAwareEnhancer, ) -from presidio_analyzer.nlp_engine import NlpArtifacts, NlpEngine, NlpEngineProvider +from presidio_analyzer.nlp_engine import ( + NlpArtifacts, + NlpEngine, + NlpEngineProvider, + NoOpNlpEngine, +) from presidio_analyzer.recognizer_registry import ( RecognizerRegistry, RecognizerRegistryProvider, @@ -97,12 +103,24 @@ def __init__( registry.load_predefined_recognizers( nlp_engine=self.nlp_engine, languages=self.supported_languages ) + registry.validate_nlp_engine_compatibility(self.nlp_engine) self.registry = registry self.log_decision_process = log_decision_process self.default_score_threshold = default_score_threshold + if isinstance(self.nlp_engine, NoOpNlpEngine) and isinstance( + context_aware_enhancer, LemmaContextAwareEnhancer + ): + warnings.warn( + "LemmaContextAwareEnhancer cannot use context words from the analyzed " + "text with NoOpNlpEngine because no tokens or lemmas are produced. " + "Only context passed explicitly to analyze() can be used.", + UserWarning, + stacklevel=2, + ) + if not context_aware_enhancer: logger.debug( "context aware enhancer not provided, creating default" diff --git a/presidio-analyzer/presidio_analyzer/conf/no_op.yaml b/presidio-analyzer/presidio_analyzer/conf/no_op.yaml new file mode 100644 index 0000000000..0b2a6c7adf --- /dev/null +++ b/presidio-analyzer/presidio_analyzer/conf/no_op.yaml @@ -0,0 +1,4 @@ +nlp_engine_name: no_op +models: + - lang_code: en + model_name: no_op diff --git a/presidio-analyzer/presidio_analyzer/nlp_engine/__init__.py b/presidio-analyzer/presidio_analyzer/nlp_engine/__init__.py index 575593950a..bafb380b65 100644 --- a/presidio-analyzer/presidio_analyzer/nlp_engine/__init__.py +++ b/presidio-analyzer/presidio_analyzer/nlp_engine/__init__.py @@ -4,6 +4,7 @@ from .ner_model_configuration import NerModelConfiguration from .nlp_artifacts import NlpArtifacts from .nlp_engine import NlpEngine +from .no_op_nlp_engine import NoOpNlpEngine from .slim_spacy_nlp_engine import SlimSpacyNlpEngine from .spacy_nlp_engine import SpacyNlpEngine from .stanza_nlp_engine import StanzaNlpEngine @@ -16,6 +17,7 @@ "NerModelConfiguration", "NlpArtifacts", "NlpEngine", + "NoOpNlpEngine", "SlimSpacyNlpEngine", "SpacyNlpEngine", "StanzaNlpEngine", diff --git a/presidio-analyzer/presidio_analyzer/nlp_engine/nlp_engine_provider.py b/presidio-analyzer/presidio_analyzer/nlp_engine/nlp_engine_provider.py index 6b3b98c6ca..1848794c41 100644 --- a/presidio-analyzer/presidio_analyzer/nlp_engine/nlp_engine_provider.py +++ b/presidio-analyzer/presidio_analyzer/nlp_engine/nlp_engine_provider.py @@ -8,6 +8,7 @@ from presidio_analyzer.nlp_engine import ( NerModelConfiguration, NlpEngine, + NoOpNlpEngine, SlimSpacyNlpEngine, SpacyNlpEngine, StanzaNlpEngine, @@ -21,7 +22,8 @@ class NlpEngineProvider: """Create different NLP engines from configuration. :param nlp_engines: List of available NLP engines - Default: (SpacyNlpEngine, StanzaNlpEngine) + Default: (SpacyNlpEngine, StanzaNlpEngine, TransformersNlpEngine, + SlimSpacyNlpEngine, NoOpNlpEngine) :param nlp_configuration: Dict containing nlp configuration :example: configuration: { @@ -30,7 +32,7 @@ class NlpEngineProvider: "model_name": "en_core_web_lg" }] } - Nlp engine names available by default: spacy, stanza. + Nlp engine names available by default: spacy, stanza, transformers, slim, no_op. :param conf_file: Path to yaml file containing nlp engine configuration. """ @@ -46,6 +48,7 @@ def __init__( StanzaNlpEngine, TransformersNlpEngine, SlimSpacyNlpEngine, + NoOpNlpEngine, ) self.nlp_engines = { @@ -102,7 +105,9 @@ def create_engine(self) -> NlpEngine: nlp_engine_class = self.nlp_engines[nlp_engine_name] nlp_models = self.nlp_configuration["models"] - if nlp_engine_name == SlimSpacyNlpEngine.engine_name: + if nlp_engine_name == NoOpNlpEngine.engine_name: + engine = nlp_engine_class(models=nlp_models) + elif nlp_engine_name == SlimSpacyNlpEngine.engine_name: generic_tokenizer = self.nlp_configuration.get("generic_tokenizer") engine = nlp_engine_class( models=nlp_models, generic_tokenizer=generic_tokenizer diff --git a/presidio-analyzer/presidio_analyzer/nlp_engine/no_op_nlp_engine.py b/presidio-analyzer/presidio_analyzer/nlp_engine/no_op_nlp_engine.py new file mode 100644 index 0000000000..cfb016143a --- /dev/null +++ b/presidio-analyzer/presidio_analyzer/nlp_engine/no_op_nlp_engine.py @@ -0,0 +1,168 @@ +import logging +from typing import Any, Dict, Iterable, Iterator, List, Tuple, Union + +from spacy.tokens import Doc +from spacy.vocab import Vocab + +from presidio_analyzer.nlp_engine import NlpArtifacts, NlpEngine + +logger = logging.getLogger("presidio-analyzer") + + +class NoOpNlpEngine(NlpEngine): + """An NLP engine that performs no linguistic processing. + + This engine is intended for AnalyzerEngine configurations where all active + recognizers are self-contained and do not require NLP artifacts. It loads no + external model and returns empty artifacts for each analyzed text. + The default LemmaContextAwareEnhancer can use context passed explicitly to + ``AnalyzerEngine.analyze()``, but cannot use context words from the analyzed + text because this engine produces no tokens or lemmas. + """ + + engine_name = "no_op" + is_available = True + + def __init__( + self, + models: List[Dict[str, str]], + ): + self.models = self._validate_models(models) + self._vocab = Vocab() + self.nlp = None + + def load(self) -> None: + """Mark configured languages as loaded without loading external models.""" + nlp = {} + for model in self.models: + nlp[model["lang_code"]] = None + + self.nlp = nlp + logger.info("Loaded no-op NLP engine with languages: %s", list(self.nlp.keys())) + + @classmethod + def _validate_models(cls, models: List[Dict[str, str]]) -> List[Dict[str, str]]: + if not isinstance(models, list) or not models: + raise ValueError("models must be a non-empty list") + + seen_languages = set() + validated_models = [] + for model in models: + validated_model = cls._validate_model_params(model) + language = validated_model["lang_code"] + if language in seen_languages: + raise ValueError( + f"Duplicate language '{language}' in no-op NLP engine models" + ) + seen_languages.add(language) + validated_models.append(validated_model) + + return validated_models + + @classmethod + def _validate_model_params(cls, model: Dict) -> Dict[str, str]: + """Validate that required model parameters are present.""" + if not isinstance(model, dict): + raise ValueError("Each model must be a dictionary") + if "lang_code" not in model: + raise ValueError("lang_code is missing from model configuration") + if "model_name" not in model: + raise ValueError("model_name is missing from model configuration") + if not isinstance(model["lang_code"], str) or not model["lang_code"]: + raise ValueError("lang_code must be a non-empty string") + if model["lang_code"] != model["lang_code"].strip(): + raise ValueError("lang_code must not contain leading or trailing spaces") + if not isinstance(model["model_name"], str): + raise ValueError("model_name must be a string") + return {"lang_code": model["lang_code"], "model_name": model["model_name"]} + + def is_loaded(self) -> bool: + """Return True if the no-op engine has been initialized.""" + return self.nlp is not None + + def process_text(self, text: str, language: str) -> NlpArtifacts: + """Return empty NLP artifacts for the given text and language.""" + self._validate_string(text, "text") + self._validate_loaded_language(language) + return self._create_empty_nlp_artifacts(language) + + def process_batch( + self, + texts: Iterable[Union[str, Tuple[str, Any]]], + language: str, + batch_size: int = 1, + n_process: int = 1, + as_tuples: bool = False, + **kwargs, + ) -> Iterator[Union[Tuple[str, NlpArtifacts], Tuple[str, NlpArtifacts, Any]]]: + """Return empty NLP artifacts for each text in a batch.""" + if kwargs: + logger.warning( + "Ignoring unsupported kwargs in NoOpNlpEngine.process_batch: %s", + sorted(kwargs.keys()), + ) + self._validate_loaded_language(language) + + for item in texts: + if as_tuples: + if not isinstance(item, tuple) or len(item) != 2: + raise ValueError( + "When 'as_tuples' is True, " + "'texts' must be an iterable of tuples (text, context)." + ) + text, context = item + text = str(text) + yield text, self._create_empty_nlp_artifacts(language), context + else: + text = str(item) + yield text, self._create_empty_nlp_artifacts(language) + + def is_stopword(self, word: str, language: str) -> bool: + """Return False because no stop-word vocabulary is available.""" + self._validate_string(word, "word") + self._validate_loaded_language(language) + return False + + def is_punct(self, word: str, language: str) -> bool: + """Return False because no punctuation vocabulary is available.""" + self._validate_string(word, "word") + self._validate_loaded_language(language) + return False + + def get_supported_entities(self) -> List[str]: + """Return no entities because this engine does not run NER.""" + return [] + + def get_supported_languages(self) -> List[str]: + """Return the languages configured for this no-op engine.""" + if not self.nlp: + raise ValueError("NLP engine is not loaded. Consider calling .load()") + return list(self.nlp.keys()) + + def _validate_loaded_language(self, language: str) -> None: + self._validate_string(language, "language") + if not self.nlp: + raise ValueError("NLP engine is not loaded. Consider calling .load()") + + if language not in self.nlp: + raise ValueError( + f"Language '{language}' is not supported by this NLP engine. " + f"Supported languages: {list(self.nlp.keys())}" + ) + + @staticmethod + def _validate_string(value: Any, parameter_name: str) -> str: + if not isinstance(value, str): + raise TypeError(f"{parameter_name} must be a string") + return value + + def _create_empty_nlp_artifacts(self, language: str) -> NlpArtifacts: + return NlpArtifacts( + entities=[], + tokens=Doc(self._vocab, words=[]), + tokens_indices=[], + lemmas=[], + nlp_engine=self, + language=language, + scores=[], + ) diff --git a/presidio-analyzer/presidio_analyzer/predefined_recognizers/ner/huggingface_ner_recognizer.py b/presidio-analyzer/presidio_analyzer/predefined_recognizers/ner/huggingface_ner_recognizer.py index 43b17683f1..ba544b53ca 100644 --- a/presidio-analyzer/presidio_analyzer/predefined_recognizers/ner/huggingface_ner_recognizer.py +++ b/presidio-analyzer/presidio_analyzer/predefined_recognizers/ner/huggingface_ner_recognizer.py @@ -55,23 +55,24 @@ class HuggingFaceNerRecognizer(LocalRecognizer): Example: >>> from presidio_analyzer import AnalyzerEngine + >>> from presidio_analyzer.nlp_engine import NlpEngineProvider >>> from presidio_analyzer.predefined_recognizers.ner import ( ... HuggingFaceNerRecognizer ... ) >>> - >>> # For Korean - >>> recognizer = HuggingFaceNerRecognizer( - ... model_name="test-owner/test-model", - ... supported_language="ko" - ... ) - >>> - >>> # For English >>> recognizer = HuggingFaceNerRecognizer( ... model_name="dslim/bert-base-NER", ... supported_language="en" ... ) - >>> - >>> analyzer = AnalyzerEngine() + >>> nlp_engine = NlpEngineProvider( + ... nlp_configuration={ + ... "nlp_engine_name": "no_op", + ... "models": [{"lang_code": "en", "model_name": "no_op"}], + ... } + ... ).create_engine() + >>> analyzer = AnalyzerEngine( + ... nlp_engine=nlp_engine, supported_languages=["en"] + ... ) >>> analyzer.registry.add_recognizer(recognizer) """ diff --git a/presidio-analyzer/presidio_analyzer/recognizer_registry/recognizer_registry.py b/presidio-analyzer/presidio_analyzer/recognizer_registry/recognizer_registry.py index 059f194f8a..fd056a7dc9 100644 --- a/presidio-analyzer/presidio_analyzer/recognizer_registry/recognizer_registry.py +++ b/presidio-analyzer/presidio_analyzer/recognizer_registry/recognizer_registry.py @@ -9,6 +9,7 @@ from presidio_analyzer import EntityRecognizer, PatternRecognizer from presidio_analyzer.nlp_engine import ( NlpEngine, + NoOpNlpEngine, SpacyNlpEngine, StanzaNlpEngine, TransformersNlpEngine, @@ -53,6 +54,23 @@ def __init__( supported_languages if supported_languages else ["en"] ) + def validate_nlp_engine_compatibility( + self, nlp_engine: Optional[NlpEngine] + ) -> None: + """Validate that registered recognizers can use the selected NLP engine.""" + if not isinstance(nlp_engine, NoOpNlpEngine): + return + + nlp_recognizers = [ + rec for rec in self.recognizers if isinstance(rec, SpacyRecognizer) + ] + if nlp_recognizers: + names = sorted({rec.__class__.__name__ for rec in nlp_recognizers}) + raise ValueError( + "NoOpNlpEngine cannot be used with NLP engine recognizers. " + f"Remove or disable these recognizers: {names}." + ) + def _create_nlp_recognizer( self, nlp_engine: Optional[NlpEngine] = None, @@ -76,6 +94,11 @@ def add_nlp_recognizer(self, nlp_engine: NlpEngine) -> None: :return: None """ + if isinstance(nlp_engine, NoOpNlpEngine): + self.validate_nlp_engine_compatibility(nlp_engine) + logger.info("Skipping NLP recognizer registration for no-op NLP engine.") + return + if not nlp_engine: supported_languages = self.supported_languages else: @@ -148,6 +171,8 @@ def get_nlp_recognizer( return StanzaRecognizer if isinstance(nlp_engine, TransformersNlpEngine): return TransformersRecognizer + if isinstance(nlp_engine, NoOpNlpEngine): + raise ValueError("NoOpNlpEngine does not have an NLP recognizer") if not nlp_engine or isinstance(nlp_engine, SpacyNlpEngine): return SpacyRecognizer else: diff --git a/presidio-analyzer/presidio_analyzer/recognizer_registry/recognizer_registry_provider.py b/presidio-analyzer/presidio_analyzer/recognizer_registry/recognizer_registry_provider.py index 143f825133..cfebe00def 100644 --- a/presidio-analyzer/presidio_analyzer/recognizer_registry/recognizer_registry_provider.py +++ b/presidio-analyzer/presidio_analyzer/recognizer_registry/recognizer_registry_provider.py @@ -7,7 +7,7 @@ from presidio_analyzer import EntityRecognizer from presidio_analyzer.input_validation import ConfigurationValidator -from presidio_analyzer.nlp_engine import NlpEngine +from presidio_analyzer.nlp_engine import NlpEngine, NoOpNlpEngine from presidio_analyzer.predefined_recognizers import SpacyRecognizer from presidio_analyzer.recognizer_registry import RecognizerRegistry from presidio_analyzer.recognizer_registry.recognizers_loader_utils import ( @@ -86,6 +86,7 @@ def create_recognizer_registry(self) -> RecognizerRegistry: supported_languages=supported_languages, global_regex_flags=global_regex_flags, ) + registry.validate_nlp_engine_compatibility(self.nlp_engine) return registry @@ -119,6 +120,12 @@ def __update_based_on_nlp_recognizer_conf( if not nlp_engine: return + if isinstance(nlp_engine, NoOpNlpEngine): + logger.info( + "Skipping NLP recognizer configuration updates for no-op NLP engine." + ) + return + for language in nlp_engine.get_supported_languages(): self.__update_based_on_nlp_recognizer_conf_and_lang( recognizers=recognizers, nlp_engine=nlp_engine, language=language diff --git a/presidio-analyzer/tests/conftest.py b/presidio-analyzer/tests/conftest.py index 890d603417..b2d1d3d521 100644 --- a/presidio-analyzer/tests/conftest.py +++ b/presidio-analyzer/tests/conftest.py @@ -57,6 +57,10 @@ def nlp_engines(request, nlp_engine_provider) -> Dict[str, NlpEngine]: available_engines[f"{name}_en"] = engine_cls( models=[{"lang_code": "en", "model_name": "en_core_web_sm"}] ) + elif name == "no_op": + available_engines[f"{name}_en"] = engine_cls( + models=[{"lang_code": "en", "model_name": "no_op"}] + ) else: raise ValueError("Unsupported engine for tests") diff --git a/presidio-analyzer/tests/test_no_op_nlp_engine.py b/presidio-analyzer/tests/test_no_op_nlp_engine.py new file mode 100644 index 0000000000..c6a979bbdc --- /dev/null +++ b/presidio-analyzer/tests/test_no_op_nlp_engine.py @@ -0,0 +1,432 @@ +from collections.abc import Iterator + +import pytest +from spacy.tokens import Doc + +from presidio_analyzer import ( + AnalyzerEngine, + AnalyzerEngineProvider, + BatchAnalyzerEngine, + LemmaContextAwareEnhancer, + Pattern, + PatternRecognizer, +) +from presidio_analyzer.nlp_engine import NlpEngineProvider, NoOpNlpEngine +from presidio_analyzer.predefined_recognizers import SpacyRecognizer +from presidio_analyzer.recognizer_registry import RecognizerRegistry +from install_nlp_models import _download_model + + +@pytest.fixture +def no_op_nlp_engine(): + engine = NoOpNlpEngine(models=[{"lang_code": "en", "model_name": "no_op"}]) + engine.load() + return engine + + +class TestNoOpNlpEngine: + def test_when_init_without_models_then_raises(self): + with pytest.raises(TypeError): + NoOpNlpEngine() + + def test_when_init_with_unexpected_argument_then_raises(self): + with pytest.raises(TypeError): + NoOpNlpEngine(unexpected_argument=True) + + def test_when_init_with_empty_models_then_raises(self): + with pytest.raises(ValueError, match="non-empty"): + NoOpNlpEngine(models=[]) + + def test_when_init_with_non_dict_model_then_raises(self): + with pytest.raises(ValueError, match="Each model must be a dictionary"): + NoOpNlpEngine(models=["no_op"]) + + def test_when_init_without_lang_code_then_raises(self): + with pytest.raises(ValueError, match="lang_code is missing"): + NoOpNlpEngine(models=[{"model_name": "no_op"}]) + + def test_when_init_without_model_name_then_raises(self): + with pytest.raises(ValueError, match="model_name is missing"): + NoOpNlpEngine(models=[{"lang_code": "en"}]) + + def test_when_init_with_empty_language_then_raises(self): + with pytest.raises(ValueError, match="lang_code must be a non-empty string"): + NoOpNlpEngine(models=[{"lang_code": "", "model_name": "no_op"}]) + + def test_when_init_with_non_string_model_name_then_raises(self): + with pytest.raises(ValueError, match="model_name must be a string"): + NoOpNlpEngine(models=[{"lang_code": "en", "model_name": 1}]) + + def test_when_init_with_any_model_name_then_succeeds(self): + engine = NoOpNlpEngine( + models=[{"lang_code": "en", "model_name": "en_core_web_lg"}] + ) + + assert engine.models == [ + {"lang_code": "en", "model_name": "en_core_web_lg"} + ] + + def test_when_init_with_extra_model_key_then_ignores_it(self): + engine = NoOpNlpEngine( + models=[ + { + "lang_code": "en", + "model_name": "no_op", + "extra_key": "ignored", + } + ] + ) + + assert engine.models == [{"lang_code": "en", "model_name": "no_op"}] + + def test_when_init_with_language_spaces_then_raises(self): + with pytest.raises(ValueError, match="leading or trailing"): + NoOpNlpEngine(models=[{"lang_code": " en", "model_name": "no_op"}]) + + def test_when_init_with_duplicate_languages_then_raises(self): + with pytest.raises(ValueError, match="Duplicate language"): + NoOpNlpEngine( + models=[ + {"lang_code": "en", "model_name": "no_op"}, + {"lang_code": "en", "model_name": "no_op"}, + ] + ) + + def test_when_load_then_supported_language_is_available(self, no_op_nlp_engine): + assert no_op_nlp_engine.is_loaded() + assert no_op_nlp_engine.get_supported_languages() == ["en"] + + def test_when_load_with_any_model_name_then_supported_language_is_available(self): + engine = NoOpNlpEngine(models=[{"lang_code": "en", "model_name": "no_op"}]) + engine.models[0]["model_name"] = "en_core_web_lg" + + engine.load() + + assert engine.is_loaded() + assert engine.get_supported_languages() == ["en"] + + def test_when_process_text_then_empty_artifacts_returned(self, no_op_nlp_engine): + artifacts = no_op_nlp_engine.process_text("John Smith", language="en") + + assert artifacts.entities == [] + assert isinstance(artifacts.tokens, Doc) + assert len(artifacts.tokens) == 0 + assert artifacts.tokens_indices == [] + assert artifacts.lemmas == [] + assert artifacts.keywords == [] + assert artifacts.scores == [] + assert artifacts.nlp_engine is no_op_nlp_engine + + def test_when_process_text_with_unsupported_language_then_raises( + self, no_op_nlp_engine + ): + with pytest.raises(ValueError, match="not supported"): + no_op_nlp_engine.process_text("John Smith", language="es") + + def test_when_process_text_with_non_string_text_then_raises(self, no_op_nlp_engine): + with pytest.raises(TypeError, match="text must be a string"): + no_op_nlp_engine.process_text(123, language="en") + + def test_when_process_text_with_non_string_language_then_raises( + self, no_op_nlp_engine + ): + with pytest.raises(TypeError, match="language must be a string"): + no_op_nlp_engine.process_text("John Smith", language=123) + + def test_when_not_loaded_then_process_text_raises(self): + engine = NoOpNlpEngine(models=[{"lang_code": "en", "model_name": "no_op"}]) + + with pytest.raises(ValueError, match="not loaded"): + engine.process_text("John Smith", language="en") + + def test_when_not_loaded_then_get_supported_languages_raises(self): + engine = NoOpNlpEngine(models=[{"lang_code": "en", "model_name": "no_op"}]) + + with pytest.raises(ValueError, match="not loaded"): + engine.get_supported_languages() + + def test_when_process_batch_then_empty_artifacts_returned(self, no_op_nlp_engine): + results = no_op_nlp_engine.process_batch(["one", "two"], language="en") + + assert isinstance(results, Iterator) + result_list = list(results) + assert [text for text, _ in result_list] == ["one", "two"] + assert all(artifacts.entities == [] for _, artifacts in result_list) + assert all(isinstance(artifacts.tokens, Doc) for _, artifacts in result_list) + assert all(len(artifacts.tokens) == 0 for _, artifacts in result_list) + + def test_when_process_batch_as_tuples_then_context_preserved( + self, no_op_nlp_engine + ): + results = list( + no_op_nlp_engine.process_batch( + [("one", {"id": 1})], language="en", as_tuples=True + ) + ) + + assert len(results) == 1 + text, artifacts, context = results[0] + assert text == "one" + assert artifacts.entities == [] + assert context == {"id": 1} + + def test_when_process_batch_invalid_tuple_then_raises(self, no_op_nlp_engine): + with pytest.raises(ValueError, match="tuples"): + list( + no_op_nlp_engine.process_batch( + ["not a tuple"], language="en", as_tuples=True + ) + ) + + def test_when_process_batch_with_non_string_text_then_converts_to_string( + self, no_op_nlp_engine + ): + result = list(no_op_nlp_engine.process_batch([123], language="en")) + + assert result[0][0] == "123" + assert result[0][1].entities == [] + + def test_when_process_batch_as_tuples_with_non_string_text_then_converts_to_string( + self, no_op_nlp_engine + ): + result = list( + no_op_nlp_engine.process_batch( + [(123, {"id": 1})], language="en", as_tuples=True + ) + ) + + text, artifacts, context = result[0] + assert text == "123" + assert artifacts.entities == [] + assert context == {"id": 1} + + def test_when_process_batch_with_unsupported_kwarg_then_warns( + self, no_op_nlp_engine, caplog + ): + with caplog.at_level("WARNING", logger="presidio-analyzer"): + results = list( + no_op_nlp_engine.process_batch( + ["one"], language="en", unsupported_option=True + ) + ) + + assert results[0][0] == "one" + assert "Ignoring unsupported kwargs" in caplog.text + + def test_when_get_supported_entities_then_empty(self, no_op_nlp_engine): + assert no_op_nlp_engine.get_supported_entities() == [] + + def test_when_stopword_or_punct_then_false(self, no_op_nlp_engine): + assert no_op_nlp_engine.is_stopword("the", language="en") is False + assert no_op_nlp_engine.is_punct(".", language="en") is False + + def test_when_stopword_or_punct_with_non_string_word_then_raises( + self, no_op_nlp_engine + ): + with pytest.raises(TypeError, match="word must be a string"): + no_op_nlp_engine.is_stopword(123, language="en") + with pytest.raises(TypeError, match="word must be a string"): + no_op_nlp_engine.is_punct(123, language="en") + + +class TestNoOpNlpEngineProvider: + def test_when_create_no_op_engine_via_provider_then_succeeds(self): + nlp_configuration = { + "nlp_engine_name": "no_op", + "models": [{"lang_code": "en", "model_name": "no_op"}], + } + + engine = NlpEngineProvider(nlp_configuration=nlp_configuration).create_engine() + + assert isinstance(engine, NoOpNlpEngine) + assert engine.get_supported_languages() == ["en"] + assert engine.get_supported_entities() == [] + + def test_when_no_op_engine_in_available_engines(self): + provider = NlpEngineProvider() + + assert "no_op" in provider.nlp_engines + + def test_when_no_op_configuration_has_extra_key_then_ignores_it(self): + nlp_configuration = { + "nlp_engine_name": "no_op", + "models": [{"lang_code": "en", "model_name": "no_op"}], + "ner_model_configuration": {}, + } + + engine = NlpEngineProvider(nlp_configuration=nlp_configuration).create_engine() + + assert isinstance(engine, NoOpNlpEngine) + assert engine.get_supported_languages() == ["en"] + + +class TestNoOpNlpEngineModelInstallation: + def test_when_download_model_for_no_op_then_no_model_is_installed(self): + _download_model("no_op", "no_op") + + +class TestNoOpNlpEngineAnalyzerIntegration: + def test_when_analyzer_uses_no_op_then_spacy_recognizer_is_not_added( + self, no_op_nlp_engine + ): + analyzer = AnalyzerEngine(nlp_engine=no_op_nlp_engine) + + nlp_recognizers = [ + recognizer + for recognizer in analyzer.registry.recognizers + if isinstance(recognizer, SpacyRecognizer) + ] + assert nlp_recognizers == [] + + def test_when_analyzer_uses_no_op_with_registered_nlp_recognizer_then_raises( + self, no_op_nlp_engine + ): + registry = RecognizerRegistry( + recognizers=[SpacyRecognizer()], supported_languages=["en"] + ) + + with pytest.raises(ValueError, match="NoOpNlpEngine cannot be used"): + AnalyzerEngine(registry=registry, nlp_engine=no_op_nlp_engine) + + def test_when_get_nlp_recognizer_with_no_op_then_raises(self, no_op_nlp_engine): + with pytest.raises(ValueError, match="does not have an NLP recognizer"): + RecognizerRegistry.get_nlp_recognizer(no_op_nlp_engine) + + def test_when_analyzer_uses_no_op_then_pattern_recognizers_still_run( + self, no_op_nlp_engine + ): + analyzer = AnalyzerEngine(nlp_engine=no_op_nlp_engine) + + results = analyzer.analyze( + text="My credit card is 4111111111111111", + language="en", + entities=["CREDIT_CARD"], + ) + + assert len(results) == 1 + assert results[0].entity_type == "CREDIT_CARD" + + def test_when_batch_analyzer_uses_no_op_then_primitive_values_still_run( + self, no_op_nlp_engine + ): + analyzer = AnalyzerEngine(nlp_engine=no_op_nlp_engine) + batch_analyzer = BatchAnalyzerEngine(analyzer_engine=analyzer) + + results = batch_analyzer.analyze_iterator( + texts=[4111111111111111], + language="en", + entities=["CREDIT_CARD"], + ) + + assert len(results) == 1 + assert len(results[0]) == 1 + assert results[0][0].entity_type == "CREDIT_CARD" + + def test_when_analyzer_uses_no_op_then_explicit_context_enhances_score( + self, no_op_nlp_engine + ): + zip_recognizer = PatternRecognizer( + supported_entity="ZIP", + patterns=[Pattern(name="zip", regex=r"\b\d{5}\b", score=0.3)], + context=["zip"], + ) + analyzer = AnalyzerEngine(nlp_engine=no_op_nlp_engine) + + results = analyzer.analyze( + text="10023", + language="en", + entities=["ZIP"], + ad_hoc_recognizers=[zip_recognizer], + context=["zip"], + return_decision_process=True, + ) + + assert len(results) == 1 + assert results[0].score > 0.3 + assert results[0].analysis_explanation.supportive_context_word == "zip" + + def test_when_analyzer_uses_no_op_then_in_text_context_does_not_enhance_score( + self, no_op_nlp_engine + ): + zip_recognizer = PatternRecognizer( + supported_entity="ZIP", + patterns=[Pattern(name="zip", regex=r"\b\d{5}\b", score=0.3)], + context=["zip"], + ) + analyzer = AnalyzerEngine(nlp_engine=no_op_nlp_engine) + + results = analyzer.analyze( + text="my zip code is 10023", + language="en", + entities=["ZIP"], + ad_hoc_recognizers=[zip_recognizer], + ) + + assert len(results) == 1 + assert results[0].score == 0.3 + + def test_when_analyzer_uses_no_op_with_lemma_context_enhancer_then_warns( + self, no_op_nlp_engine + ): + with pytest.warns(UserWarning, match="cannot use context words"): + AnalyzerEngine( + nlp_engine=no_op_nlp_engine, + context_aware_enhancer=LemmaContextAwareEnhancer(), + ) + + def test_when_provider_uses_no_op_then_spacy_recognizer_is_not_added( + self, tmp_path + ): + analyzer_conf_file = tmp_path / "no_op_analyzer.yaml" + analyzer_conf_file.write_text( + """ +supported_languages: + - en +default_score_threshold: 0 +nlp_configuration: + nlp_engine_name: no_op + models: + - lang_code: en + model_name: no_op +recognizer_registry: + recognizers: + - name: CreditCardRecognizer + type: predefined +""", + encoding="utf-8", + ) + + analyzer = AnalyzerEngineProvider( + analyzer_engine_conf_file=analyzer_conf_file + ).create_engine() + + assert isinstance(analyzer.nlp_engine, NoOpNlpEngine) + assert not any( + isinstance(recognizer, SpacyRecognizer) + for recognizer in analyzer.registry.recognizers + ) + + def test_when_provider_uses_no_op_with_nlp_recognizer_then_raises(self, tmp_path): + analyzer_conf_file = tmp_path / "no_op_with_nlp_recognizer.yaml" + analyzer_conf_file.write_text( + """ +supported_languages: + - en +default_score_threshold: 0 +nlp_configuration: + nlp_engine_name: no_op + models: + - lang_code: en + model_name: no_op +recognizer_registry: + recognizers: + - name: SpacyRecognizer + type: predefined +""", + encoding="utf-8", + ) + + provider = AnalyzerEngineProvider(analyzer_engine_conf_file=analyzer_conf_file) + + with pytest.raises(ValueError, match="NoOpNlpEngine cannot be used"): + provider.create_engine()