Coverage for presidio_analyzer / entity_recognizer.py: 94%

65 statements  

« prev     ^ index     » next       coverage.py v7.13.1, created at 2026-03-29 09:03 +0000

1import logging 

2from abc import abstractmethod 

3from typing import TYPE_CHECKING, Dict, List, Optional, Tuple 

4 

5from presidio_analyzer import RecognizerResult 

6 

7if TYPE_CHECKING: 

8 from presidio_analyzer.nlp_engine import NlpArtifacts 

9 

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

11 

12 

13class EntityRecognizer: 

14 """ 

15 A class representing an abstract PII entity recognizer. 

16 

17 EntityRecognizer is an abstract class to be inherited by 

18 Recognizers which hold the logic for recognizing specific PII entities. 

19 

20 EntityRecognizer exposes a method called enhance_using_context which 

21 can be overridden in case a custom context aware enhancement is needed 

22 in derived class of a recognizer. 

23 

24 :param supported_entities: the entities supported by this recognizer 

25 (for example, phone number, address, etc.) 

26 :param supported_language: the language supported by this recognizer. 

27 The supported language code is iso6391Name 

28 :param name: the name of this recognizer (optional) 

29 :param version: the recognizer current version 

30 :param context: a list of words which can help boost confidence score 

31 when they appear in context of the matched entity 

32 """ 

33 

34 MIN_SCORE = 0 

35 MAX_SCORE = 1.0 

36 

37 def __init__( 

38 self, 

39 supported_entities: List[str], 

40 name: str = None, 

41 supported_language: str = "en", 

42 version: str = "0.0.1", 

43 context: Optional[List[str]] = None, 

44 ): 

45 self.supported_entities = supported_entities 

46 

47 if name is None: 

48 self.name = self.__class__.__name__ # assign class name as name 

49 else: 

50 self.name = name 

51 

52 self._id = f"{self.name}_{id(self)}" 

53 

54 self.supported_language = supported_language 

55 self.version = version 

56 self.is_loaded = False 

57 self.context = context if context else [] 

58 

59 self.load() 

60 logger.info("Loaded recognizer: %s", self.name) 

61 self.is_loaded = True 

62 

63 @property 

64 def id(self): 

65 """Return a unique identifier of this recognizer.""" 

66 

67 return self._id 

68 

69 @abstractmethod 

70 def load(self) -> None: 

71 """ 

72 Initialize the recognizer assets if needed. 

73 

74 (e.g. machine learning models) 

75 """ 

76 

77 @abstractmethod 

78 def analyze( 

79 self, text: str, entities: List[str], nlp_artifacts: "NlpArtifacts" 

80 ) -> List[RecognizerResult]: 

81 """ 

82 Analyze text to identify entities. 

83 

84 :param text: The text to be analyzed 

85 :param entities: The list of entities this recognizer is able to detect 

86 :param nlp_artifacts: A group of attributes which are the result of 

87 an NLP process over the input text. 

88 :return: List of results detected by this recognizer. 

89 """ 

90 return None 

91 

92 def enhance_using_context( 

93 self, 

94 text: str, 

95 raw_recognizer_results: List[RecognizerResult], 

96 other_raw_recognizer_results: List[RecognizerResult], 

97 nlp_artifacts: "NlpArtifacts", 

98 context: Optional[List[str]] = None, 

99 ) -> List[RecognizerResult]: 

100 """Enhance confidence score using context of the entity. 

101 

102 Override this method in derived class in case a custom logic 

103 is needed, otherwise return value will be equal to 

104 raw_results. 

105 

106 in case a result score is boosted, derived class need to update 

107 result.recognition_metadata[RecognizerResult.IS_SCORE_ENHANCED_BY_CONTEXT_KEY] 

108 

109 :param text: The actual text that was analyzed 

110 :param raw_recognizer_results: This recognizer's results, to be updated 

111 based on recognizer specific context. 

112 :param other_raw_recognizer_results: Other recognizer results matched in 

113 the given text to allow related entity context enhancement 

114 :param nlp_artifacts: The nlp artifacts contains elements 

115 such as lemmatized tokens for better 

116 accuracy of the context enhancement process 

117 :param context: list of context words 

118 """ 

119 return raw_recognizer_results 

120 

121 def get_supported_entities(self) -> List[str]: 

122 """ 

123 Return the list of entities this recognizer can identify. 

124 

125 :return: A list of the supported entities by this recognizer 

126 """ 

127 return self.supported_entities 

128 

129 def get_supported_language(self) -> str: 

130 """ 

131 Return the language this recognizer can support. 

132 

133 :return: A list of the supported language by this recognizer 

134 """ 

135 return self.supported_language 

136 

137 def get_version(self) -> str: 

138 """ 

139 Return the version of this recognizer. 

140 

141 :return: The current version of this recognizer 

142 """ 

143 return self.version 

144 

145 def to_dict(self) -> Dict: 

146 """ 

147 Serialize self to dictionary. 

148 

149 :return: a dictionary 

150 """ 

151 return_dict = { 

152 "supported_entities": self.supported_entities, 

153 "supported_language": self.supported_language, 

154 "name": self.name, 

155 "version": self.version, 

156 } 

157 return return_dict 

158 

159 @classmethod 

160 def from_dict(cls, entity_recognizer_dict: Dict) -> "EntityRecognizer": 

161 """ 

162 Create EntityRecognizer from a dict input. 

163 

164 :param entity_recognizer_dict: Dict containing keys and values for instantiation 

165 """ 

166 return cls(**entity_recognizer_dict) 

167 

168 @staticmethod 

169 def remove_duplicates(results: List[RecognizerResult]) -> List[RecognizerResult]: 

170 """ 

171 Remove duplicate results. 

172 

173 Remove duplicates in case the two results 

174 have identical start and ends and types. 

175 :param results: List[RecognizerResult] 

176 :return: List[RecognizerResult] 

177 """ 

178 results = list(set(results)) 

179 results = sorted(results, key=lambda x: (-x.score, x.start, -(x.end - x.start))) 

180 filtered_results = [] 

181 

182 for result in results: 

183 if result.score == 0: 

184 continue 

185 

186 to_keep = result not in filtered_results # equals based comparison 

187 if to_keep: 

188 for filtered in filtered_results: 

189 # If result is contained in one of the other results 

190 if ( 

191 result.contained_in(filtered) 

192 and result.entity_type == filtered.entity_type 

193 ): 

194 to_keep = False 

195 break 

196 

197 if to_keep: 

198 filtered_results.append(result) 

199 

200 return filtered_results 

201 

202 @staticmethod 

203 def sanitize_value(text: str, replacement_pairs: List[Tuple[str, str]]) -> str: 

204 """ 

205 Cleanse the input string of the replacement pairs specified as argument. 

206 

207 :param text: input string 

208 :param replacement_pairs: pairs of what has to be replaced with which value 

209 :return: cleansed string 

210 """ 

211 for search_string, replacement_string in replacement_pairs: 

212 text = text.replace(search_string, replacement_string) 

213 return text