app/rag/anonymizer.py

"""PII scrubbing for the adjudicated-cases collection.

Closed cases are stored as anonymized clinical summaries: no names, no member or
claim identifiers, no dates. The metadata allowlist plus the forbidden-key check
defend against future call sites accidentally leaking identifiers into Chroma.
"""

from __future__ import annotations

import hashlib

ALLOWED_METADATA = frozenset(
    {
        "case_ref",
        "procedure_code",
        "diagnosis_code",
        "modality",
        "claim_type",
        "recommendation",
        "decision",
    }
)

# Hard-forbidden identifier keys: passing any of these raises instead of dropping,
# so a leaking call site fails loudly in tests rather than silently storing PII.
_FORBIDDEN_KEYS = frozenset({"claimant_id", "member_id", "claim_id", "user_id", "patient_id"})
_FORBIDDEN_FRAGMENTS = ("name",)  # full_name, patient_name, first_name, ...


def build_case_summary(
    *,
    modality: str | None,
    claim_type: str,
    procedure_code: str,
    diagnosis_code: str,
    recommendation: str | None,
    key_findings: list[str],
    decision: str,
) -> str:
    """One compact, PII-free paragraph describing a closed case.

    Deliberately takes no names, identifiers, or dates, the signature is the
    anonymization boundary.
    """
    modality_phrase = f"{modality} imaging" if modality else "no imaging modality recorded"
    findings_phrase = "; ".join(key_findings) if key_findings else "none recorded"
    recommendation_phrase = recommendation if recommendation else "none recorded"
    return (
        f"Closed {claim_type} claim with {modality_phrase}, "
        f"procedure code {procedure_code}, diagnosis code {diagnosis_code}. "
        f"Specialist recommendation: {recommendation_phrase}. "
        f"Key findings: {findings_phrase}. "
        f"Final decision: {decision}."
    )


def make_case_ref(claim_ref: str) -> str:
    """Deterministic case reference, unlinkable without the original claim_ref."""
    digest = hashlib.sha256(claim_ref.encode("utf-8")).hexdigest()
    return "CASE-" + digest[:8].upper()


def anonymized_metadata(**kwargs: object) -> dict[str, object]:
    """Filter metadata down to the allowlist.

    Drops any key not in ALLOWED_METADATA (and None values, which Chroma rejects).
    Raises ValueError for identifier-like keys so leaks fail loudly.
    """
    for key in kwargs:
        lowered = key.lower()
        if lowered in _FORBIDDEN_KEYS or any(frag in lowered for frag in _FORBIDDEN_FRAGMENTS):
            raise ValueError(f"identifier-like metadata key not allowed: {key!r}")
    return {
        key: value for key, value in kwargs.items() if key in ALLOWED_METADATA and value is not None
    }