Coverage for presidio_structured / data / data_processors.py: 80%
93 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-03-29 08:58 +0000
1import logging
2from abc import ABC, abstractmethod
3from typing import Any, Callable, Dict, List, Union
5from pandas import DataFrame
6from presidio_anonymizer.entities import OperatorConfig
7from presidio_anonymizer.operators import OperatorsFactory, OperatorType
9from presidio_structured.config import StructuredAnalysis
12class DataProcessorBase(ABC):
13 """Abstract class to handle logic of operations over text using the operators."""
15 def __init__(self) -> None:
16 """Initialize DataProcessorBase object."""
17 self.logger = logging.getLogger("presidio-structured")
19 def operate(
20 self,
21 data: Any,
22 structured_analysis: StructuredAnalysis,
23 operators: Dict[str, OperatorConfig],
24 ) -> Any:
25 """
26 Perform operations over the text using the operators, as per the structured analysis.
28 :param data: Data to be operated on.
29 :param structured_analysis: Analysis schema as per the structured data.
30 :param operators: Dictionary containing operator configuration objects.
31 :return: Data after being operated upon.
32 """ # noqa: E501
33 key_to_operator_mapping = self._generate_operator_mapping(
34 structured_analysis, operators
35 )
36 return self._process(data, key_to_operator_mapping)
38 @abstractmethod
39 def _process(
40 self,
41 data: Union[Dict, DataFrame],
42 key_to_operator_mapping: Dict[str, Callable],
43 ) -> Union[Dict, DataFrame]:
44 """
45 Abstract method for subclasses to provide operation implementation.
47 :param data: Data to be operated on.
48 :param key_to_operator_mapping: Mapping of keys to operators.
49 :return: Operated data.
50 """
51 pass
53 @staticmethod
54 def _create_operator_callable(operator, params):
55 def operator_callable(text):
56 return operator.operate(params=params, text=text)
58 return operator_callable
60 def _generate_operator_mapping(
61 self, config, operators: Dict[str, OperatorConfig]
62 ) -> Dict[str, Callable]:
63 """
64 Generate a mapping of keys to operator callables.
66 :param config: Configuration object containing mapping of entity types to keys.
67 :param operators: Dictionary containing operator configuration objects.
68 :return: Dictionary mapping keys to operator callables.
69 """
70 key_to_operator_mapping = {}
72 operators_factory = OperatorsFactory()
73 for key, entity in config.entity_mapping.items():
74 self.logger.debug(f"Creating operator for key {key} and entity {entity}")
75 operator_config = operators.get(entity, operators.get("DEFAULT", None))
76 if operator_config is None:
77 raise ValueError(f"Operator for entity {entity} not found")
78 # NOTE: hardcoded OperatorType.Anonymize, as this is the only one supported.
79 operator = operators_factory.create_operator_class(
80 operator_config.operator_name, OperatorType.Anonymize
81 )
82 operator_callable = self._create_operator_callable(
83 operator, operator_config.params
84 )
85 key_to_operator_mapping[key] = operator_callable
87 return key_to_operator_mapping
89 def _operate_on_text(
90 self,
91 text_to_operate_on: str,
92 operator_callable: Callable,
93 ) -> str:
94 """
95 Operates on the provided text using the operator callable.
97 :param text_to_operate_on: Text to be operated on.
98 :param operator_callable: Callable that performs operation on the text.
99 :return: Text after operation.
100 """
101 return operator_callable(text_to_operate_on)
104class PandasDataProcessor(DataProcessorBase):
105 """Pandas Data Processor."""
107 def _process(
108 self, data: DataFrame, key_to_operator_mapping: Dict[str, Callable]
109 ) -> DataFrame:
110 """
111 Operates on the given pandas DataFrame based on the provided operators.
113 :param data: DataFrame to be operated on.
114 :param key_to_operator_mapping: Mapping of keys to operator callables.
115 :return: DataFrame after the operation.
116 """
118 if not isinstance(data, DataFrame):
119 raise ValueError("Data must be a pandas DataFrame")
121 for key, operator_callable in key_to_operator_mapping.items():
122 self.logger.debug(f"Operating on column {key}")
123 for row in data.itertuples(index=True):
124 text_to_operate_on = getattr(row, key)
125 operated_text = self._operate_on_text(
126 text_to_operate_on, operator_callable
127 )
128 data.at[row.Index, key] = operated_text
129 return data
132class JsonDataProcessor(DataProcessorBase):
133 """JSON Data Processor, Supports arbitrary nesting of dictionaries and lists."""
135 @staticmethod
136 def _get_nested_value(data: Union[Dict, List, None], path: List[str]) -> Any:
137 """
138 Recursively retrieves the value from nested data using a given path.
140 :param data: Nested data (list or dictionary).
141 :param path: List of keys/indexes representing the path.
142 :return: Retrieved value.
143 """
144 for i, key in enumerate(path):
145 if isinstance(data, list):
146 if key.isdigit():
147 data = data[int(key)]
148 else:
149 return [
150 JsonDataProcessor._get_nested_value(item, path[i:])
151 for item in data
152 ]
153 elif isinstance(data, dict):
154 data = data.get(key)
155 else:
156 return data
157 return data
159 @staticmethod
160 def _set_nested_value(data: Union[Dict, List], path: List[str], value: Any) -> None:
161 """
162 Recursively sets a value in nested data using a given path.
164 :param data: Nested data (JSON-like).
165 :param path: List of keys/indexes representing the path.
166 :param value: Value to be set.
167 """
168 for i, key in enumerate(path):
169 if isinstance(data, list):
170 if i + 1 < len(path) and path[i + 1].isdigit():
171 idx = int(path[i + 1])
172 while len(data) <= idx:
173 data.append({})
174 data = data[idx]
175 continue
176 else:
177 for item in data:
178 JsonDataProcessor._set_nested_value(item, path[i:], value)
179 return
180 elif isinstance(data, dict):
181 if i == len(path) - 1:
182 data[key] = value
183 else:
184 data = data.setdefault(key, {})
186 def _process(
187 self,
188 data: Union[Dict, List],
189 key_to_operator_mapping: Dict[str, Callable],
190 ) -> Union[Dict, List]:
191 """
192 Operates on the given JSON-like data based on the provided configuration.
194 :param data: JSON-like data to be operated on.
195 :param key_to_operator_mapping: maps keys to Callable operators.
196 :return: JSON-like data after the operation.
197 """
199 if not isinstance(data, (dict, list)):
200 raise ValueError("Data must be a JSON-like object")
202 for key, operator_callable in key_to_operator_mapping.items():
203 self.logger.debug(f"Operating on key {key}")
204 keys = key.split(".")
205 if isinstance(data, list):
206 for item in data:
207 self._process(item, key_to_operator_mapping)
208 else:
209 text_to_operate_on = self._get_nested_value(data, keys)
210 if text_to_operate_on:
211 if isinstance(text_to_operate_on, list):
212 for text in text_to_operate_on:
213 operated_text = self._operate_on_text(
214 text, operator_callable
215 )
216 self._set_nested_value(data, keys, operated_text)
217 else:
218 operated_text = self._operate_on_text(
219 text_to_operate_on, operator_callable
220 )
221 self._set_nested_value(data, keys, operated_text)
222 return data