| from presidio_analyzer import AnalyzerEngine, PatternRecognizer, Pattern |
| from presidio_analyzer.nlp_engine import NlpEngineProvider |
| from presidio_anonymizer import AnonymizerEngine |
|
|
| _provider = NlpEngineProvider(nlp_configuration={ |
| "nlp_engine_name": "spacy", |
| "models": [{"lang_code": "en", "model_name": "en_core_web_lg"}] |
| }) |
|
|
| analyzer = AnalyzerEngine(nlp_engine=_provider.create_engine()) |
| anonymizer = AnonymizerEngine() |
|
|
| analyzer.registry.add_recognizer(PatternRecognizer( |
| supported_entity="ACCOUNT_NUMBER", |
| patterns=[Pattern(name="account_number", regex=r"\d{3}-\d{3}-\d{3}", score=0.8)], |
| context=["account", "acct", "num", "number", "no.", "#"] |
| )) |
|
|
| _ENTITIES = ["ACCOUNT_NUMBER", "PERSON", "PHONE_NUMBER", "EMAIL_ADDRESS", "CREDIT_CARD"] |
|
|
| |
| def check_pii(text: str) -> bool: |
| """Returns True if PII is detected — used by planner to block the query.""" |
| results = analyzer.analyze(text=text, language="en", entities=_ENTITIES) |
| return len(results) > 0 |
|
|
| def scrub_pii(text: str) -> str: |
| """Replaces PII with placeholders — use this if you want to proceed with a sanitised query instead of blocking.""" |
| results = analyzer.analyze(text=text, language="en", entities=_ENTITIES) |
| return anonymizer.anonymize(text=text, analyzer_results=results).text |
