Presidio Python 教程：敏感信息识别、误报漏报与脱敏结果验证

python3.11 -m venv .venv
source .venv/bin/activate
python -m pip install 'presidio-analyzer==2.2.364' 'presidio-anonymizer==2.2.364'

from presidio_analyzer import Pattern, PatternRecognizer
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig

# Deliberately narrow synthetic rules: no NER model or cloud API is loaded.
RULES = [
    PatternRecognizer(
        supported_entity="EMAIL_ADDRESS",
        patterns=[Pattern("ascii_email", r"(?<![A-Za-z0-9._%+-])[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}(?![A-Za-z0-9.-])", 0.8)],
    ),
    PatternRecognizer(
        supported_entity="ACCOUNT_CODE",
        patterns=[Pattern("synthetic_account", r"(?<![A-Za-z0-9_])acct_[0-9]{4}(?![A-Za-z0-9_])", 0.8)],
    ),
]
OPERATORS = {
    "EMAIL_ADDRESS": OperatorConfig("replace", {"new_value": "[EMAIL]"}),
    "ACCOUNT_CODE": OperatorConfig("replace", {"new_value": "[ACCOUNT]"}),
}
ENGINE = AnonymizerEngine()


def detect(text):
    hits = [h for rule in RULES for h in rule.analyze(text=text, entities=rule.supported_entities)
            if h.score >= 0.5]
    return sorted(hits, key=lambda h: (h.start, h.end, h.entity_type))


def validate_spans(text, hits):
    previous_end = 0
    for hit in sorted(hits, key=lambda h: (h.start, h.end)):
        if not (0 <= hit.start < hit.end <= len(text)):
            raise ValueError("Invalid character span")
        if hit.start < previous_end:
            raise ValueError("Resolve overlapping spans before transformation")
        if hit.entity_type not in OPERATORS:
            raise ValueError("No approved operator")
        previous_end = hit.end


def redact(text):
    hits = detect(text)
    validate_spans(text, hits)
    result = ENGINE.anonymize(text=text, analyzer_results=hits, operators=OPERATORS)
    # Return no matched values or original text as metadata.
    return result.text, {"detected_count": len(hits), "ruleset": "synthetic-v1"}


def verify_known_values(output, values):
    # This checks only explicitly labelled fixtures; it cannot prove anonymity.
    if any(value and value in output for value in values):
        raise AssertionError("Known fixture value remains")


def exact_span_counts(text, expected):
    predicted = {(h.entity_type, h.start, h.end) for h in detect(text)}
    expected = set(expected)
    return {"tp": len(predicted & expected), "fp": len(predicted - expected),
            "fn": len(expected - predicted)}

python - <<'PY'
from pii_pipeline import redact
output, metadata = redact('客服🙂[email protected]；账号 acct_1234；订单 ord_1234')
assert output == '客服🙂[EMAIL]；账号 [ACCOUNT]；订单 ord_1234'
assert metadata['detected_count'] == 2
print(output)
PY

客服🙂[EMAIL]；账号 [ACCOUNT]；订单 ord_1234

import json
import unittest
from importlib.metadata import version
from pathlib import Path
from presidio_analyzer import RecognizerResult
from pii_pipeline import detect, redact, validate_spans, verify_known_values, exact_span_counts


class PipelineTests(unittest.TestCase):
    def test_unicode_and_repeated_values(self):
        text = "客服🙂[email protected]；再次 [email protected]"
        hits = detect(text)
        self.assertEqual([text[h.start:h.end] for h in hits], ["[email protected]"] * 2)
        output, meta = redact(text)
        self.assertEqual(output, "客服🙂[EMAIL]；再次 [EMAIL]")
        self.assertEqual(meta, {"detected_count": 2, "ruleset": "synthetic-v1"})
        verify_known_values(output, ["[email protected]"])

    def test_account_and_order_number(self):
        output, _ = redact("账号 acct_1234，订单 ord_1234")
        self.assertEqual(output, "账号 [ACCOUNT]，订单 ord_1234")

    def test_start_and_end(self):
        output, _ = redact("[email protected] acct_1234")
        self.assertEqual(output, "[EMAIL] [ACCOUNT]")

    def test_invalid_span_rejected(self):
        with self.assertRaises(ValueError):
            validate_spans("abc", [RecognizerResult("EMAIL_ADDRESS", 0, 10, 0.8)])

    def test_overlap_rejected(self):
        with self.assertRaises(ValueError):
            validate_spans("abcdef", [RecognizerResult("EMAIL_ADDRESS", 0, 4, 0.8),
                                      RecognizerResult("ACCOUNT_CODE", 3, 6, 0.8)])

    def test_known_leak_verifier(self):
        with self.assertRaises(AssertionError):
            verify_known_values("[email protected]", ["[email protected]"])

    def test_zero_hits_is_not_proof(self):
        text = "demo [at] example [dot] test"
        output, meta = redact(text)
        self.assertEqual(meta["detected_count"], 0)
        self.assertEqual(output, text)
        counts = exact_span_counts(text, [("EMAIL_ADDRESS", 0, len(text))])
        self.assertEqual(counts, {"tp": 0, "fp": 0, "fn": 1})

    def test_measure_false_positive_and_false_negative(self):
        # Policy: the manual's public example should be preserved; customer email removed.
        positive = "客户 [email protected]"
        negative = "手册样例 [email protected]"
        missed = "demo [at] example [dot] test"
        cases = [(positive, [("EMAIL_ADDRESS", positive.index("demo"), len(positive))]),
                 (negative, []), (missed, [("EMAIL_ADDRESS", 0, len(missed))])]
        total = {k: 0 for k in ("tp", "fp", "fn")}
        for text, expected in cases:
            for k, value in exact_span_counts(text, expected).items():
                total[k] += value
        self.assertEqual(total, {"tp": 1, "fp": 1, "fn": 1})


if __name__ == "__main__":
    suite = unittest.defaultTestLoader.loadTestsFromTestCase(PipelineTests)
    result = unittest.TextTestRunner(verbosity=2).run(suite)
    report = {"date": "2026-10-18", "python": "3.11", "packages": {
        p: version(p) for p in ["presidio-analyzer", "presidio-anonymizer"]},
        "tests_run": result.testsRun, "failures": len(result.failures), "errors": len(result.errors),
        "synthetic_metric_fixture": {"tp": 1, "fp": 1, "fn": 1, "precision": 0.5, "recall": 0.5},
        "scope": "Custom ASCII email/account patterns and transformation contracts only; no NER, real PII or cloud API benchmark"}
    Path(__file__).with_name("snippet-checks.json").write_text(json.dumps(report, ensure_ascii=False, indent=2))
    raise SystemExit(0 if result.wasSuccessful() else 1)

python test_pipeline.py