app.py

"""Server-side NER scrubbing service.

Runs Presidio + spaCy to catch names, locations, and organizations
that client-side regex scrubbing can't detect. Deployed on HuggingFace
Spaces (free tier) as a Docker SDK Space.

The client already handles structured PII (emails, phones, SSNs, IPs,
file paths, API keys). This service is a defense-in-depth layer that
catches unstructured PII (names mentioned in conversation text).
"""

import os

from fastapi import FastAPI, Header, HTTPException
from presidio_analyzer import AnalyzerEngine
from presidio_analyzer.nlp_engine import NlpEngineProvider
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
from pydantic import BaseModel

app = FastAPI(title="Common Parlance NER Service", docs_url=None, redoc_url=None)

# Initialize once at startup (not per-request)
nlp_provider = NlpEngineProvider(nlp_configuration={
    "nlp_engine_name": "spacy",
    "models": [{"lang_code": "en", "model_name": "en_core_web_sm"}],
})
analyzer = AnalyzerEngine(nlp_engine=nlp_provider.create_engine())
anonymizer = AnonymizerEngine()

API_KEY = os.environ.get("API_KEY", "")

# Only detect entity types that regex can't handle.
# Emails, phones, IPs, etc. are already scrubbed client-side.
NER_ENTITIES = ["PERSON", "LOCATION", "ORGANIZATION"]

OPERATORS = {
    "PERSON": OperatorConfig("replace", {"new_value": "[NAME]"}),
    "LOCATION": OperatorConfig("replace", {"new_value": "[LOCATION]"}),
    "ORGANIZATION": OperatorConfig("replace", {"new_value": "[ORG]"}),
}


class ScrubRequest(BaseModel):
    turns: list[dict]


class ScrubResponse(BaseModel):
    turns: list[dict]
    entities_found: int


MAX_TURNS = 200
MAX_CONTENT_LENGTH = 100_000  # 100KB per turn


@app.post("/scrub", response_model=ScrubResponse)
async def scrub(payload: ScrubRequest, x_api_key: str = Header(None)):
    if API_KEY and x_api_key != API_KEY:
        raise HTTPException(status_code=401, detail="Invalid API key")

    if len(payload.turns) > MAX_TURNS:
        raise HTTPException(status_code=413, detail=f"Too many turns (max {MAX_TURNS})")

    total_entities = 0
    scrubbed_turns = []

    for turn in payload.turns:
        if len(turn.get("content", "")) > MAX_CONTENT_LENGTH:
            raise HTTPException(
                status_code=413,
                detail=f"Turn content too large (max {MAX_CONTENT_LENGTH} bytes)",
            )
        text = turn.get("content", "")
        role = turn.get("role", "")

        results = analyzer.analyze(
            text=text,
            entities=NER_ENTITIES,
            language="en",
            score_threshold=0.5,
        )

        if results:
            anonymized = anonymizer.anonymize(
                text=text,
                analyzer_results=results,
                operators=OPERATORS,
            )
            text = anonymized.text
            total_entities += len(results)

        scrubbed_turns.append({"role": role, "content": text})

    return ScrubResponse(turns=scrubbed_turns, entities_found=total_entities)


@app.get("/health")
async def health():
    return {"ok": True, "model": "en_core_web_sm", "entities": NER_ENTITIES}