Coverage for presidio_analyzer / recognizer_registry / recognizer_registry_provider.py: 98%

1from __future__ import annotations 

2 

3import logging 

4import warnings 

5from pathlib import Path 

6from typing import Any, Dict, List, Optional, Union 

7 

8from presidio_analyzer import EntityRecognizer 

9from presidio_analyzer.input_validation import ConfigurationValidator 

10from presidio_analyzer.nlp_engine import NlpEngine 

11from presidio_analyzer.predefined_recognizers import SpacyRecognizer 

12from presidio_analyzer.recognizer_registry import RecognizerRegistry 

13from presidio_analyzer.recognizer_registry.recognizers_loader_utils import ( 

14 RecognizerConfigurationLoader, 

15 RecognizerListLoader, 

16) 

17 

18logger = logging.getLogger("presidio-analyzer") 

19 

20 

21class RecognizerRegistryProvider: 

22 r""" 

23 Utility class for loading Recognizer Registry. 

24 

25 Use this class to load recognizer registry from a yaml file 

26 

27 :param conf_file: Path to yaml file containing registry configuration 

28 :param registry_configuration: Dict containing registry configuration 

29 :example: 

30 { 

31 "supported_languages": ["de", "es"], 

32 "recognizers": [ 

33 { 

34 "name": "Zip code Recognizer", 

35 "supported_language": "en", 

36 "patterns": [ 

37 { 

38 "name": "zip code (weak)", 

39 "regex": "(\\b\\d{5}(?:\\-\\d{4})?\\b)", 

40 "score": 0.01, 

41 } 

42 ], 

43 "context": ["zip", "code"], 

44 "supported_entity": "ZIP", 

45 } 

46 ] 

47 } 

48 """ 

49 

50 def __init__( 

51 self, 

52 conf_file: Optional[Union[Path, str]] = None, 

53 registry_configuration: Optional[Dict] = None, 

54 nlp_engine: Optional[NlpEngine] = None, 

55 ): 

56 self.configuration = RecognizerConfigurationLoader.get( 

57 conf_file=conf_file, registry_configuration=registry_configuration 

58 ) 

59 

60 self.configuration = ( 

61 ConfigurationValidator.validate_recognizer_registry_configuration( 

62 self.configuration 

63 ) 

64 ) 

65 self.nlp_engine = nlp_engine 

66 

67 def create_recognizer_registry(self) -> RecognizerRegistry: 

68 """Create a recognizer registry according to configuration loaded previously.""" 

69 supported_languages = self.configuration.get("supported_languages") 

70 global_regex_flags = self.configuration.get("global_regex_flags") 

71 recognizers_conf = self.configuration.get("recognizers") 

72 recognizers = RecognizerListLoader.get( 

73 recognizers_conf, 

74 supported_languages, 

75 global_regex_flags, 

76 ) 

77 

78 recognizers = list(recognizers) 

79 

80 self.__update_based_on_nlp_recognizer_conf( 

81 recognizers, recognizers_conf, supported_languages 

82 ) 

83 

84 registry = RecognizerRegistry( 

85 recognizers=recognizers, 

86 supported_languages=supported_languages, 

87 global_regex_flags=global_regex_flags, 

88 ) 

89 

90 return registry 

91 

92 def __update_based_on_nlp_recognizer_conf( 

93 self, 

94 recognizers: List[EntityRecognizer], 

95 recognizers_conf: Optional[Dict], 

96 supported_languages: List[str], 

97 ) -> None: 

98 """Update the list of recognizers based on the NLP recognizer configuration. 

99 

100 The method adds the NLP recognizer to the list of recognizers 

101 if it is not already present, 

102 or removes it if it is not enabled in the configuration. 

103 Furthermore, it checks if there are 

104 any inconsistencies in configuration. For example: 

105 - Multiple enabled NLP recognizers in the configuration for one language. 

106 - The NLP recognizer in the configuration does not match the Nlp Engine. 

107 

108 :param recognizers: List of recognizers to update. 

109 :param recognizers_conf: Configuration of the recognizers from the YAML file 

110 :param supported_languages: List of supported languages. 

111 

112 :raises ValueError: If there are multiple enabled NLP recognizers 

113 in the configuration. 

114 :raises ValueError: If the NLP recognizer 

115 in the configuration does not match the Nlp Engine. 

116 """ 

117 nlp_engine = self.nlp_engine 

118 

119 if not nlp_engine: 

120 return 

121 

122 for language in nlp_engine.get_supported_languages(): 

123 self.__update_based_on_nlp_recognizer_conf_and_lang( 

124 recognizers=recognizers, nlp_engine=nlp_engine, language=language 

125 ) 

126 self.__remove_disabled_nlp_recognizers( 

127 recognizers=recognizers, 

128 recognizers_conf=recognizers_conf, 

129 language=language, 

130 ) 

131 

132 @staticmethod 

133 def __update_based_on_nlp_recognizer_conf_and_lang( 

134 recognizers: List[EntityRecognizer], 

135 nlp_engine: NlpEngine, 

136 language: str, 

137 ): 

138 """ 

139 Update the list of recognizers with nlp recognizers. 

140 

141 Update based on the NLP recognizer configuration for a specific language. 

142 """ 

143 

144 nlp_recognizers = [ 

145 rec 

146 for rec in recognizers 

147 if isinstance(rec, SpacyRecognizer) and rec.supported_language == language 

148 ] 

149 

150 # Case 1: NLP recognizer is not in the list of recognizers 

151 if not nlp_recognizers: 

152 warning_text = ( 

153 f"NLP recognizer (e.g. SpacyRecognizer, StanzaRecognizer) " 

154 f"is not in the list of recognizers " 

155 f"for language {language}. " 

156 f"Adding the default recognizer to the list." 

157 f"If you wish to remove the NLP recognizer, " 

158 f"define it as `enabled=false`." 

159 ) 

160 logger.warning(warning_text) 

161 warnings.warn(warning_text) 

162 if nlp_engine: 

163 nlp_recognizer_cls = RecognizerRegistry.get_nlp_recognizer( 

164 nlp_engine=nlp_engine 

165 ) 

166 recognizers.append( 

167 nlp_recognizer_cls( 

168 supported_language=language, 

169 supported_entities=nlp_engine.get_supported_entities(), 

170 ) 

171 ) 

172 else: 

173 recognizers.append(SpacyRecognizer(supported_language=language)) 

174 return 

175 

176 # Case 2: There are multiple NLP recognizers for this language, throw error 

177 if len(nlp_recognizers) > 1: 

178 raise ValueError( 

179 f"Multiple NLP recognizers for language {language} " 

180 f"found in the configuration. " 

181 f"Please remove the duplicates." 

182 ) 

183 

184 # Case 3: There is a mismatch between the NLP Engine and the NLP Recognizer 

185 nlp_recognizer = nlp_recognizers[0] 

186 expected_nlp_recognizer_cls = RecognizerRegistry.get_nlp_recognizer(nlp_engine) 

187 if nlp_recognizer.__class__ != expected_nlp_recognizer_cls: 

188 raise ValueError( 

189 f"There is a mismatch between the NLP Engine defined " 

190 f"({nlp_engine.__class__.__name__})," 

191 f"and the configured NLP recognizer " 

192 f"({nlp_recognizer.__class__.__name__})." 

193 f"Make sure the NLP recognizer is aligned with the " 

194 f"NLP engine and that all others are removed/disabled." 

195 ) 

196 

197 @staticmethod 

198 def __remove_disabled_nlp_recognizers( 

199 recognizers: List[EntityRecognizer], 

200 recognizers_conf: Dict[str, Any], 

201 language: str, 

202 ): 

203 """ 

204 Remove recognizers that are disabled in the configuration. 

205 

206 Goes through the recognizer conf provided by the user, 

207 and checks if a recognizer for a given language is disabled. 

208 If yes, it removes it from the recognizers list 

209 (as some are not removed in the previous step). 

210 """ 

211 

212 disabled = [ 

213 rec_conf 

214 for rec_conf in recognizers_conf 

215 if not RecognizerListLoader.is_recognizer_enabled(rec_conf) 

216 ] 

217 

218 disabled_rec_names = [ 

219 RecognizerListLoader.get_recognizer_name(rec) for rec in disabled 

220 ] 

221 

222 if not disabled: 

223 return 

224 

225 disabled_rec_classes = [ 

226 cls 

227 for cls in RecognizerListLoader.get_all_existing_recognizers() 

228 if cls.__name__ in disabled_rec_names 

229 ] 

230 

231 lang_recognizers = [ 

232 rec for rec in recognizers if rec.supported_language == language 

233 ] 

234 

235 for recognizer in lang_recognizers: 

236 if type(recognizer) in disabled_rec_classes: 

237 recognizers.remove(recognizer) 

238 logger.info( 

239 f"Disabled {recognizer.__class__.__name__} " 

240 f"recognizer for language {language}." 

241 )