Etendre PIIGhost¶

PIIGhost est concu autour de protocoles (typage structurel Python). Chaque etape du pipeline est un point d'injection ou vous pouvez brancher votre propre implementation sans toucher au reste du code.

flowchart LR
    A[AnonymizationPipeline] -->|inject| B[AnyDetector]
    A -->|inject| C[AnySpanConflictResolver]
    A -->|inject| D[AnyEntityLinker]
    A -->|inject| E[AnyEntityConflictResolver]
    A -->|inject| F[AnyAnonymizer]
    F -->|inject| G[AnyPlaceholderFactory]

Aucune classe de base a heriter. Il suffit d'implementer la methode requise Python verifie la compatibilite au moment de l'appel.

Creer un AnyDetector personnalise¶

Quand l'utiliser : remplacer GLiNER2 par spaCy, un appel API distant, une liste blanche, etc.

Protocole¶

class AnyDetector(Protocol):
    async def detect(self, text: str) -> list[Detection]: ...

import spacy
from piighost.models import Detection, Span

class SpacyDetector:
    """Detecteur NER base sur spaCy."""

    def __init__(self, model_name: str = "fr_core_news_sm"):
        self._nlp = spacy.load(model_name)

    async def detect(self, text: str) -> list[Detection]:
        doc = self._nlp(text)
        return [
            Detection(
                text=ent.text,
                label=ent.label_,
                position=Span(start_pos=ent.start_char, end_pos=ent.end_char),
                confidence=1.0,
            )
            for ent in doc.ents
        ]

import re
from piighost.models import Detection, Span

class AllowlistDetector:
    """Detecte les entites d'une liste fixe (utile pour les tests ou les donnees structurees)."""

    def __init__(self, allowlist: dict[str, str]):
        # {"Patrick Dupont": "PERSON", "Paris": "LOCATION"}
        self._allowlist = allowlist

    async def detect(self, text: str) -> list[Detection]:
        detections = []
        for fragment, label in self._allowlist.items():
            for match in re.finditer(re.escape(fragment), text):
                detections.append(Detection(
                    text=match.group(),
                    label=label,
                    position=Span(start_pos=match.start(), end_pos=match.end()),
                    confidence=1.0,
                ))
        return detections

Utilisation¶

from piighost.pipeline import AnonymizationPipeline

pipeline = AnonymizationPipeline(
    detector=SpacyDetector("fr_core_news_sm"),
    ...,
)

Packs regex prêts à l'emploi¶

Pour les PII structurées dont la syntaxe est standardisée (e-mails, IBAN, téléphones, SSN), PIIGhost fournit des dictionnaires regex organisés par zone géographique. Vous piochez uniquement ceux dont vous avez besoin, et vous les fusionnez librement.

from piighost.detector import RegexDetector
from piighost.detector.patterns import FR_PATTERNS, GENERIC_PATTERNS

detector = RegexDetector(patterns={**GENERIC_PATTERNS, **FR_PATTERNS})

Les packs sont volontairement permissifs sur la syntaxe : le motif CREDIT_CARD accepte n'importe quelle séquence de 13 à 19 chiffres, IBAN accepte un préfixe de 2 lettres suivi de 11 à 30 alphanumériques, FR_NIR accepte la forme complète du NIR sans contrôler la clé. Sans validateur, ces motifs produisent beaucoup de faux positifs (toute longue séquence numérique ressemble à un numéro de carte).

Validateurs de checksum¶

PIIGhost fournit des validateurs dans piighost.validators qui se branchent directement sur RegexDetector pour filtrer les matches syntaxiquement corrects mais invalides selon un contrôle métier :

from piighost.detector import RegexDetector
from piighost.detector.patterns import FR_PATTERNS, GENERIC_PATTERNS
from piighost.validators import validate_iban, validate_luhn, validate_nir

detector = RegexDetector(
    patterns={**GENERIC_PATTERNS, **FR_PATTERNS},
    validators={
        "CREDIT_CARD": validate_luhn,
        "FR_IBAN": validate_iban,
        "FR_NIR": validate_nir,
    },
)

Un label sans entrée dans validators est accepté sur la seule base du match regex. Les matches rejetés par un validateur sont silencieusement écartés (aucun log, aucune exception) ; chaînez un autre détecteur si vous voulez enregistrer le rejet.

Mapping de labels NER¶

Les détecteurs NER fournis (SpacyDetector, Gliner2Detector, TransformersDetector) héritent tous de BaseNERDetector, qui supporte le mapping de labels : découpler le label produit en interne par le modèle du label qui apparaît dans Detection.label (et donc dans les placeholders, datasets, etc.).

Passez un dict {externe: interne} au lieu d'une liste pour activer le mapping :

from piighost.detector.spacy import SpacyDetector

# Sans mapping (identité) : Detection.label sera "PER" / "LOC"
detector = SpacyDetector(model=nlp, labels=["PER", "LOC"])

# Avec mapping : Detection.label sera "PERSON" / "LOCATION"
detector = SpacyDetector(
    model=nlp,
    labels={"PERSON": "PER", "LOCATION": "LOC"},
)

Pour GLiNER2, c'est particulièrement utile car certaines chaînes de requête fonctionnent mieux que d'autres :

from piighost.detector.gliner2 import Gliner2Detector

# Interroger GLiNER2 avec "person" et "company" (meilleure détection)
# mais produire des labels propres "PERSON" / "COMPANY" dans les Detection.
detector = Gliner2Detector(
    model=model,
    labels={"PERSON": "person", "COMPANY": "company"},
)

Cela permet de changer le modèle sous-jacent sans modifier le code en aval (placeholder factories, entity resolvers, assertions de tests). C'est aussi le prérequis pour construire des datasets NER stables à partir des saisies utilisateur.

Vous pouvez inspecter les labels résultants avec detector.external_labels et detector.internal_labels.

Creer un AnySpanConflictResolver personnalise¶

Quand l'utiliser : strategie differente pour gerer les detections qui se chevauchent (ex: preferer les spans les plus longs).

Protocole¶

class AnySpanConflictResolver(Protocol):
    def resolve(self, detections: list[Detection]) -> list[Detection]: ...

Désactiver le composant¶

Passer DisabledSpanConflictResolver() pour conserver toutes les détections telles quelles. Utile quand le détecteur garantit déjà des spans non chevauchants, ou quand on veut que les chevauchements remontent jusqu'au linker.

from piighost import DisabledSpanConflictResolver

pipeline = AnonymizationPipeline(
    detector=detector,
    span_resolver=DisabledSpanConflictResolver(),  # ← passe-plat
    entity_linker=...,
    entity_resolver=...,
    anonymizer=...,
)

Creer un AnyEntityLinker personnalise¶

Quand l'utiliser : logique differente pour grouper les detections en entites (ex: correspondance floue, variantes phonetiques).

Protocole¶

class AnyEntityLinker(Protocol):
    def link(self, text: str, detections: list[Detection]) -> list[Entity]: ...

Désactiver le composant¶

Passer DisabledEntityLinker() pour mapper chaque détection 1:1 à une Entity. Pas d'expansion (pas de recherche d'occurrences manquées), pas de regroupement, pas de liaison entre messages. Utile quand le détecteur produit déjà des détections propres et déduplicquées.

from piighost import DisabledEntityLinker

pipeline = AnonymizationPipeline(
    detector=detector,
    span_resolver=...,
    entity_linker=DisabledEntityLinker(),  # ← passe-plat
    entity_resolver=...,
    anonymizer=...,
)

Creer un AnyEntityConflictResolver personnalise¶

Quand l'utiliser : strategie differente pour fusionner les entites qui referent au meme PII.

Protocole¶

class AnyEntityConflictResolver(Protocol):
    def resolve(self, entities: list[Entity]) -> list[Entity]: ...

Implementations fournies :

• MergeEntityConflictResolver algorithme union-find fusionnant les entites avec des detections communes
• FuzzyEntityConflictResolver fusionne les entites avec un texte canonique similaire via similarite Jaro-Winkler
• DisabledEntityConflictResolver passe-plat qui retourne les entités telles quelles (pour désactiver entièrement la fusion)

Creer un AnyPlaceholderFactory personnalise¶

Quand l'utiliser : tags UUID pour l'anonymat total, format personnalise, integration avec un systeme de tokens externe.

Protocole¶

class AnyPlaceholderFactory(Protocol[PreservationT_co]):
    def create(self, entities: list[Entity]) -> dict[Entity, str]: ...

Chaque factory porte un tag de préservation fantôme (PreservesIdentity, PreservesLabel, PreservesShape, PreservesNothing) que le type-checker utilise pour verrouiller les consommateurs comme PIIAnonymizationMiddleware. Voir Placeholder factories pour la taxonomie complète, les exemples détaillés (UUIDPlaceholderFactory, BracketPlaceholderFactory) et le raisonnement derrière la contrainte.

Utilisation¶

from piighost.anonymizer import Anonymizer

anonymizer = Anonymizer(ph_factory=UUIDPlaceholderFactory())

Composition complete¶

Tous les composants sont independants et peuvent etre combines librement :

from piighost.anonymizer import Anonymizer
from piighost.pipeline import ThreadAnonymizationPipeline
from piighost.linker.entity import ExactEntityLinker
from piighost.entity_resolver import FuzzyEntityConflictResolver
from piighost.middleware import PIIAnonymizationMiddleware
from piighost.span_resolver import ConfidenceSpanConflictResolver

detector = SpacyDetector("fr_core_news_sm")              # Votre detecteur
span_resolver = ConfidenceSpanConflictResolver()         # Ou votre resolver
entity_linker = ExactEntityLinker()                      # Ou votre linker
entity_resolver = FuzzyEntityConflictResolver()          # Fusion floue
anonymizer = Anonymizer(UUIDPlaceholderFactory())        # Tags UUID opaques

pipeline = ThreadAnonymizationPipeline(
    detector=detector,
    span_resolver=span_resolver,
    entity_linker=entity_linker,
    entity_resolver=entity_resolver,
    anonymizer=anonymizer,
)

middleware = PIIAnonymizationMiddleware(pipeline=pipeline)

Pour tester unitairement vos composants personnalisés avec ExactMatchDetector et pytest, voir le guide Tests.