processors/anonymizer.py

"""
Anonymization / pseudonymization of personal data (PII).

Primary strategy: Microsoft Presidio (NER via spaCy + regex recognizers + context).
Fallback strategy: regular expressions only (used if Presidio/spaCy are unavailable).

Per GDPR terminology this is *pseudonymization*: identifiers are replaced by
placeholders and the original text is not stored in readable form (only a hash
is kept for deduplication). The data therefore remains personal data and must be
handled accordingly.
"""

from __future__ import annotations

import re
from typing import Any

# ---------------------------------------------------------------------------
# Try to load Presidio. If it is not installed (or the spaCy model is missing),
# the module gracefully degrades to a regex-only anonymizer so the system never
# crashes on startup.
# ---------------------------------------------------------------------------
_PRESIDIO_AVAILABLE = False
_analyzer = None
_anonymizer = None

# spaCy models to try, from most to least accurate. The Docker image installs
# en_core_web_lg; the smaller ones are accepted as fallbacks.
_SPACY_MODELS = ["en_core_web_lg", "en_core_web_md", "en_core_web_sm"]


def _find_spacy_model() -> str | None:
    try:
        import spacy
    except Exception:
        return None
    for name in _SPACY_MODELS:
        try:
            spacy.load(name)
            return name
        except Exception:
            continue
    return None


try:
    from presidio_analyzer import AnalyzerEngine
    from presidio_analyzer.nlp_engine import NlpEngineProvider
    from presidio_anonymizer import AnonymizerEngine
    from presidio_anonymizer.entities import OperatorConfig

    _model = _find_spacy_model()
    if _model is None:
        raise RuntimeError("no spaCy model available")

    _provider = NlpEngineProvider(
        nlp_configuration={
            "nlp_engine_name": "spacy",
            "models": [{"lang_code": "en", "model_name": _model}],
        }
    )
    _analyzer = AnalyzerEngine(nlp_engine=_provider.create_engine())
    _anonymizer = AnonymizerEngine()
    _PRESIDIO_AVAILABLE = True
    print(f"[anonymizer] Presidio loaded (spaCy model: {_model}).")
except Exception as e:  # pragma: no cover - depends on runtime environment
    print(f"[anonymizer] Presidio unavailable, falling back to regex only: {e}")


# ---------------------------------------------------------------------------
# Regex fallback (also used to enrich Presidio for strictly formatted IDs)
# ---------------------------------------------------------------------------
_EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
_PHONE_RE = re.compile(r"\+?\d[\d\s\-]{7,}\d")
_URL_RE = re.compile(r"https?://\S+|www\.\S+")
_CREDIT_CARD_RE = re.compile(r"\b(?:\d[ -]*?){13,16}\b")

# Map Presidio entity types -> placeholder labels we expose to users.
_PLACEHOLDERS = {
    "PERSON": "[NAME]",
    "EMAIL_ADDRESS": "[EMAIL]",
    "PHONE_NUMBER": "[PHONE]",
    "LOCATION": "[LOCATION]",
    "CREDIT_CARD": "[CARD]",
    "IBAN_CODE": "[IBAN]",
    "IP_ADDRESS": "[IP]",
    "URL": "[URL]",
    "DATE_TIME": "[DATE]",
    "NRP": "[ID]",
    "US_SSN": "[ID]",
}


def _regex_anonymize(text: str) -> str:
    """Pure-regex anonymization for the formatted identifiers we can match safely."""
    text = _EMAIL_RE.sub("[EMAIL]", text)
    text = _URL_RE.sub("[URL]", text)
    text = _CREDIT_CARD_RE.sub("[CARD]", text)
    text = _PHONE_RE.sub("[PHONE]", text)
    return text


def anonymize_text(text: str, language: str = "en") -> str:
    """
    Return a pseudonymized version of ``text``.

    With Presidio: NER-based detection of names/locations + built-in regex
    recognizers for emails, phones, cards, IBANs, etc.
    Without Presidio: regex-only fallback (emails, phones, URLs, cards).
    """
    if not isinstance(text, str) or not text.strip():
        return ""

    if not _PRESIDIO_AVAILABLE:
        return _regex_anonymize(text)

    try:
        results = _analyzer.analyze(text=text, language=language)
        operators = {
            entity: OperatorConfig("replace", {"new_value": placeholder})
            for entity, placeholder in _PLACEHOLDERS.items()
        }
        # default operator for anything detected but not explicitly mapped
        operators["DEFAULT"] = OperatorConfig("replace", {"new_value": "[REDACTED]"})

        anonymized = _anonymizer.anonymize(
            text=text,
            analyzer_results=results,
            operators=operators,
        )
        return anonymized.text
    except Exception as e:  # pragma: no cover
        print(f"[anonymizer] Presidio failed at runtime, using regex fallback: {e}")
        return _regex_anonymize(text)


def detect_pii(text: str, language: str = "en") -> list[dict[str, Any]]:
    """
    Return the list of detected PII spans (for evaluation / debugging).
    Empty list if Presidio is unavailable.
    """
    if not _PRESIDIO_AVAILABLE or not isinstance(text, str) or not text.strip():
        return []
    try:
        results = _analyzer.analyze(text=text, language=language)
        return [
            {
                "type": r.entity_type,
                "start": r.start,
                "end": r.end,
                "score": float(r.score),
                "text": text[r.start : r.end],
            }
            for r in results
        ]
    except Exception:
        return []


def backend_name() -> str:
    """Report which anonymization backend is active (for /health and the UI)."""
    return "presidio" if _PRESIDIO_AVAILABLE else "regex-fallback"