Spaces:
Sleeping
Sleeping
File size: 3,118 Bytes
73376a6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine, DeanonymizeEngine, OperatorConfig
from presidio_anonymizer.operators import Operator, OperatorType
from typing import Dict
from pprint import pprint
# text = "Peter gave his book to Heidi which later gave it to Nicole. Peter lives in London and Nicole lives in Tashkent."
# print("original text:")
# pprint(text)
# analyzer = AnalyzerEngine()
# analyzer_results = analyzer.analyze(text=text, language="en")
# # print("analyzer results:")
# # pprint(analyzer_results)
class InstanceCounterAnonymizer(Operator):
"""
Anonymizer which replaces the entity value
with an instance counter per entity.
"""
REPLACING_FORMAT = "<{entity_type}_{index}>"
def operate(self, text: str, params: Dict = None) -> str:
"""Anonymize the input text."""
entity_type: str = params["entity_type"]
# entity_mapping is a dict of dicts containing mappings per entity type
entity_mapping: Dict[Dict:str] = params["entity_mapping"]
entity_mapping_for_type = entity_mapping.get(entity_type)
if not entity_mapping_for_type:
new_text = self.REPLACING_FORMAT.format(
entity_type=entity_type, index=0
)
entity_mapping[entity_type] = {}
else:
if text in entity_mapping_for_type:
return entity_mapping_for_type[text]
previous_index = self._get_last_index(entity_mapping_for_type)
new_text = self.REPLACING_FORMAT.format(
entity_type=entity_type, index=previous_index + 1
)
entity_mapping[entity_type][text] = new_text
return new_text
@staticmethod
def _get_last_index(entity_mapping_for_type: Dict) -> int:
"""Get the last index for a given entity type."""
def get_index(value: str) -> int:
return int(value.split("_")[-1][:-1])
indices = [get_index(v) for v in entity_mapping_for_type.values()]
return max(indices)
def validate(self, params: Dict = None) -> None:
"""Validate operator parameters."""
if "entity_mapping" not in params:
raise ValueError("An input Dict called `entity_mapping` is required.")
if "entity_type" not in params:
raise ValueError("An entity_type param is required.")
def operator_name(self) -> str:
return "entity_counter"
def operator_type(self) -> OperatorType:
return OperatorType.Anonymize
# Create Anonymizer engine and add the custom anonymizer
# anonymizer_engine = AnonymizerEngine()
# anonymizer_engine.add_anonymizer(InstanceCounterAnonymizer)
# # Create a mapping between entity types and counters
# entity_mapping = dict()
# # Anonymize the text
# anonymized_result = anonymizer_engine.anonymize(
# text,
# analyzer_results,
# {
# "DEFAULT": OperatorConfig(
# "entity_counter", {"entity_mapping": entity_mapping}
# )
# },
# )
# print(anonymized_result.text)
# pprint(entity_mapping, indent=2) |