Coverage for presidio_analyzer / analyzer_engine.py: 97%
136 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-03-29 09:03 +0000
1import json
2import logging
3import os
4from collections import Counter
5from typing import List, Optional
7import regex as re
9from presidio_analyzer import (
10 EntityRecognizer,
11 RecognizerResult,
12)
13from presidio_analyzer.app_tracer import AppTracer
14from presidio_analyzer.context_aware_enhancers import (
15 ContextAwareEnhancer,
16 LemmaContextAwareEnhancer,
17)
18from presidio_analyzer.nlp_engine import NlpArtifacts, NlpEngine, NlpEngineProvider
19from presidio_analyzer.recognizer_registry import (
20 RecognizerRegistry,
21 RecognizerRegistryProvider,
22)
24logger = logging.getLogger("presidio-analyzer")
26REGEX_TIMEOUT_SECONDS = int(os.environ.get("REGEX_TIMEOUT_SECONDS", 60))
28class AnalyzerEngine:
29 """
30 Entry point for Presidio Analyzer.
32 Orchestrating the detection of PII entities and all related logic.
34 :param registry: instance of type RecognizerRegistry
35 :param nlp_engine: instance of type NlpEngine
36 (for example SpacyNlpEngine)
37 :param app_tracer: instance of type AppTracer, used to trace the logic
38 used during each request for interpretability reasons.
39 :param log_decision_process: bool,
40 defines whether the decision process within the analyzer should be logged or not.
41 :param default_score_threshold: Minimum confidence value
42 for detected entities to be returned
43 :param supported_languages: List of possible languages this engine could be run on.
44 Used for loading the right NLP models and recognizers for these languages.
45 :param context_aware_enhancer: instance of type ContextAwareEnhancer for enhancing
46 confidence score based on context words, (LemmaContextAwareEnhancer will be created
47 by default if None passed)
48 """
50 def __init__(
51 self,
52 registry: RecognizerRegistry = None,
53 nlp_engine: NlpEngine = None,
54 app_tracer: AppTracer = None,
55 log_decision_process: bool = False,
56 default_score_threshold: float = 0,
57 supported_languages: List[str] = None,
58 context_aware_enhancer: Optional[ContextAwareEnhancer] = None,
59 ):
60 if not supported_languages:
61 supported_languages = ["en"]
63 if not nlp_engine:
64 logger.info("nlp_engine not provided, creating default.")
65 provider = NlpEngineProvider()
66 nlp_engine = provider.create_engine()
68 if not app_tracer:
69 app_tracer = AppTracer()
70 self.app_tracer = app_tracer
72 self.supported_languages = supported_languages
74 self.nlp_engine = nlp_engine
75 if not self.nlp_engine.is_loaded():
76 self.nlp_engine.load()
78 if not registry:
79 logger.info("registry not provided, creating default.")
80 provider = RecognizerRegistryProvider(
81 registry_configuration={"supported_languages": self.supported_languages}
82 )
83 registry = provider.create_recognizer_registry()
84 registry.add_nlp_recognizer(nlp_engine=self.nlp_engine)
85 else:
86 if Counter(registry.supported_languages) != Counter(
87 self.supported_languages
88 ):
89 raise ValueError(
90 f"Misconfigured engine, supported languages have to be consistent"
91 f"registry.supported_languages: {registry.supported_languages}, "
92 f"analyzer_engine.supported_languages: {self.supported_languages}"
93 )
95 # added to support the previous interface
96 if not registry.recognizers:
97 registry.load_predefined_recognizers(
98 nlp_engine=self.nlp_engine, languages=self.supported_languages
99 )
101 self.registry = registry
103 self.log_decision_process = log_decision_process
104 self.default_score_threshold = default_score_threshold
106 if not context_aware_enhancer:
107 logger.debug(
108 "context aware enhancer not provided, creating default"
109 + " lemma based enhancer."
110 )
111 context_aware_enhancer = LemmaContextAwareEnhancer()
113 self.context_aware_enhancer = context_aware_enhancer
115 def get_recognizers(self, language: Optional[str] = None) -> List[EntityRecognizer]:
116 """
117 Return a list of PII recognizers currently loaded.
119 :param language: Return the recognizers supporting a given language.
120 :return: List of [Recognizer] as a RecognizersAllResponse
121 """
122 if not language:
123 languages = self.supported_languages
124 else:
125 languages = [language]
127 recognizers = []
128 for language in languages:
129 logger.info(f"Fetching all recognizers for language {language}")
130 recognizers.extend(
131 self.registry.get_recognizers(language=language, all_fields=True)
132 )
134 return list(set(recognizers))
136 def get_supported_entities(self, language: Optional[str] = None) -> List[str]:
137 """
138 Return a list of the entities that can be detected.
140 :param language: Return only entities supported in a specific language.
141 :return: List of entity names
142 """
143 recognizers = self.get_recognizers(language=language)
144 supported_entities = []
145 for recognizer in recognizers:
146 supported_entities.extend(recognizer.get_supported_entities())
148 return list(set(supported_entities))
150 def analyze(
151 self,
152 text: str,
153 language: str,
154 entities: Optional[List[str]] = None,
155 correlation_id: Optional[str] = None,
156 score_threshold: Optional[float] = None,
157 return_decision_process: Optional[bool] = False,
158 ad_hoc_recognizers: Optional[List[EntityRecognizer]] = None,
159 context: Optional[List[str]] = None,
160 allow_list: Optional[List[str]] = None,
161 allow_list_match: Optional[str] = "exact",
162 regex_flags: Optional[int] = re.DOTALL | re.MULTILINE | re.IGNORECASE,
163 nlp_artifacts: Optional[NlpArtifacts] = None,
164 ) -> List[RecognizerResult]:
165 """
166 Find PII entities in text using different PII recognizers for a given language.
168 :param text: the text to analyze
169 :param language: the language of the text
170 :param entities: List of PII entities that should be looked for in the text.
171 If entities=None then all entities are looked for.
172 :param correlation_id: cross call ID for this request
173 :param score_threshold: A minimum value for which
174 to return an identified entity
175 :param return_decision_process: Whether the analysis decision process steps
176 returned in the response.
177 :param ad_hoc_recognizers: List of recognizers which will be used only
178 for this specific request.
179 :param context: List of context words to enhance confidence score if matched
180 with the recognized entity's recognizer context
181 :param allow_list: List of words that the user defines as being allowed to keep
182 in the text
183 :param allow_list_match: How the allow_list should be interpreted; either as "exact" or as "regex".
184 - If `regex`, results which match with any regex condition in the allow_list would be allowed and not be returned as potential PII.
185 - if `exact`, results which exactly match any value in the allow_list would be allowed and not be returned as potential PII.
186 :param regex_flags: regex flags to be used for when allow_list_match is "regex"
187 :param nlp_artifacts: precomputed NlpArtifacts
188 :return: an array of the found entities in the text
190 :Example:
192 ```python
193 from presidio_analyzer import AnalyzerEngine
195 # Set up the engine, loads the NLP module (spaCy model by default)
196 # and other PII recognizers
197 analyzer = AnalyzerEngine()
199 # Call analyzer to get results
200 results = analyzer.analyze(text='My phone number is 212-555-5555', entities=['PHONE_NUMBER'], language='en')
201 print(results)
202 ```
204 """ # noqa: E501
206 all_fields = not entities
208 recognizers = self.registry.get_recognizers(
209 language=language,
210 entities=entities,
211 all_fields=all_fields,
212 ad_hoc_recognizers=ad_hoc_recognizers,
213 )
215 if all_fields:
216 # Since all_fields=True, list all entities by iterating
217 # over all recognizers
218 entities = self.get_supported_entities(language=language)
220 # run the nlp pipeline over the given text, store the results in
221 # a NlpArtifacts instance
222 if not nlp_artifacts:
223 nlp_artifacts = self.nlp_engine.process_text(text, language)
225 if self.log_decision_process:
226 self.app_tracer.trace(
227 correlation_id, "nlp artifacts:" + nlp_artifacts.to_json()
228 )
230 results = []
231 for recognizer in recognizers:
232 # Lazy loading of the relevant recognizers
233 if not recognizer.is_loaded:
234 recognizer.load()
235 recognizer.is_loaded = True
237 # analyze using the current recognizer and append the results
238 current_results = recognizer.analyze(
239 text=text, entities=entities, nlp_artifacts=nlp_artifacts
240 )
241 if current_results:
242 # add recognizer name to recognition metadata inside results
243 # if not exists
244 self.__add_recognizer_id_if_not_exists(current_results, recognizer)
245 results.extend(current_results)
247 results = self._enhance_using_context(
248 text, results, nlp_artifacts, recognizers, context
249 )
251 if self.log_decision_process:
252 self.app_tracer.trace(
253 correlation_id,
254 json.dumps([str(result.to_dict()) for result in results]),
255 )
257 # Remove duplicates or low score results
258 results = EntityRecognizer.remove_duplicates(results)
259 results = self.__remove_low_scores(results, score_threshold)
261 if allow_list:
262 results = self._remove_allow_list(
263 results, allow_list, text, regex_flags, allow_list_match
264 )
266 if not return_decision_process:
267 results = self.__remove_decision_process(results)
269 return results
271 def _enhance_using_context(
272 self,
273 text: str,
274 raw_results: List[RecognizerResult],
275 nlp_artifacts: NlpArtifacts,
276 recognizers: List[EntityRecognizer],
277 context: Optional[List[str]] = None,
278 ) -> List[RecognizerResult]:
279 """
280 Enhance confidence score using context words.
282 :param text: The actual text that was analyzed
283 :param raw_results: Recognizer results which didn't take
284 context into consideration
285 :param nlp_artifacts: The nlp artifacts contains elements
286 such as lemmatized tokens for better
287 accuracy of the context enhancement process
288 :param recognizers: the list of recognizers
289 :param context: list of context words
290 """
291 results = []
293 for recognizer in recognizers:
294 recognizer_results = [
295 r
296 for r in raw_results
297 if r.recognition_metadata[RecognizerResult.RECOGNIZER_IDENTIFIER_KEY]
298 == recognizer.id
299 ]
300 other_recognizer_results = [
301 r
302 for r in raw_results
303 if r.recognition_metadata[RecognizerResult.RECOGNIZER_IDENTIFIER_KEY]
304 != recognizer.id
305 ]
307 # enhance score using context in recognizer level if implemented
308 recognizer_results = recognizer.enhance_using_context(
309 text=text,
310 # each recognizer will get access to all recognizer results
311 # to allow related entities contex enhancement
312 raw_recognizer_results=recognizer_results,
313 other_raw_recognizer_results=other_recognizer_results,
314 nlp_artifacts=nlp_artifacts,
315 context=context,
316 )
318 results.extend(recognizer_results)
320 # Update results in case surrounding words or external context are relevant to
321 # the context words.
322 results = self.context_aware_enhancer.enhance_using_context(
323 text=text,
324 raw_results=results,
325 nlp_artifacts=nlp_artifacts,
326 recognizers=recognizers,
327 context=context,
328 )
330 return results
332 def __remove_low_scores(
333 self, results: List[RecognizerResult], score_threshold: float = None
334 ) -> List[RecognizerResult]:
335 """
336 Remove results for which the confidence is lower than the threshold.
338 :param results: List of RecognizerResult
339 :param score_threshold: float value for minimum possible confidence
340 :return: List[RecognizerResult]
341 """
342 if score_threshold is None:
343 score_threshold = self.default_score_threshold
345 new_results = [result for result in results if result.score >= score_threshold]
346 return new_results
348 @staticmethod
349 def _remove_allow_list(
350 results: List[RecognizerResult],
351 allow_list: List[str],
352 text: str,
353 regex_flags: Optional[int],
354 allow_list_match: str,
355 ) -> List[RecognizerResult]:
356 """
357 Remove results which are part of the allow list.
359 :param results: List of RecognizerResult
360 :param allow_list: list of allowed terms
361 :param text: the text to analyze
362 :param regex_flags: regex flags to be used for when allow_list_match is "regex"
363 :param allow_list_match: How the allow_list
364 should be interpreted; either as "exact" or as "regex"
365 :return: List[RecognizerResult]
366 """
367 new_results = []
368 if allow_list_match == "regex":
369 pattern = "|".join(allow_list)
370 re_compiled = re.compile(pattern, flags=regex_flags)
372 for result in results:
373 word = text[result.start : result.end]
375 # if the word is not specified to be allowed, keep in the PII entities
376 try:
377 if not re_compiled.search(word, timeout=REGEX_TIMEOUT_SECONDS):
378 new_results.append(result)
379 except TimeoutError:
380 logger.warning(
381 "Allow list regex timed out after %s seconds"
382 " (word length: %d), keeping result.",
383 REGEX_TIMEOUT_SECONDS,
384 len(word),
385 exc_info=True,
386 )
387 new_results.append(result)
388 elif allow_list_match == "exact":
389 for result in results:
390 word = text[result.start : result.end]
392 # if the word is not specified to be allowed, keep in the PII entities
393 if word not in allow_list:
394 new_results.append(result)
395 else:
396 raise ValueError(
397 "allow_list_match must either be set to 'exact' or 'regex'."
398 )
400 return new_results
402 @staticmethod
403 def __add_recognizer_id_if_not_exists(
404 results: List[RecognizerResult], recognizer: EntityRecognizer
405 ) -> None:
406 """Ensure recognition metadata with recognizer id existence.
408 Ensure recognizer result list contains recognizer id inside recognition
409 metadata dictionary, and if not create it. recognizer_id is needed
410 for context aware enhancement.
412 :param results: List of RecognizerResult
413 :param recognizer: Entity recognizer
414 """
415 for result in results:
416 if not result.recognition_metadata:
417 result.recognition_metadata = dict()
418 if (
419 RecognizerResult.RECOGNIZER_IDENTIFIER_KEY
420 not in result.recognition_metadata
421 ):
422 result.recognition_metadata[
423 RecognizerResult.RECOGNIZER_IDENTIFIER_KEY
424 ] = recognizer.id
425 if RecognizerResult.RECOGNIZER_NAME_KEY not in result.recognition_metadata:
426 result.recognition_metadata[RecognizerResult.RECOGNIZER_NAME_KEY] = (
427 recognizer.name
428 )
430 @staticmethod
431 def __remove_decision_process(
432 results: List[RecognizerResult],
433 ) -> List[RecognizerResult]:
434 """Remove decision process / analysis explanation from response."""
436 for result in results:
437 result.analysis_explanation = None
439 return results