Pasul 1 – Instalare

Pe un server Linux sau Windows cu Python 3.10+:

python -m venv /opt/pii-guard/venv
source /opt/pii-guard/venv/bin/activate

pip install presidio-analyzer presidio-anonymizer fastapi uvicorn httpx
# Model spaCy pentru limba română (recunoaștere nume, organizații, locații)
python -m spacy download ro_core_news_lg

Presidio vine implicit configurat pentru engleză. Pentru texte românești îl configurăm explicit cu modelul spaCy românesc și cu recunoaștere pentru date specifice României.

Pasul 2 – Recunoaștere pentru CNP cu validarea cifrei de control

Un regex simplu de 13 cifre ar marca și numere de comandă sau de telefon. Validăm deci structura CNP-ului (sex/secol, lună, zi, județ) și cifra de control cu cheia oficială 279146358279.

recognizers_ro.py:

from presidio_analyzer import Pattern, PatternRecognizer

class RoCnpRecognizer(PatternRecognizer):
    """Detectează CNP-uri românești valide (structură + cifră de control)."""

    CONTROL_KEY = "279146358279"

    def __init__(self):
        patterns = [Pattern("CNP", r"\b[1-9]\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{6}\b", 0.5)]
        super().__init__(
            supported_entity="RO_CNP",
            patterns=patterns,
            context=["cnp", "cod numeric personal", "c.n.p"],
            supported_language="ro",
        )

    def validate_result(self, pattern_text: str) -> bool:
        if len(pattern_text) != 13 or not pattern_text.isdigit():
            return False
        judet = int(pattern_text[7:9])
        if not (1 <= judet <= 52):
            return False
        s = sum(int(c) * int(k) for c, k in zip(pattern_text[:12], self.CONTROL_KEY))
        control = s % 11
        control = 1 if control == 10 else control
        return control == int(pattern_text[12])


class RoPhoneRecognizer(PatternRecognizer):
    """Telefoane mobile și fixe românești: 07xx xxx xxx, +40 7xx..., 021 xxx xx xx."""

    def __init__(self):
        patterns = [
            Pattern("RO_MOBIL", r"(?<!\d)(?:\+40|0040|0)\s?7\d{2}[\s.-]?\d{3}[\s.-]?\d{3}(?!\d)", 0.7),
            Pattern("RO_FIX",   r"(?<!\d)(?:\+40|0040|0)\s?[23]\d{1,2}[\s.-]?\d{3}[\s.-]?\d{2,3}[\s.-]?\d{0,2}(?!\d)", 0.4),
        ]
        super().__init__(
            supported_entity="PHONE_NUMBER",
            patterns=patterns,
            context=["tel", "telefon", "mobil", "nr"],
            supported_language="ro",
        )

validate_result este apelat de Presidio pentru fiecare potrivire. Dacă întoarce False, rezultatul este eliminat, deci un număr aleator de 13 cifre nu va fi tratat ca CNP.

Pasul 3 – Motorul de analiză configurat pentru română

engine.py:

from presidio_analyzer import AnalyzerEngine, RecognizerRegistry
from presidio_analyzer.nlp_engine import NlpEngineProvider
from presidio_analyzer.predefined_recognizers import (
    SpacyRecognizer, EmailRecognizer, IbanRecognizer, CreditCardRecognizer, IpRecognizer,
)
from recognizers_ro import RoCnpRecognizer, RoPhoneRecognizer

NLP_CONFIG = {
    "nlp_engine_name": "spacy",
    "models": [{"lang_code": "ro", "model_name": "ro_core_news_lg"}],
    "ner_model_configuration": {
        "model_to_presidio_entity_mapping": {
            "PERSON": "PERSON",
            "ORGANIZATION": "ORGANIZATION",
            "GPE": "LOCATION",
            "LOC": "LOCATION",
        },
        "labels_to_ignore": ["DATETIME", "NUMERIC_VALUE", "ORDINAL", "MONEY", "QUANTITY", "PERIOD"],
    },
}

def build_analyzer() -> AnalyzerEngine:
    nlp_engine = NlpEngineProvider(nlp_configuration=NLP_CONFIG).create_engine()

    registry = RecognizerRegistry(supported_languages=["ro"])
    registry.add_recognizer(SpacyRecognizer(supported_language="ro",
                                            supported_entities=["PERSON", "LOCATION", "ORGANIZATION"]))
    registry.add_recognizer(EmailRecognizer(supported_language="ro"))
    registry.add_recognizer(IbanRecognizer(supported_language="ro"))
    registry.add_recognizer(CreditCardRecognizer(supported_language="ro"))
    registry.add_recognizer(IpRecognizer(supported_language="ro"))
    registry.add_recognizer(RoCnpRecognizer())
    registry.add_recognizer(RoPhoneRecognizer())

    return AnalyzerEngine(nlp_engine=nlp_engine, registry=registry, supported_languages=["ro"])

IbanRecognizer din Presidio validează deja checksum-ul IBAN, deci funcționează corect și pentru conturile românești (RO + 22 caractere).

Pasul 4 – Anonimizare reversibilă

Presidio Anonymizer poate înlocui sau masca valorile, dar pentru de-anonimizare avem nevoie de o mapare placeholder ↔ valoare păstrată doar în memorie, pe durata cererii. Aceeași valoare primește același placeholder, ca LLM-ul să înțeleagă că e vorba de aceeași persoană.

pii_guard.py:

from engine import build_analyzer

ENTITY_LABELS = {
    "PERSON": "PERSOANA", "LOCATION": "LOCATIE", "ORGANIZATION": "ORGANIZATIE",
    "EMAIL_ADDRESS": "EMAIL", "IBAN_CODE": "IBAN", "RO_CNP": "CNP",
    "PHONE_NUMBER": "TELEFON", "CREDIT_CARD": "CARD", "IP_ADDRESS": "IP",
}

analyzer = build_analyzer()

def new_state():
    """Starea unei conversații: aceeași valoare primește același placeholder în toate mesajele."""
    return {"mapping": {}, "reverse": {}, "counters": {}}

def anonymize(text: str, state=None, score_threshold: float = 0.5):
    state = state or new_state()
    results = analyzer.analyze(text=text, language="ro", score_threshold=score_threshold)

    # Eliminăm suprapunerile: păstrăm rezultatul cu scorul cel mai mare
    results = sorted(results, key=lambda r: (-r.score, r.start))
    chosen = []
    for r in results:
        if all(r.end <= c.start or r.start >= c.end for c in chosen):
            chosen.append(r)
    chosen.sort(key=lambda r: r.start)

    mapping, reverse, counters = state["mapping"], state["reverse"], state["counters"]
    out, last = [], 0
    for r in chosen:
        value = text[r.start:r.end]
        label = ENTITY_LABELS.get(r.entity_type, r.entity_type)
        key = (label, value)
        if key not in reverse:
            counters[label] = counters.get(label, 0) + 1
            placeholder = f"<{label}_{counters[label]}>"
            reverse[key] = placeholder
            mapping[placeholder] = value
        out.append(text[last:r.start])
        out.append(reverse[key])
        last = r.end
    out.append(text[last:])
    return "".join(out), state

def deanonymize(text: str, mapping: dict) -> str:
    # Înlocuim întâi placeholder-ele lungi (<PERSOANA_10> înainte de <PERSOANA_1>)
    for placeholder in sorted(mapping, key=len, reverse=True):
        text = text.replace(placeholder, mapping[placeholder])
    return text

Test rapid:

from pii_guard import anonymize, deanonymize

text = ("Clientul Ion Popescu, CNP 1850312400127, telefon 0722 123 456, "
        "solicită rambursarea a 450 lei în contul RO49AAAA1B31007593840000. "
        "Email: ion.popescu@exemplu.ro")

anon, state = anonymize(text)
print(anon)
# Clientul <PERSOANA_1>, CNP <CNP_1>, telefon <TELEFON_1>, solicită rambursarea
# a 450 lei în contul <IBAN_1>. Email: <EMAIL_1>

Rezultatul exact depinde de model și de text. Testați pe exemple reale din activitatea voastră (anonimizate) și ajustați score_threshold.

Pasul 5 – Proxy compatibil OpenAI cu FastAPI

Ca aplicațiile existente să nu fie modificate, expunem un endpoint /v1/chat/completions identic cu cel OpenAI. Proxy-ul anonimizează mesajele, trimite cererea mai departe și de-anonimizează răspunsul.

app.py:

import os
import httpx
from fastapi import FastAPI, Request, HTTPException
import logging
from pii_guard import anonymize, deanonymize, new_state

log = logging.getLogger("pii_guard")
UPSTREAM = os.environ.get("UPSTREAM_URL", "https://api.openai.com/v1/chat/completions")
UPSTREAM_KEY = os.environ["UPSTREAM_API_KEY"]

SYSTEM_NOTE = ("Textul conține placeholder-e de forma <TIP_N> (ex. <PERSOANA_1>, <IBAN_1>). "
               "Păstrează-le exact așa în răspuns, fără să le modifici sau să le traduci.")

app = FastAPI(title="PII Guard")

@app.post("/v1/chat/completions")
async def chat(request: Request):
    body = await request.json()
    if body.get("stream"):
        raise HTTPException(400, "Streaming nu este suportat de PII Guard.")

    # O singură stare pentru toată conversația: placeholder-e consecvente între mesaje
    state = new_state()
    for msg in body.get("messages", []):
        if isinstance(msg.get("content"), str):
            msg["content"], state = anonymize(msg["content"], state)
    mapping = state["mapping"]

    body["messages"].insert(0, {"role": "system", "content": SYSTEM_NOTE})

    async with httpx.AsyncClient(timeout=120) as client:
        r = await client.post(UPSTREAM, json=body,
                              headers={"Authorization": f"Bearer {UPSTREAM_KEY}"})
    if r.status_code != 200:
        raise HTTPException(r.status_code, r.text)

    data = r.json()
    for choice in data.get("choices", []):
        content = choice.get("message", {}).get("content")
        if isinstance(content, str):
            choice["message"]["content"] = deanonymize(content, mapping)

    # Audit: logăm DOAR tipurile și numărul entităților, niciodată valorile
    tipuri = {}
    for ph in mapping:
        tip = ph.strip("<>").rsplit("_", 1)[0]
        tipuri[tip] = tipuri.get(tip, 0) + 1
    log.info("Cerere anonimizată: %s", tipuri)
    return data

Pornire:

export UPSTREAM_API_KEY="sk-..."
uvicorn app:app --host 0.0.0.0 --port 8080 --workers 2

Pasul 6 – Integrare

• Open WebUI: Admin Settings → Connections → OpenAI API → URL http://pii-guard:8080/v1.
• LiteLLM: setați api_base: http://pii-guard:8080/v1 pentru modelele cloud din config.yaml. Astfel toate cererile către furnizori externi trec automat prin anonimizare, iar modelele locale rămân directe.
• Scripturi și aplicații interne: schimbați doar base_url-ul clientului OpenAI.

Pasul 7 – Rulare în Docker

Dockerfile:

FROM python:3.12-slim
WORKDIR /app
RUN pip install --no-cache-dir presidio-analyzer presidio-anonymizer fastapi uvicorn httpx \
 && python -m spacy download ro_core_news_lg
COPY recognizers_ro.py engine.py pii_guard.py app.py ./
EXPOSE 8080
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8080", "--workers", "2"]

docker build -t pii-guard .
docker run -d --name pii-guard -p 8080:8080 -e UPSTREAM_API_KEY=sk-... --restart unless-stopped pii-guard

Limitări și recomandări

• Nicio detecție nu e 100% sigură. Numele rare, scrise cu literă mică sau fără diacritice pot scăpa modelului NER. Presidio reduce mult riscul, dar nu înlocuiește instruirea angajaților.
• Nu logați niciodată maparea placeholder ↔ valoare. Ea există doar în memorie, pe durata unei cereri.
• Pentru documente foarte sensibile (HR, medical, juridic), soluția corectă rămâne un LLM local, la care datele nu pleacă deloc din companie.
• Adăugați recunoaștere pentru date specifice companiei: numere de contract, coduri de client sau serii CI (PatternRecognizer cu regex și cuvinte de context).
• Construiți un set de test cu 50–100 de texte reale anonimizate manual și măsurați periodic câte entități scapă detecției, mai ales după actualizări de model.