pii-detection-pipeline

name	pii-detection-pipeline
description	Build automated PII detection and redaction pipelines using spaCy NER, Microsoft Presidio, and AWS Macie integration. Includes confidence scoring, custom entity type definitions, batch processing workflows, and multi-format document scanning for structured and unstructured data sources.
license	Apache-2.0
metadata	{"author":"mukul975","version":"1.0","domain":"privacy","subdomain":"privacy-engineering","tags":"pii-detection, presidio, spacy-ner, data-redaction, aws-macie"}

Automated PII Detection and Redaction Pipeline

Overview

Automated PII detection is a foundational capability for privacy engineering, enabling organizations to discover, classify, and protect personal data at scale. This skill covers building production-grade PII detection pipelines that combine rule-based pattern matching, machine learning-based Named Entity Recognition (NER), and cloud-native discovery services.

PII Entity Types Catalog

Direct Identifiers

Entity Type	Examples	Detection Method	Risk Level
PERSON_NAME	"John Smith", "Maria Garcia"	NER model	High
EMAIL_ADDRESS	"[e-mail]"	Regex pattern	High
PHONE_NUMBER	"[phone]", "[phone]"	Regex + validation	High
SSN	"123-45-6789"	Regex + checksum	Critical
PASSPORT_NUMBER	"AB1234567"	Regex per country format	Critical
DRIVER_LICENSE	"D123-4567-8901"	Regex per state/country	Critical
CREDIT_CARD	"4111-1111-1111-1111"	Regex + Luhn checksum	Critical
IBAN	"GB82 WEST 1234 5698 7654 32"	Regex + modulo-97 check	High
IP_ADDRESS	"192.168.1.1", "2001:db8::1"	Regex (IPv4/IPv6)	Medium
MAC_ADDRESS	"00:1A:2B:3C:4D:5E"	Regex pattern	Medium

Quasi-Identifiers

Entity Type	Examples	Detection Method	Risk Level
DATE_OF_BIRTH	"1990-01-15", "January 15, 1990"	NER + date parsing	Medium
POSTAL_CODE	"10001", "SW1A 1AA"	Regex per country	Medium
AGE	"35 years old", "age: 42"	NER + context	Low-Medium
GENDER	"male", "female", "non-binary"	Dictionary + context	Low
NATIONALITY	"British", "Japanese"	Dictionary + NER	Low-Medium
LOCATION	"123 Main St", "New York"	NER model	Medium

Sensitive Categories

Entity Type	Examples	Detection Method	Risk Level
MEDICAL_RECORD	"MRN: 12345678"	Regex + context	Critical
HEALTH_CONDITION	"diabetes", "HIV positive"	Medical NER + dictionary	Critical
RELIGIOUS_BELIEF	"Muslim", "Catholic"	Dictionary + context	High
POLITICAL_OPINION	"Democratic Party member"	Dictionary + context	High
SEXUAL_ORIENTATION	"gay", "bisexual"	Dictionary + context	High
BIOMETRIC_DATA	"fingerprint hash: ..."	Context + pattern	Critical
GENETIC_DATA	"BRCA1 positive"	Medical dictionary	Critical

Microsoft Presidio Implementation

Pipeline Architecture

Input Data --> Presidio Analyzer --> Detected Entities --> Presidio Anonymizer --> Redacted Output
                    |                      |                      |
                    v                      v                      v
             +------------+        +-------------+        +---------------+
             | Recognizers|        | Score Filter |        | Operators     |
             | - Pattern  |        | (threshold)  |        | - Replace     |
             | - NER      |        |              |        | - Redact      |
             | - Custom   |        |              |        | - Hash        |
             +------------+        +-------------+        | - Mask        |
                                                          | - Encrypt     |
                                                          +---------------+

Core Implementation

"""
PII detection and redaction pipeline using Microsoft Presidio.
Supports structured and unstructured text with configurable
entity types, confidence thresholds, and redaction strategies.
"""

from presidio_analyzer import AnalyzerEngine, PatternRecognizer, Pattern
from presidio_analyzer.nlp_engine import NlpEngineProvider
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
import json
from dataclasses import dataclass, field


@dataclass
class DetectionResult:
    entity_type: str
    text: str
    start: int
    end: int
    score: float
    source: str


@dataclass
class PipelineConfig:
    language: str = "en"
    score_threshold: float = 0.5
    entities_to_detect: list[str] = field(default_factory=lambda: [
        "PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER", "CREDIT_CARD",
        "US_SSN", "US_DRIVER_LICENSE", "IBAN_CODE", "IP_ADDRESS",
        "LOCATION", "DATE_TIME", "NRP", "MEDICAL_LICENSE",
        "US_PASSPORT", "US_BANK_NUMBER", 
    ])
    redaction_strategy:  =   


 :
    

     ():
        .config = config

        
        nlp_config = {
            : ,
            : [{: config.language, : }]
        }
        nlp_engine = NlpEngineProvider(nlp_configuration=nlp_config).create_engine()

        
        .analyzer = AnalyzerEngine(nlp_engine=nlp_engine)

        
        ._register_custom_recognizers()

        
        .anonymizer = AnonymizerEngine()

     ():
        

        
        nino_pattern = Pattern(
            name=,
            regex=,
            score=
        )
        nino_recognizer = PatternRecognizer(
            supported_entity=,
            patterns=[nino_pattern],
            supported_language=
        )
        .analyzer.registry.add_recognizer(nino_recognizer)

        
        vrn_pattern = Pattern(
            name=,
            regex=,
            score=
        )
        vrn_recognizer = PatternRecognizer(
            supported_entity=,
            patterns=[vrn_pattern],
            supported_language=
        )
        .analyzer.registry.add_recognizer(vrn_recognizer)

        
        emp_id_pattern = Pattern(
            name=,
            regex=,
            score=
        )
        emp_id_recognizer = PatternRecognizer(
            supported_entity=,
            patterns=[emp_id_pattern],
            supported_language=
        )
        .analyzer.registry.add_recognizer(emp_id_recognizer)

     () -> [DetectionResult]:
        
        results = .analyzer.analyze(
            text=text,
            entities=.config.entities_to_detect,
            language=.config.language,
            score_threshold=.config.score_threshold
        )

         [
            DetectionResult(
                entity_type=r.entity_type,
                text=text[r.start:r.end],
                start=r.start,
                end=r.end,
                score=r.score,
                source=r.analysis_explanation.recognizer  r.analysis_explanation  
            )
             r  results
        ]

     () -> [, [DetectionResult]]:
        
        
        analyzer_results = .analyzer.analyze(
            text=text,
            entities=.config.entities_to_detect,
            language=.config.language,
            score_threshold=.config.score_threshold
        )

        
        operators = ._get_operators()

        
        anonymized = .anonymizer.anonymize(
            text=text,
            analyzer_results=analyzer_results,
            operators=operators
        )

        detections = [
            DetectionResult(
                entity_type=r.entity_type,
                text=text[r.start:r.end],
                start=r.start,
                end=r.end,
                score=r.score,
                source=
            )
             r  analyzer_results
        ]

         anonymized.text, detections

     () -> :
        
         .config.redaction_strategy == :
             {: OperatorConfig(, {: })}
         .config.redaction_strategy == :
             {: OperatorConfig(, {: })}
         .config.redaction_strategy == :
             {: OperatorConfig(, {
                : ,
                : ,
                : ,
                : 
            })}
         .config.redaction_strategy == :
             {: OperatorConfig(, {})}
        :
             {: OperatorConfig(, {: })}


 :
    

     ():
        .pipeline = pipeline

     () -> :
        
         pandas  pd

        df = pd.read_csv(input_path)
        stats = {: (df), : {}, : []}

        scan_columns = columns_to_scan  df.columns.tolist()

         col  scan_columns:
             col   df.columns:
                

            stats[].append(col)
            col_detections = []

             idx, value  df[col].items():
                 pd.isna(value):
                    

                text = (value)
                redacted, detections = .pipeline.redact(text)
                df.at[idx, col] = redacted

                 d  detections:
                    col_detections.append(d.entity_type)

            
             entity_type  (col_detections):
                count = col_detections.count(entity_type)
                key = 
                stats[][key] = count

        df.to_csv(output_path, index=)
        stats[] = (stats[].values())

         stats

     () -> :
        
         os

        stats = {
            : ,
            : ,
            : {},
            : []
        }

         file_path  file_paths:
             (file_path, , encoding=)  f:
                text = f.read()

            redacted, detections = .pipeline.redact(text)

            
            output_path = os.path.join(output_dir, os.path.basename(file_path))
             (output_path, , encoding=)  f:
                f.write(redacted)

            
            stats[] += 
            stats[] += (detections)

             d  detections:
                stats[][d.entity_type] = (
                    stats[].get(d.entity_type, ) + 
                )

            
            critical_types = {, , , }
             (d.entity_type  critical_types  d  detections):
                stats[].append(file_path)

         stats

spaCy NER Custom Training

Training Custom Entity Types

"""
Train custom spaCy NER model for domain-specific PII entities.
"""

import spacy
from spacy.training import Example
import random


def create_training_data() -> list[tuple[str, dict]]:
    """
    Create training examples for custom PII entity types.

    Returns list of (text, annotations) tuples.
    """
    training_data = [
        (
            "Patient MRN 12345678 was admitted on 2024-01-15",
            {"entities": [(12, 20, "MEDICAL_RECORD_NUMBER")]}
        ),
        (
            "Employee CEL-00142 reported the incident",
            {"entities": [(9, 18, "EMPLOYEE_ID")]}
        ),
        (
            "Policy holder number PLH-2024-99887 filed a claim",
            {"entities": [(22, 35, "POLICY_NUMBER")]}
        ),
        (
            "The customer with loyalty ID LYL-A1B2C3 requested data export",
            {"entities": [(29, 39, "LOYALTY_ID")]}
        ),
    ]
    return training_data


def train_custom_ner(
    base_model: str = "en_core_web_lg",
    training_data: list = None,
    n_iter: int = 30,
    output_dir: str = "./custom_ner_model"
):
    
     training_data  :
        training_data = create_training_data()

    nlp = spacy.load(base_model)

    
        nlp.pipe_names:
        ner = nlp.add_pipe(, last=)
    :
        ner = nlp.get_pipe()

    
    custom_labels = ()
     _, annotations  training_data:
         ent  annotations.get(, []):
            custom_labels.add(ent[])

     label  custom_labels:
        ner.add_label(label)

    
    optimizer = nlp.resume_training()
    other_pipes = [pipe  pipe  nlp.pipe_names  pipe != ]

     nlp.disable_pipes(*other_pipes):
         iteration  (n_iter):
            random.shuffle(training_data)
            losses = {}

             text, annotations  training_data:
                doc = nlp.make_doc(text)
                example = Example.from_dict(doc, annotations)
                nlp.update([example], drop=, sgd=optimizer, losses=losses)

             iteration %  == :
                ()

    nlp.to_disk(output_dir)
     nlp

AWS Macie Integration

Architecture

S3 Buckets --> Macie Classification Jobs --> Findings --> EventBridge --> Lambda
                                                |                          |
                                                v                          v
                                        Security Hub              Remediation
                                        (centralized)             - Tag sensitive
                                                                  - Encrypt
                                                                  - Notify owner
                                                                  - Quarantine

Macie Job Configuration

"""
AWS Macie integration for cloud-native PII detection in S3 buckets.
"""

import boto3
from datetime import datetime


class MacieIntegration:
    """
    Configure and manage AWS Macie classification jobs for
    automated PII detection across S3 data stores.
    """

    def __init__(self, region: str = "us-east-1"):
        self.macie_client = boto3.client("macie2", region_name=region)

    def create_classification_job(
        self,
        bucket_name: str,
        job_name: str,
        custom_data_identifiers: list[str] = None,
        schedule: str = "ONE_TIME"
    ) -> str:
        """
        Create a Macie classification job for an S3 bucket.

        Args:
            bucket_name: Target S3 bucket
            job_name: Descriptive job name
            custom_data_identifiers: IDs of custom data identifier resources
            schedule: ONE_TIME or SCHEDULED

        Returns:
            Job ID
        """
        job_config = {
            "name": job_name,
            "description": f"PII detection scan for {bucket_name}",
            "jobType": schedule,
            "s3JobDefinition": {
                "bucketDefinitions": [
                    {
                        "accountId": self._get_account_id(),
                        "buckets": [bucket_name]
                    }
                ],
                "scoping": {
                    : {
                        : [
                            {
                                : {
                                    : ,
                                    : ,
                                    : [, , , , , , ]
                                }
                            }
                        ]
                    }
                }
            },
            : ,
            : {
                : ,
                : 
            }
        }

         custom_data_identifiers:
            job_config[] = custom_data_identifiers

        response = .macie_client.create_classification_job(**job_config)
         response[]

     () -> :
        
        params = {
            : name,
            : description,
            : regex,
            : maximum_match_distance
        }

         keywords:
            params[] = keywords

        response = .macie_client.create_custom_data_identifier(**params)
         response[]

     () -> :
        
        sts = boto3.client()
         sts.get_caller_identity()[]

     () -> :
        
        response = .macie_client.list_findings(
            findingCriteria={
                : {
                    : {
                        : [job_id]
                    }
                }
            }
        )

        findings = []
         response[]:
            details = .macie_client.get_findings(findingIds=response[])
            findings = details[]

        summary = {
            : (findings),
            : {},
            : {},
            : []
        }

         finding  findings:
            severity = finding.get(, {}).get(, )
            summary[][severity] = (
                summary[].get(severity, ) + 
            )

            sensitive_data = finding.get(, {}).get(
                , {}
            ).get(, [])

             sd  sensitive_data:
                category = sd.get(, )
                summary[][category] = (
                    summary[].get(category, )
                    + sd.get(, )
                )

            resource = finding.get(, {}).get(, {})
             resource:
                summary[].append(resource.get(, ))

         summary

Confidence Scoring Framework

Score Range	Confidence Level	Recommended Action
0.95 - 1.00	Very High	Auto-redact
0.80 - 0.94	High	Auto-redact with logging
0.60 - 0.79	Medium	Flag for human review
0.40 - 0.59	Low	Log only, no action
0.00 - 0.39	Very Low	Ignore

References

• Microsoft Presidio Documentation: microsoft.github.io/presidio
• spaCy NER Documentation: spacy.io/usage/linguistic-features#named-entities
• AWS Macie Documentation: docs.aws.amazon.com/macie
• Google Cloud DLP API Documentation
• NIST SP 800-188 — De-Identifying Government Datasets
• Article 29 WP Opinion 05/2014 on Anonymisation Techniques