Coverage for presidio_analyzer / analyzer_engine_provider.py: 100%

1import logging 

2from pathlib import Path 

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

4 

5import yaml 

6 

7from presidio_analyzer import AnalyzerEngine, RecognizerRegistry 

8from presidio_analyzer.input_validation import ConfigurationValidator 

9from presidio_analyzer.nlp_engine import NlpEngine, NlpEngineProvider 

10from presidio_analyzer.recognizer_registry import RecognizerRegistryProvider 

11 

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

13 

14 

15class AnalyzerEngineProvider: 

16 """ 

17 Utility function for loading Presidio Analyzer. 

18 

19 Use this class to load presidio analyzer engine from a yaml file 

20 

21 :param analyzer_engine_conf_file: the path to the analyzer configuration file 

22 :param nlp_engine_conf_file: the path to the nlp engine configuration file 

23 :param recognizer_registry_conf_file: the path to the recognizer 

24 registry configuration file 

25 """ 

26 

27 def __init__( 

28 self, 

29 analyzer_engine_conf_file: Optional[Union[Path, str]] = None, 

30 nlp_engine_conf_file: Optional[Union[Path, str]] = None, 

31 recognizer_registry_conf_file: Optional[Union[Path, str]] = None, 

32 ): 

33 if analyzer_engine_conf_file: 

34 ConfigurationValidator.validate_file_path(analyzer_engine_conf_file) 

35 if nlp_engine_conf_file: 

36 ConfigurationValidator.validate_file_path(nlp_engine_conf_file) 

37 if recognizer_registry_conf_file: 

38 ConfigurationValidator.validate_file_path(recognizer_registry_conf_file) 

39 

40 self.configuration = self.get_configuration(conf_file=analyzer_engine_conf_file) 

41 self.nlp_engine_conf_file = nlp_engine_conf_file 

42 self.recognizer_registry_conf_file = recognizer_registry_conf_file 

43 

44 def get_configuration( 

45 self, conf_file: Optional[Union[Path, str]] 

46 ) -> Union[Dict[str, Any]]: 

47 """Retrieve analyzer engine configuration from the provided file.""" 

48 

49 if not conf_file: 

50 default_conf_file = self._get_full_conf_path() 

51 with open(default_conf_file) as file: 

52 configuration = yaml.safe_load(file) 

53 logger.info( 

54 f"Analyzer Engine configuration file " 

55 f"not provided. Using {default_conf_file}." 

56 ) 

57 else: 

58 try: 

59 logger.info(f"Reading analyzer configuration from {conf_file}") 

60 with open(conf_file) as file: 

61 configuration = yaml.safe_load(file) 

62 except OSError: 

63 logger.warning( 

64 f"configuration file {conf_file} not found. " 

65 f"Using default config." 

66 ) 

67 with open(self._get_full_conf_path()) as file: 

68 configuration = yaml.safe_load(file) 

69 except Exception: 

70 logger.warning( 

71 f"Failed to parse file {conf_file}, resorting to default" 

72 ) 

73 with open(self._get_full_conf_path()) as file: 

74 configuration = yaml.safe_load(file) 

75 

76 ConfigurationValidator.validate_analyzer_configuration(configuration) 

77 logger.debug("Analyzer configuration validation passed") 

78 

79 return configuration 

80 

81 def create_engine(self) -> AnalyzerEngine: 

82 """ 

83 Load Presidio Analyzer from yaml configuration file. 

84 

85 :return: analyzer engine initialized with yaml configuration 

86 """ 

87 

88 nlp_engine = self._load_nlp_engine() 

89 supported_languages = self.configuration.get("supported_languages", ["en"]) 

90 default_score_threshold = self.configuration.get("default_score_threshold", 0) 

91 

92 registry = self._load_recognizer_registry( 

93 supported_languages=supported_languages, nlp_engine=nlp_engine 

94 ) 

95 

96 analyzer = AnalyzerEngine( 

97 nlp_engine=nlp_engine, 

98 registry=registry, 

99 supported_languages=supported_languages, 

100 default_score_threshold=default_score_threshold, 

101 ) 

102 

103 return analyzer 

104 

105 def _load_recognizer_registry( 

106 self, 

107 supported_languages: List[str], 

108 nlp_engine: NlpEngine, 

109 ) -> RecognizerRegistry: 

110 if self.recognizer_registry_conf_file: 

111 logger.info( 

112 f"Reading recognizer registry " 

113 f"configuration from {self.recognizer_registry_conf_file}" 

114 ) 

115 provider = RecognizerRegistryProvider( 

116 conf_file=self.recognizer_registry_conf_file, nlp_engine=nlp_engine 

117 ) 

118 elif "recognizer_registry" in self.configuration: 

119 registry_configuration = self.configuration["recognizer_registry"] 

120 provider = RecognizerRegistryProvider( 

121 registry_configuration={ 

122 **registry_configuration, 

123 "supported_languages": supported_languages, 

124 }, 

125 nlp_engine=nlp_engine, 

126 ) 

127 else: 

128 logger.warning( 

129 "configuration file is missing for 'recognizer_registry'. " 

130 "Using default configuration for recognizer registry" 

131 ) 

132 registry_configuration = self.configuration.get("recognizer_registry", {}) 

133 provider = RecognizerRegistryProvider( 

134 registry_configuration={ 

135 **registry_configuration, 

136 "supported_languages": supported_languages, 

137 }, 

138 nlp_engine=nlp_engine, 

139 ) 

140 registry = provider.create_recognizer_registry() 

141 

142 return registry 

143 

144 def _load_nlp_engine(self) -> NlpEngine: 

145 if self.nlp_engine_conf_file: 

146 logger.info(f"Reading nlp configuration from {self.nlp_engine_conf_file}") 

147 provider = NlpEngineProvider(conf_file=self.nlp_engine_conf_file) 

148 elif "nlp_configuration" in self.configuration: 

149 nlp_configuration = self.configuration["nlp_configuration"] 

150 provider = NlpEngineProvider(nlp_configuration=nlp_configuration) 

151 else: 

152 logger.warning( 

153 "configuration file is missing for 'nlp_configuration'." 

154 "Using default configuration for nlp engine" 

155 ) 

156 provider = NlpEngineProvider() 

157 

158 return provider.create_engine() 

159 

160 @staticmethod 

161 def _get_full_conf_path( 

162 default_conf_file: Union[Path, str] = "default_analyzer.yaml", 

163 ) -> Path: 

164 """Return a Path to the default conf file.""" 

165 return Path(Path(__file__).parent, "conf", default_conf_file)