| """Reversible redaction. |
| |
| Replace each detected PII span with a stable placeholder like [PERSON_1]. |
| The same underlying value always maps to the same placeholder, so a document |
| stays internally consistent ("John" referenced twice -> [PERSON_1] twice). |
| |
| The mapping placeholder -> original value is kept locally so a downstream |
| model's answer can be rehydrated without the model ever seeing real PII. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import re |
| from typing import Dict, List, Tuple |
|
|
| from .detect import Entity |
|
|
|
|
| def _slug(label: str) -> str: |
| return re.sub(r"[^A-Z0-9]+", "_", label.upper()).strip("_") |
|
|
|
|
| def redact(text: str, entities: List[Entity]) -> Tuple[str, Dict[str, str]]: |
| """Return (redacted_text, mapping) where mapping is placeholder -> original. |
| |
| Placeholders are assigned per (label, normalized-value) so repeats reuse |
| the same token. Spans are replaced right-to-left to keep offsets valid. |
| """ |
| mapping: Dict[str, str] = {} |
| value_to_token: Dict[Tuple[str, str], str] = {} |
| counters: Dict[str, int] = {} |
|
|
| |
| redacted = text |
| for ent in sorted(entities, key=lambda e: e.start, reverse=True): |
| key = (ent.label, ent.text.strip().lower()) |
| token = value_to_token.get(key) |
| if token is None: |
| slug = _slug(ent.label) |
| counters[slug] = counters.get(slug, 0) + 1 |
| token = f"[{slug}_{counters[slug]}]" |
| value_to_token[key] = token |
| mapping[token] = ent.text |
| redacted = redacted[: ent.start] + token + redacted[ent.end :] |
|
|
| return redacted, mapping |
|
|
|
|
| def rehydrate(text: str, mapping: Dict[str, str]) -> str: |
| """Swap placeholders back to their original values (longest token first to |
| avoid partial overlaps like [X_1] inside [X_10]).""" |
| out = text |
| for token in sorted(mapping, key=len, reverse=True): |
| out = out.replace(token, mapping[token]) |
| return out |
|
|
