windlass.providers.preprocessors.privacy

privacy

PII detection and redaction.

Once personal data is embedded into a vector index it is very hard to get back out — you cannot un-ring that bell, and "delete the row" does not undo the copies in your backups. Redacting at ingestion is the only reliable point of control.

This preprocessor ships a dependency-free detector covering the categories that actually turn up in enterprise documents (email, phone, SSN, credit card, IBAN, IP, passport, API keys), with a Luhn check on card numbers to keep false positives down. When presidio-analyzer is installed it is used instead, for NER-based detection of names, locations and organisations.

Example

from windlass.core.types import Document p = PIIPreprocessor() p.process([Document(content="Reach me at ada@example.com or 555-123-4567")])[0].content 'Reach me at [EMAIL] or [PHONE]'

PIIMatch

Bases: NamedTuple

One detected piece of personal data.

Attributes:

Name	Type	Description
kind	str	Category, e.g. "email".
value	str	The matched text.
start	int	Start offset in the source string.
end	int	Exclusive end offset.

PIIPreprocessor

PIIPreprocessor(
    *,
    kinds: Sequence[str] | None = None,
    action: str = "redact",
    placeholder: str | None = None,
    use_presidio: bool = False,
    language: str = "en",
    **config: Any
)

Bases: Preprocessor

Detects and optionally redacts personal data.

Parameters:

Name	Type	Description	Default
kinds	Sequence[str] | None	Categories to act on. None means every category in :data:PII_PATTERNS.	None
action	str	"redact" replaces matches, "drop" discards any document containing PII, "tag" only records what it found.	'redact'
placeholder	str | None	Fixed replacement token, or None for per-category tokens.	None
use_presidio	bool	Use presidio-analyzer when installed, adding NER-based detection of names, locations and organisations.	False
language	str	Language passed to Presidio.	'en'
**config	Any	Forwarded to :class:~windlass.interfaces.preprocessor.Preprocessor.	{}

Raises:

Type	Description
ValueError	For an unknown action.

Note

Regex detection is precise for structured identifiers and blind to unstructured ones — it will not find "Ada Lovelace" as a person's name. Install windlass[pii] and set use_presidio=True when you need that, and treat any automated detector as a control, not a guarantee.

Example

from windlass.core.types import Document p = PIIPreprocessor(action="tag") p.process([Document(content="a@b.com")])[0].metadata["pii_kinds"]['email']

Source code in src\windlass\providers\preprocessors\privacy.py

def __init__(
    self,
    *,
    kinds: Sequence[str] | None = None,
    action: str = "redact",
    placeholder: str | None = None,
    use_presidio: bool = False,
    language: str = "en",
    **config: Any,
) -> None:
    if action not in {"redact", "drop", "tag"}:
        raise ValueError("action must be 'redact', 'drop' or 'tag'")
    super().__init__(**config)
    self.kinds = list(kinds) if kinds else None
    self.action = action
    self.placeholder = placeholder
    self.language = language
    self.use_presidio = use_presidio and is_available("presidio_analyzer")
    if use_presidio and not self.use_presidio:
        self._log.warning(
            "presidio-analyzer is not installed; falling back to regex detection. "
            'Install it with: pip install "windlass[pii]"'
        )
    self._analyzer: Any = None

aprocess_one async

aprocess_one(document: Document) -> list[Document]

Scan a document and apply the configured action.

Parameters:

Name	Type	Description	Default
document	Document	The document to scan.	required

Returns:

Type	Description
list[Document]	The processed document, or [] when action='drop' and PII was
list[Document]	found.

Source code in src\windlass\providers\preprocessors\privacy.py

async def aprocess_one(self, document: Document) -> list[Document]:
    """Scan a document and apply the configured action.

    Args:
        document: The document to scan.

    Returns:
        The processed document, or ``[]`` when ``action='drop'`` and PII was
        found.
    """
    if self.use_presidio:
        text, matches = self._presidio(document.content)
    else:
        text, matches = redact_pii(document.content, self.kinds, placeholder=self.placeholder)

    if not matches:
        return [document]

    kinds = sorted({m.kind for m in matches})
    if self.action == "drop":
        self._log.info(
            "Dropping %s: contains %s", document.source or document.id, ", ".join(kinds)
        )
        return []

    metadata = {
        **document.metadata,
        "pii_detected": True,
        "pii_kinds": kinds,
        "pii_count": len(matches),
    }
    content = text if self.action == "redact" else document.content
    return [document.model_copy(update={"content": content, "metadata": metadata})]

detect_pii

detect_pii(text: str, kinds: Sequence[str] | None = None) -> list[PIIMatch]

Find personal data in text.

Overlapping matches are resolved in favour of the earlier, longer one, so an email address is never also reported as a phone number.

Parameters:

Name	Type	Description	Default
text	str	The text to scan.	required
kinds	Sequence[str] | None	Categories to look for. None scans every category in :data:PII_PATTERNS.	None

Returns:

Type	Description
list[PIIMatch]	Matches sorted by position.

Example

[m.kind for m in detect_pii("write to a@b.com")]['email']

Source code in src\windlass\providers\preprocessors\privacy.py

def detect_pii(text: str, kinds: Sequence[str] | None = None) -> list[PIIMatch]:
    """Find personal data in ``text``.

    Overlapping matches are resolved in favour of the earlier, longer one, so an
    email address is never also reported as a phone number.

    Args:
        text: The text to scan.
        kinds: Categories to look for. ``None`` scans every category in
            :data:`PII_PATTERNS`.

    Returns:
        Matches sorted by position.

    Example:
        >>> [m.kind for m in detect_pii("write to a@b.com")]
        ['email']
    """
    wanted = list(kinds) if kinds else list(PII_PATTERNS)
    found: list[PIIMatch] = []

    for kind in wanted:
        pattern = PII_PATTERNS.get(kind)
        if pattern is None:
            continue
        for match in pattern.finditer(text):
            value = match.group(0)
            if kind == "credit_card" and not _luhn(value):
                continue
            if kind == "phone" and sum(c.isdigit() for c in value) < 7:
                continue
            found.append(PIIMatch(kind, value, match.start(), match.end()))

    found.sort(key=lambda m: (m.start, -(m.end - m.start)))
    resolved: list[PIIMatch] = []
    cursor = -1
    for candidate in found:
        if candidate.start >= cursor:
            resolved.append(candidate)
            cursor = candidate.end
    return resolved

redact_pii

redact_pii(
    text: str, kinds: Sequence[str] | None = None, *, placeholder: str | None = None
) -> tuple[str, list[PIIMatch]]

Replace personal data in text with placeholders.

Parameters:

Name	Type	Description	Default
text	str	The text to redact.	required
kinds	Sequence[str] | None	Categories to redact. None redacts every category.	None
placeholder	str | None	Fixed replacement for every match. None uses a per-category token such as [EMAIL], which preserves the semantic hint that something was there.	None

Returns:

Type	Description
tuple[str, list[PIIMatch]]	A (redacted_text, matches) pair.

Example

redact_pii("call 555-123-4567")[0] 'call [PHONE]'

Source code in src\windlass\providers\preprocessors\privacy.py

def redact_pii(
    text: str,
    kinds: Sequence[str] | None = None,
    *,
    placeholder: str | None = None,
) -> tuple[str, list[PIIMatch]]:
    """Replace personal data in ``text`` with placeholders.

    Args:
        text: The text to redact.
        kinds: Categories to redact. ``None`` redacts every category.
        placeholder: Fixed replacement for every match. ``None`` uses a
            per-category token such as ``[EMAIL]``, which preserves the semantic
            hint that *something* was there.

    Returns:
        A ``(redacted_text, matches)`` pair.

    Example:
        >>> redact_pii("call 555-123-4567")[0]
        'call [PHONE]'
    """
    matches = detect_pii(text, kinds)
    if not matches:
        return text, []
    out: list[str] = []
    cursor = 0
    for match in matches:
        out.append(text[cursor : match.start])
        out.append(placeholder or _PLACEHOLDERS.get(match.kind, "[REDACTED]"))
        cursor = match.end
    out.append(text[cursor:])
    return "".join(out), matches