deploy: privacy-sanitizer backend 2026-04-27

Dockerfile

@@ -0,0 +1,34 @@
+# HuggingFace Spaces Docker — read the doc:
+# https://huggingface.co/docs/hub/spaces-sdks-docker
+
+FROM python:3.12-slim
+
+# HF Spaces requires a non-root user with uid 1000
+RUN useradd -m -u 1000 user
+USER user
+ENV PATH="/home/user/.local/bin:$PATH"
+
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+	build-essential \
+	&& rm -rf /var/lib/apt/lists/* || true
+
+COPY --chown=user requirements.txt .
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+
+COPY --chown=user . /app
+
+# Pre-download default model at build time — avoids cold-start delay
+RUN python -c "\
+	from transformers import pipeline; \
+	pipeline('token-classification', \
+	model='iiiorg/piiranha-v1-detect-personal-information', \
+	aggregation_strategy='simple')"
+
+# HuggingFace Spaces requires port 7860
+ENV PORT=7860
+ENV APP_ENV=staging
+EXPOSE 7860
+
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,10 +1,35 @@
 ---
-title: Privacy Sanitizer Api
-emoji: 📉
-colorFrom: yellow
-colorTo: pink
+title: Privacy Sanitizer API
+emoji: 🔒
+colorFrom: purple
+colorTo: blue
 sdk: docker
+app_port: 7860
 pinned: false
+private: true
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Privacy Sanitizer — Backend API
+
+FastAPI backend for the Privacy Text Sanitizer app.  
+NER-based PII detection using HuggingFace Transformers.
+
+## Endpoints
+
+| Method | Path            | Description               |
+| ------ | --------------- | ------------------------- |
+| `GET`  | `/health`       | Health check              |
+| `GET`  | `/api/models`   | List available NER models |
+| `POST` | `/api/sanitize` | Sanitize text             |
+
+## Environment variables (set in Space Secrets)
+
+| Variable             | Description                                                       |
+| -------------------- | ----------------------------------------------------------------- |
+| `FRONTEND_URL`       | Vercel frontend URL for CORS (e.g. `https://your-app.vercel.app`) |
+| `DEFAULT_MODEL_NAME` | Override default NER model                                        |
+
+## Deploy
+
+This Space is deployed from `deploy/hf-space/` in the main repo.  
+Push only the backend `app/` and `requirements.txt` alongside this `Dockerfile`.

app/config.py

@@ -0,0 +1,66 @@
+from pydantic_settings import BaseSettings
+
+
+# Each entry: full metadata shown in the model picker UI
+AVAILABLE_MODELS: list[dict] = [
+    {
+        "id": "iiiorg/piiranha-v1-detect-personal-information",
+        "label": "Piiranha v1",
+        "provider": "iiiorg",
+        "params": "110M",
+        "quality": "good",
+        "quality_score": 2,
+        "speed": "fast",
+        "entity_types": ["PER", "EMAIL", "PHONE", "ADDRESS", "ORG", "LOC", "DATE", "ID"],
+        "description": "Lightweight BERT-based model fine-tuned specifically for PII detection across 8+ entity types. Best for general-purpose anonymization with minimal latency.",
+    },
+    {
+        "id": "dslim/bert-base-NER",
+        "label": "BERT-base NER",
+        "provider": "dslim",
+        "params": "110M",
+        "quality": "good",
+        "quality_score": 2,
+        "speed": "fast",
+        "entity_types": ["PER", "ORG", "LOC", "MISC"],
+        "description": "Standard CoNLL-2003 NER model. High precision on person names, organizations and locations. Limited to 4 entity types — ideal when false positives matter more than coverage.",
+    },
+    {
+        "id": "Jean-Baptiste/roberta-large-ner-english",
+        "label": "RoBERTa-large NER",
+        "provider": "Jean-Baptiste",
+        "params": "355M",
+        "quality": "best",
+        "quality_score": 3,
+        "speed": "slow",
+        "entity_types": ["PER", "ORG", "LOC", "MISC"],
+        "description": "355M parameter RoBERTa fine-tuned on OntoNotes 5.0. Highest accuracy on complex/ambiguous text. Significantly slower — use when precision is critical.",
+    },
+]
+
+DEFAULT_MODEL_ID: str = AVAILABLE_MODELS[0]["id"]
+
+
+class Settings(BaseSettings):
+    # Environment: "local" | "staging" | "production"
+    app_env: str = "local"
+
+    default_model_name: str = DEFAULT_MODEL_ID
+    host: str = "0.0.0.0"
+    port: int = 8000
+
+    # Staging: set FRONTEND_URL=https://your-app.vercel.app in HF Space secrets
+    frontend_url: str = ""
+
+    # Base origins always allowed; staging/production add the deployed frontend URL
+    @property
+    def allowed_origins(self) -> list[str]:
+        origins = ["http://localhost:3000", "http://127.0.0.1:3000"]
+        if self.frontend_url:
+            origins.append(self.frontend_url)
+        return origins
+
+    model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}
+
+
+settings = Settings()

app/main.py

@@ -0,0 +1,36 @@
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from app.config import settings
+from app.routers.sanitize import router as sanitize_router
+from app.services.sanitizer import preload_pipeline
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Pre-load default model at startup; other models are lazy-loaded on first use
+    preload_pipeline(settings.default_model_name)
+    yield
+
+
+app = FastAPI(
+    title="Privacy Text Sanitizer API",
+    version="1.0.0",
+    lifespan=lifespan,
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=settings.allowed_origins,
+    allow_methods=["GET", "POST", "OPTIONS"],
+    allow_headers=["Content-Type"],
+)
+
+app.include_router(sanitize_router, prefix="/api")
+
+
+@app.get("/health")
+async def health() -> dict[str, str]:
+    return {"status": "ok"}

app/routers/sanitize.py

@@ -0,0 +1,22 @@
+from fastapi import APIRouter, HTTPException
+
+from app.config import AVAILABLE_MODELS
+from app.schemas.sanitize import ModelInfo, SanitizeRequest, SanitizeResponse
+from app.services.sanitizer import sanitize_text
+
+router = APIRouter()
+
+
+@router.get("/models", response_model=list[ModelInfo])
+async def list_models() -> list[ModelInfo]:
+    return [ModelInfo(**m) for m in AVAILABLE_MODELS]
+
+
+@router.post("/sanitize", response_model=SanitizeResponse)
+async def sanitize(body: SanitizeRequest) -> SanitizeResponse:
+    try:
+        return sanitize_text(body.text, model_id=body.model_name)
+    except RuntimeError as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail="Model inference failed") from exc

app/schemas/sanitize.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from typing import Optional
+
+from pydantic import BaseModel
+
+
+class SanitizeRequest(BaseModel):
+    text: str
+    model_name: Optional[str] = None
+
+
+class ModelInfo(BaseModel):
+    id: str
+    label: str
+    provider: str
+    params: str
+    quality: str
+    quality_score: int
+    speed: str
+    entity_types: list[str]
+    description: str
+
+
+class EntityDetail(BaseModel):
+    type: str
+    original_value: str
+    replacement: str
+    start: int
+    end: int
+    score: Optional[float] = None
+
+
+class SanitizeStats(BaseModel):
+    total_entities: int
+    by_type: dict[str, int]
+    original_length: int
+    sanitized_length: int
+    sensitive_ratio: float
+
+
+class SanitizeResponse(BaseModel):
+    original_text: str
+    sanitized_text: str
+    entities: list[EntityDetail]
+    stats: SanitizeStats

app/services/sanitizer.py

@@ -0,0 +1,158 @@
+"""
+Sanitizer service — lazy-loads NER pipelines per model and caches them.
+The default pipeline is pre-loaded at startup; additional models are loaded
+on first use and kept in memory for the lifetime of the process.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+from transformers import pipeline as hf_pipeline
+
+from app.schemas.sanitize import EntityDetail, SanitizeResponse, SanitizeStats
+
+# Replacement map: NER label → placeholder
+LABEL_REPLACEMENTS: dict[str, str] = {
+    "PER": "[PRIVATE_PERSON]",
+    "PERSON": "[PRIVATE_PERSON]",
+    "EMAIL": "[PRIVATE_EMAIL]",
+    "PHONE": "[PRIVATE_PHONE]",
+    "PHONE_NUM": "[PRIVATE_PHONE]",
+    "ADDRESS": "[PRIVATE_ADDRESS]",
+    "ADDR": "[PRIVATE_ADDRESS]",
+    "LOC": "[PRIVATE_LOCATION]",
+    "LOCATION": "[PRIVATE_LOCATION]",
+    "GPE": "[PRIVATE_LOCATION]",
+    "ORG": "[PRIVATE_ORGANIZATION]",
+    "ORGANIZATION": "[PRIVATE_ORGANIZATION]",
+    "DATE": "[PRIVATE_DATE]",
+    "TIME": "[PRIVATE_DATE]",
+    "ID": "[PRIVATE_ID]",
+    "ID_NUM": "[PRIVATE_ID]",
+    "ID_NUMBER": "[PRIVATE_ID]",
+}
+
+DEFAULT_REPLACEMENT = "[PRIVATE_DATA]"
+
+# Canonical display label per replacement (for stats grouping)
+REPLACEMENT_TO_TYPE: dict[str, str] = {
+    "[PRIVATE_PERSON]": "PERSON",
+    "[PRIVATE_EMAIL]": "EMAIL",
+    "[PRIVATE_PHONE]": "PHONE",
+    "[PRIVATE_ADDRESS]": "ADDRESS",
+    "[PRIVATE_LOCATION]": "LOCATION",
+    "[PRIVATE_ORGANIZATION]": "ORGANIZATION",
+    "[PRIVATE_DATE]": "DATE",
+    "[PRIVATE_ID]": "ID_NUMBER",
+    "[PRIVATE_DATA]": "OTHER",
+}
+
+# Pipeline cache: model_id → loaded HuggingFace pipeline
+_pipeline_cache: dict[str, Any] = {}
+_default_model: Optional[str] = None
+
+
+def preload_pipeline(model_id: str) -> None:
+    """Pre-load and cache a pipeline (called at startup for the default model)."""
+    global _default_model
+    _get_or_load(model_id)
+    _default_model = model_id
+
+
+def _get_or_load(model_id: str) -> Any:
+    """Return cached pipeline or load it on first use."""
+    if model_id not in _pipeline_cache:
+        _pipeline_cache[model_id] = hf_pipeline(
+            "token-classification",
+            model=model_id,
+            aggregation_strategy="simple",
+        )
+    return _pipeline_cache[model_id]
+
+
+# Keep backward-compatible set_pipeline for tests
+def set_pipeline(pipeline: Any, model_id: str = "__test__") -> None:
+    global _default_model
+    _pipeline_cache[model_id] = pipeline
+    if _default_model is None:
+        _default_model = model_id
+
+
+def sanitize_text(text: str, model_id: Optional[str] = None) -> SanitizeResponse:
+    if not text or not text.strip():
+        return _empty_response(text)
+
+    resolved_model = model_id or _default_model
+    if resolved_model is None:
+        raise RuntimeError("NER pipeline has not been initialised")
+
+    active_pipeline = _get_or_load(resolved_model)
+    raw_entities: list[dict] = active_pipeline(text)
+
+    # Sort by start position descending so replacements don't shift indices
+    raw_entities.sort(key=lambda e: e["start"], reverse=True)
+
+    sanitized = text
+    entity_details: list[EntityDetail] = []
+
+    for ent in raw_entities:
+        label: str = ent.get("entity_group", ent.get("entity", "OTHER")).upper()
+        replacement = LABEL_REPLACEMENTS.get(label, DEFAULT_REPLACEMENT)
+        start: int = ent["start"]
+        end: int = ent["end"]
+        original_value: str = text[start:end]
+        score: Optional[float] = ent.get("score")
+
+        sanitized = sanitized[:start] + replacement + sanitized[end:]
+
+        entity_details.append(
+            EntityDetail(
+                type=REPLACEMENT_TO_TYPE.get(replacement, "OTHER"),
+                original_value=original_value,
+                replacement=replacement,
+                start=start,
+                end=end,
+                score=round(score, 4) if score is not None else None,
+            )
+        )
+
+    # Re-sort entities by original position (ascending) for the response
+    entity_details.sort(key=lambda e: e.start)
+
+    by_type: dict[str, int] = {}
+    for ent in entity_details:
+        by_type[ent.type] = by_type.get(ent.type, 0) + 1
+
+    original_length = len(text)
+    sanitized_length = len(sanitized)
+    sensitive_chars = sum(e.end - e.start for e in entity_details)
+    sensitive_ratio = round(sensitive_chars / original_length, 4) if original_length > 0 else 0.0
+
+    return SanitizeResponse(
+        original_text=text,
+        sanitized_text=sanitized,
+        entities=entity_details,
+        stats=SanitizeStats(
+            total_entities=len(entity_details),
+            by_type=by_type,
+            original_length=original_length,
+            sanitized_length=sanitized_length,
+            sensitive_ratio=sensitive_ratio,
+        ),
+    )
+
+
+def _empty_response(text: str) -> SanitizeResponse:
+    return SanitizeResponse(
+        original_text=text,
+        sanitized_text=text,
+        entities=[],
+        stats=SanitizeStats(
+            total_entities=0,
+            by_type={},
+            original_length=len(text),
+            sanitized_length=len(text),
+            sensitive_ratio=0.0,
+        ),
+    )

requirements.txt

@@ -0,0 +1,7 @@
+fastapi==0.115.12
+uvicorn[standard]==0.34.2
+pydantic==2.11.4
+pydantic-settings==2.9.1
+transformers==4.51.3
+torch==2.7.0
+python-dotenv==1.1.0