Coverage for presidio_analyzer / recognizer_registry / recognizer_registry.py: 87%
114 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-03-29 09:03 +0000
1import copy
2import logging
3from pathlib import Path
4from typing import Dict, Iterable, List, Optional, Type, Union
6import regex as re
7import yaml
9from presidio_analyzer import EntityRecognizer, PatternRecognizer
10from presidio_analyzer.nlp_engine import (
11 NlpEngine,
12 SpacyNlpEngine,
13 StanzaNlpEngine,
14 TransformersNlpEngine,
15)
16from presidio_analyzer.predefined_recognizers import (
17 SpacyRecognizer,
18 StanzaRecognizer,
19 TransformersRecognizer,
20)
21from presidio_analyzer.recognizer_registry.recognizers_loader_utils import (
22 RecognizerConfigurationLoader,
23 RecognizerListLoader,
24)
26logger = logging.getLogger("presidio-analyzer")
29class RecognizerRegistry:
30 """
31 Detect, register and hold all recognizers to be used by the analyzer.
33 :param recognizers: An optional list of recognizers,
34 that will be available instead of the predefined recognizers
35 :param global_regex_flags: regex flags to be used in regex matching,
36 including deny-lists
37 :param supported_languages: List of languages supported by this registry.
39 """
41 def __init__(
42 self,
43 recognizers: Optional[Iterable[EntityRecognizer]] = None,
44 global_regex_flags: Optional[int] = re.DOTALL | re.MULTILINE | re.IGNORECASE,
45 supported_languages: Optional[List[str]] = None,
46 ):
47 if recognizers:
48 self.recognizers = recognizers
49 else:
50 self.recognizers = []
51 self.global_regex_flags = global_regex_flags
52 self.supported_languages = (
53 supported_languages if supported_languages else ["en"]
54 )
56 def _create_nlp_recognizer(
57 self,
58 nlp_engine: Optional[NlpEngine] = None,
59 supported_language: Optional[str] = None,
60 ) -> SpacyRecognizer:
61 nlp_recognizer = self.get_nlp_recognizer(nlp_engine)
63 if nlp_engine:
64 return nlp_recognizer(
65 supported_language=supported_language,
66 supported_entities=nlp_engine.get_supported_entities(),
67 )
69 return nlp_recognizer(supported_language=supported_language)
71 def add_nlp_recognizer(self, nlp_engine: NlpEngine) -> None:
72 """
73 Adding NLP recognizer in accordance with the nlp engine.
75 :param nlp_engine: The NLP engine.
76 :return: None
77 """
79 if not nlp_engine:
80 supported_languages = self.supported_languages
81 else:
82 supported_languages = nlp_engine.get_supported_languages()
84 self.recognizers.extend(
85 [
86 self._create_nlp_recognizer(
87 nlp_engine=nlp_engine, supported_language=supported_language
88 )
89 for supported_language in supported_languages
90 ]
91 )
93 def load_predefined_recognizers(
94 self, languages: Optional[List[str]] = None, nlp_engine: NlpEngine = None
95 ) -> None:
96 """
97 Load the existing recognizers into memory.
99 :param languages: List of languages for which to load recognizers
100 :param nlp_engine: The NLP engine to use.
101 :return: None
102 """
104 registry_configuration = {"global_regex_flags": self.global_regex_flags}
105 if languages is not None:
106 registry_configuration["supported_languages"] = languages
108 configuration = RecognizerConfigurationLoader.get(
109 registry_configuration=registry_configuration
110 )
111 recognizers = RecognizerListLoader.get(**configuration)
113 self.recognizers.extend(recognizers)
114 self.add_nlp_recognizer(nlp_engine=nlp_engine)
116 @staticmethod
117 def get_nlp_recognizer(
118 nlp_engine: NlpEngine,
119 ) -> Type[SpacyRecognizer]:
120 """Return the recognizer leveraging the selected NLP Engine."""
122 if isinstance(nlp_engine, StanzaNlpEngine):
123 return StanzaRecognizer
124 if isinstance(nlp_engine, TransformersNlpEngine):
125 return TransformersRecognizer
126 if not nlp_engine or isinstance(nlp_engine, SpacyNlpEngine):
127 return SpacyRecognizer
128 else:
129 logger.warning(
130 "nlp engine should be either SpacyNlpEngine,"
131 "StanzaNlpEngine or TransformersNlpEngine"
132 )
133 # Returning default
134 return SpacyRecognizer
136 def get_recognizers(
137 self,
138 language: str,
139 entities: Optional[List[str]] = None,
140 all_fields: bool = False,
141 ad_hoc_recognizers: Optional[List[EntityRecognizer]] = None,
142 ) -> List[EntityRecognizer]:
143 """
144 Return a list of recognizers which supports the specified name and language.
146 :param entities: the requested entities
147 :param language: the requested language
148 :param all_fields: a flag to return all fields of a requested language.
149 :param ad_hoc_recognizers: Additional recognizers provided by the user
150 as part of the request
151 :return: A list of the recognizers which supports the supplied entities
152 and language
153 """
154 if language is None:
155 raise ValueError("No language provided")
157 if entities is None and all_fields is False:
158 raise ValueError("No entities provided")
160 all_possible_recognizers = copy.copy(self.recognizers)
161 if ad_hoc_recognizers:
162 all_possible_recognizers.extend(ad_hoc_recognizers)
164 # filter out unwanted recognizers
165 to_return = set()
166 if all_fields:
167 to_return = [
168 rec
169 for rec in all_possible_recognizers
170 if language == rec.supported_language
171 ]
172 else:
173 for entity in entities:
174 subset = [
175 rec
176 for rec in all_possible_recognizers
177 if entity in rec.supported_entities
178 and language == rec.supported_language
179 ]
181 if not subset:
182 logger.warning(
183 "Entity %s doesn't have the corresponding"
184 " recognizer in language : %s",
185 entity,
186 language,
187 )
188 else:
189 to_return.update(set(subset))
191 logger.debug(
192 "Returning a total of %s recognizers",
193 str(len(to_return)),
194 )
196 if not to_return:
197 raise ValueError("No matching recognizers were found to serve the request.")
199 return list(to_return)
201 def add_recognizer(self, recognizer: EntityRecognizer) -> None:
202 """
203 Add a new recognizer to the list of recognizers.
205 :param recognizer: Recognizer to add
206 """
207 if not isinstance(recognizer, EntityRecognizer):
208 raise ValueError("Input is not of type EntityRecognizer")
210 self.recognizers.append(recognizer)
212 def remove_recognizer(
213 self, recognizer_name: str, language: Optional[str] = None
214 ) -> None:
215 """
216 Remove a recognizer based on its name.
218 :param recognizer_name: Name of recognizer to remove
219 :param language: The supported language of the recognizer to be removed,
220 in case multiple recognizers with the same name are present,
221 and only one should be removed.
222 """
224 if not language:
225 new_recognizers = [
226 rec for rec in self.recognizers if rec.name != recognizer_name
227 ]
229 logger.info(
230 "Removed %s recognizers which had the name %s",
231 str(len(self.recognizers) - len(new_recognizers)),
232 recognizer_name,
233 )
235 else:
236 new_recognizers = [
237 rec
238 for rec in self.recognizers
239 if rec.name != recognizer_name or rec.supported_language != language
240 ]
242 logger.info(
243 "Removed %s recognizers which had the name %s and language %s",
244 str(len(self.recognizers) - len(new_recognizers)),
245 recognizer_name,
246 language,
247 )
249 self.recognizers = new_recognizers
251 def add_pattern_recognizer_from_dict(self, recognizer_dict: Dict) -> None:
252 """
253 Load a pattern recognizer from a Dict into the recognizer registry.
255 :param recognizer_dict: Dict holding a serialization of an PatternRecognizer
257 :example:
258 >>> registry = RecognizerRegistry()
259 >>> recognizer = { "name": "Titles Recognizer", "supported_language": "en","supported_entity": "TITLE", "deny_list": ["Mr.","Mrs."]}
260 >>> registry.add_pattern_recognizer_from_dict(recognizer)
261 """ # noqa: E501
263 recognizer = PatternRecognizer.from_dict(recognizer_dict)
264 self.add_recognizer(recognizer)
266 def add_recognizers_from_yaml(self, yml_path: Union[str, Path]) -> None:
267 r"""
268 Read YAML file and load recognizers into the recognizer registry.
270 See example yaml file here:
271 https://github.com/microsoft/presidio/blob/main/presidio-analyzer/presidio_analyzer/conf/example_recognizers.yaml
273 :example:
274 >>> yaml_file = "recognizers.yaml"
275 >>> registry = RecognizerRegistry()
276 >>> registry.add_recognizers_from_yaml(yaml_file)
278 """
280 try:
281 with open(yml_path) as stream:
282 yaml_recognizers = yaml.safe_load(stream)
284 for yaml_recognizer in yaml_recognizers["recognizers"]:
285 self.add_pattern_recognizer_from_dict(yaml_recognizer)
286 except OSError as io_error:
287 print(f"Error reading file {yml_path}")
288 raise io_error
289 except yaml.YAMLError as yaml_error:
290 print(f"Failed to parse file {yml_path}")
291 raise yaml_error
292 except TypeError as yaml_error:
293 print(f"Failed to parse file {yml_path}")
294 raise yaml_error
296 def __instantiate_recognizer(
297 self, recognizer_class: Type[EntityRecognizer], supported_language: str
298 ):
299 """
300 Instantiate a recognizer class given type and input.
302 :param recognizer_class: Class object of the recognizer
303 :param supported_language: Language this recognizer should support
304 """
306 inst = recognizer_class(supported_language=supported_language)
307 if isinstance(inst, PatternRecognizer):
308 inst.global_regex_flags = self.global_regex_flags
309 return inst
311 def _get_supported_languages(self) -> List[str]:
312 languages = []
313 for rec in self.recognizers:
314 languages.append(rec.supported_language)
316 return list(set(languages))
318 def get_supported_entities(
319 self, languages: Optional[List[str]] = None
320 ) -> List[str]:
321 """
322 Return the supported entities by the set of recognizers loaded.
324 :param languages: The languages to get the supported entities for.
325 If languages=None, returns all entities for all languages.
326 """
327 if not languages:
328 languages = self._get_supported_languages()
330 supported_entities = []
331 for language in languages:
332 recognizers = self.get_recognizers(language=language, all_fields=True)
334 for recognizer in recognizers:
335 supported_entities.extend(recognizer.get_supported_entities())
337 return list(set(supported_entities))