Coverage for presidio_analyzer / pattern_recognizer.py: 100%
105 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-03-29 09:03 +0000
1import datetime
2import logging
3import os
4from typing import TYPE_CHECKING, Dict, List, Optional
6import regex as re
8from presidio_analyzer import (
9 AnalysisExplanation,
10 EntityRecognizer,
11 LocalRecognizer,
12 Pattern,
13 RecognizerResult,
14)
16if TYPE_CHECKING:
17 from presidio_analyzer.nlp_engine import NlpArtifacts
19logger = logging.getLogger("presidio-analyzer")
21REGEX_TIMEOUT_SECONDS = int(os.environ.get("REGEX_TIMEOUT_SECONDS", 60))
24class PatternRecognizer(LocalRecognizer):
25 """
26 PII entity recognizer using regular expressions or deny-lists.
28 :param patterns: A list of patterns to detect
29 :param deny_list: A list of words to detect,
30 in case our recognizer uses a predefined list of words (deny list)
31 :param context: list of context words
32 :param deny_list_score: confidence score for a term
33 identified using a deny-list
34 :param global_regex_flags: regex flags to be used in regex matching,
35 including deny-lists.
36 """
38 def __init__(
39 self,
40 supported_entity: str,
41 name: str = None,
42 supported_language: str = "en",
43 patterns: List[Pattern] = None,
44 deny_list: List[str] = None,
45 context: List[str] = None,
46 deny_list_score: float = 1.0,
47 global_regex_flags: Optional[int] = re.DOTALL | re.MULTILINE | re.IGNORECASE,
48 version: str = "0.0.1",
49 ):
50 if not supported_entity:
51 raise ValueError("Pattern recognizer should be initialized with entity")
53 if not patterns and not deny_list:
54 raise ValueError(
55 "Pattern recognizer should be initialized with patterns"
56 " or with deny list"
57 )
59 super().__init__(
60 supported_entities=[supported_entity],
61 supported_language=supported_language,
62 name=name,
63 version=version,
64 )
65 if patterns is None:
66 self.patterns = []
67 else:
68 self.patterns = patterns
69 self.context = context
70 self.deny_list_score = deny_list_score
71 self.global_regex_flags = global_regex_flags
73 if deny_list:
74 deny_list_pattern = self._deny_list_to_regex(deny_list)
75 self.patterns.append(deny_list_pattern)
76 self.deny_list = deny_list
77 else:
78 self.deny_list = []
80 def load(self): # noqa: D102
81 pass
83 def analyze(
84 self,
85 text: str,
86 entities: List[str],
87 nlp_artifacts: Optional["NlpArtifacts"] = None,
88 regex_flags: Optional[int] = None,
89 ) -> List[RecognizerResult]:
90 """
91 Analyzes text to detect PII using regular expressions or deny-lists.
93 :param text: Text to be analyzed
94 :param entities: Entities this recognizer can detect
95 :param nlp_artifacts: Output values from the NLP engine
96 :param regex_flags: regex flags to be used in regex matching
97 :return:
98 """
99 results = []
101 if self.patterns:
102 pattern_result = self.__analyze_patterns(text, regex_flags)
103 results.extend(pattern_result)
105 return results
107 def _deny_list_to_regex(self, deny_list: List[str]) -> Pattern:
108 """
109 Convert a list of words to a matching regex.
111 To be analyzed by the analyze method as any other regex patterns.
113 :param deny_list: the list of words to detect
114 :return:the regex of the words for detection
115 """
117 # Escape deny list elements as preparation for regex
118 escaped_deny_list = [re.escape(element) for element in deny_list]
119 regex = r"(?:^|(?<=\W))(" + "|".join(escaped_deny_list) + r")(?:(?=\W)|$)"
120 return Pattern(name="deny_list", regex=regex, score=self.deny_list_score)
122 def validate_result(self, pattern_text: str) -> Optional[bool]:
123 """
124 Validate the pattern logic e.g., by running checksum on a detected pattern.
126 :param pattern_text: the text to validated.
127 Only the part in text that was detected by the regex engine
128 :return: A bool indicating whether the validation was successful.
129 """
130 return None
132 def invalidate_result(self, pattern_text: str) -> Optional[bool]:
133 """
134 Logic to check for result invalidation by running pruning logic.
136 For example, each SSN number group should not consist of all the same digits.
138 :param pattern_text: the text to validated.
139 Only the part in text that was detected by the regex engine
140 :return: A bool indicating whether the result is invalidated
141 """
142 return None
144 @staticmethod
145 def build_regex_explanation(
146 recognizer_name: str,
147 pattern_name: str,
148 pattern: str,
149 original_score: float,
150 validation_result: bool,
151 regex_flags: int,
152 ) -> AnalysisExplanation:
153 """
154 Construct an explanation for why this entity was detected.
156 :param recognizer_name: Name of recognizer detecting the entity
157 :param pattern_name: Regex pattern name which detected the entity
158 :param pattern: Regex pattern logic
159 :param original_score: Score given by the recognizer
160 :param validation_result: Whether validation was used and its result
161 :param regex_flags: Regex flags used in the regex matching
162 :return: Analysis explanation
163 """
164 textual_explanation = (
165 f"Detected by `{recognizer_name}` " f"using pattern `{pattern_name}`"
166 )
168 explanation = AnalysisExplanation(
169 recognizer=recognizer_name,
170 original_score=original_score,
171 pattern_name=pattern_name,
172 pattern=pattern,
173 validation_result=validation_result,
174 regex_flags=regex_flags,
175 textual_explanation=textual_explanation,
176 )
177 return explanation
179 def __analyze_patterns(
180 self, text: str, flags: int = None
181 ) -> List[RecognizerResult]:
182 """
183 Evaluate all patterns in the provided text.
185 Including words in the provided deny-list
187 :param text: text to analyze
188 :param flags: regex flags
189 :return: A list of RecognizerResult
190 """
191 flags = flags if flags else self.global_regex_flags
192 results = []
193 for pattern in self.patterns:
194 match_start_time = datetime.datetime.now()
196 # Compile regex if flags differ from flags the regex was compiled with
197 if not pattern.compiled_regex or pattern.compiled_with_flags != flags:
198 pattern.compiled_with_flags = flags
199 pattern.compiled_regex = re.compile(pattern.regex, flags=flags)
201 try:
202 matches = pattern.compiled_regex.finditer(
203 text, timeout=REGEX_TIMEOUT_SECONDS
204 )
205 match_time = datetime.datetime.now() - match_start_time
206 logger.debug(
207 "--- match_time[%s]: %.6f seconds",
208 pattern.name,
209 match_time.total_seconds(),
210 )
212 for match in matches:
213 start, end = match.span()
214 current_match = text[start:end]
216 # Skip empty results
217 if current_match == "":
218 continue
220 score = pattern.score
222 validation_result = self.validate_result(current_match)
223 description = self.build_regex_explanation(
224 self.name,
225 pattern.name,
226 pattern.regex,
227 score,
228 validation_result,
229 flags,
230 )
231 pattern_result = RecognizerResult(
232 entity_type=self.supported_entities[0],
233 start=start,
234 end=end,
235 score=score,
236 analysis_explanation=description,
237 recognition_metadata={
238 RecognizerResult.RECOGNIZER_NAME_KEY: self.name,
239 RecognizerResult.RECOGNIZER_IDENTIFIER_KEY: self.id,
240 },
241 )
243 if validation_result is not None:
244 if validation_result:
245 pattern_result.score = EntityRecognizer.MAX_SCORE
246 else:
247 pattern_result.score = EntityRecognizer.MIN_SCORE
249 invalidation_result = self.invalidate_result(current_match)
250 if invalidation_result is not None and invalidation_result:
251 pattern_result.score = EntityRecognizer.MIN_SCORE
253 if pattern_result.score > EntityRecognizer.MIN_SCORE:
254 results.append(pattern_result)
256 # Update analysis explanation score after validation or invalidation
257 description.score = pattern_result.score
258 except TimeoutError:
259 logger.warning(
260 "Regex pattern '%s' timed out after %s seconds, skipping.",
261 pattern.name,
262 REGEX_TIMEOUT_SECONDS,
263 exc_info=True,
264 )
266 results = EntityRecognizer.remove_duplicates(results)
267 return results
269 def to_dict(self) -> Dict:
270 """Serialize instance into a dictionary."""
271 return_dict = super().to_dict()
273 return_dict["patterns"] = [pat.to_dict() for pat in self.patterns]
274 return_dict["deny_list"] = self.deny_list
275 return_dict["context"] = self.context
276 return_dict["supported_entity"] = return_dict["supported_entities"][0]
277 del return_dict["supported_entities"]
279 return return_dict
281 @classmethod
282 def from_dict(cls, entity_recognizer_dict: Dict) -> "PatternRecognizer":
283 """Create instance from a serialized dict."""
284 # Make a copy to avoid mutating the input
285 entity_recognizer_dict = entity_recognizer_dict.copy()
287 patterns = entity_recognizer_dict.get("patterns")
288 if patterns:
289 patterns_list = [Pattern.from_dict(pat) for pat in patterns]
290 entity_recognizer_dict["patterns"] = patterns_list
292 # Transform supported_entities (plural) to supported_entity (singular)
293 # PatternRecognizer only accepts supported_entity (singular)
294 if (
295 "supported_entity" in entity_recognizer_dict
296 and "supported_entities" in entity_recognizer_dict
297 ):
298 raise ValueError(
299 "Both 'supported_entity' and 'supported_entities' "
300 "are present in the input dictionary. "
301 "Only one should be provided."
302 )
303 if "supported_entities" in entity_recognizer_dict:
304 supported_entities = entity_recognizer_dict.pop("supported_entities")
305 if supported_entities and len(supported_entities) > 0:
306 # Only set if not already present
307 if "supported_entity" not in entity_recognizer_dict:
308 entity_recognizer_dict["supported_entity"] = supported_entities[0]
310 return cls(**entity_recognizer_dict)