core/engine.py

"""Facade del motore di dominio.

Unisce pipeline (rilevamento) + anonymizer (offuscamento) + builder dei
risultati, dietro un'interfaccia semplice usata dal livello web. I componenti
sono iniettati (Dependency Inversion): l'engine non istanzia modelli pesanti.

Distingue due fasi:
- `detect`: costosa (esegue i riconoscitori). Da fare una sola volta.
- `render`: economica (applica modalita'/gravita' su span gia' rilevati).
  Cambiare modalita' o toggle gravita' chiama solo `render`.
"""

from __future__ import annotations

from dataclasses import dataclass, replace

from config.catalog.entities import EntityCatalog
from core.contracts import LayerPriority, ScoringConfig, Severity, Span
from core.output.anonymizer import Anonymizer
from core.output.results import (
    Report,
    Segment,
    build_json_export,
    build_report,
    build_segments,
)
from core.output.trace import ClusterTrace
from core.pipeline import DetectionPipeline


@dataclass(slots=True)
class RenderResult:
    segments: list[Segment]
    anonymized_text: str
    report: Report
    json_export: dict


class AnonymizationEngine:
    def __init__(
        self,
        pipeline: DetectionPipeline,
        anonymizer: Anonymizer,
        catalog: EntityCatalog,
        config: ScoringConfig,
    ) -> None:
        self._pipeline = pipeline
        self._anonymizer = anonymizer
        self._catalog = catalog
        self._config = config

    @property
    def config(self) -> ScoringConfig:
        return self._config

    @property
    def has_verifier(self) -> bool:
        """True se è agganciato il verificatore LLM (giudice a valle)."""
        return self._pipeline.verifier is not None

    @property
    def verifier_error(self) -> str | None:
        """Messaggio d'errore del giudice dopo l'ULTIMA verifica, se `generate()`
        ha fallito (es. modello non caricato). None se assente o non fallito."""
        return getattr(self._pipeline.verifier, "error", None)

    def detect(self, text: str, min_confidence: float | None = None,
               progress=None, timings=None) -> list[Span]:
        """Fase costosa: ritorna gli span rilevati (da mettere in cache).

        `progress(frazione, messaggio)` opzionale per l'avanzamento in UI.
        `timings` (dict) opzionale: riempito coi secondi per processo/giudice.
        """
        return self._pipeline.detect(
            text, min_confidence, progress=progress, timings=timings
        )

    def detect_raw(self, text: str) -> list[Span]:
        """Span grezzi per-riconoscitore (no boost, no fusione, no soglia).

        Espone i rilevamenti di ogni processo cosi' come sono, per confronto
        tra metodologie. Vedi `DetectionPipeline.detect_raw`.
        """
        return self._pipeline.detect_raw(text)

    def detect_with_trace(
        self, text: str, min_confidence: float | None = None,
        progress=None, timings=None,
    ) -> tuple[list[Span], list[ClusterTrace]]:
        """Come `detect`, ma restituisce anche la traccia di debug per cluster.

        Arricchisce ogni traccia con etichetta e gravita' del catalogo (che la
        pipeline neutra non conosce). `progress(frazione, messaggio)` opzionale
        per l'avanzamento in UI. `timings` (dict) opzionale: riempito coi
        secondi per processo/giudice.
        """
        spans, traces = self._pipeline.detect_with_trace(
            text, min_confidence, progress=progress, timings=timings
        )
        enriched = [
            replace(
                t,
                label=self._catalog.get(t.entity_type).label,
                severity=self._catalog.severity(t.entity_type).value,
            )
            for t in traces
        ]
        return spans, enriched

    def build_manual_spans(
        self, text: str, ranges: list[tuple[int, int]]
    ) -> list[Span]:
        """Crea span manuali (correzione: l'utente forza l'offuscamento).

        Score 1.0 e layer RULES cosi' vincono sempre nelle sovrapposizioni.
        """
        manual: list[Span] = []
        for start, end in ranges:
            start = max(0, min(start, len(text)))
            end = max(start, min(end, len(text)))
            if end <= start:
                continue
            manual.append(
                Span(
                    start=start,
                    end=end,
                    text=text[start:end],
                    entity_type="DATO_SENSIBILE",
                    score=1.0,
                    layer=LayerPriority.RULES,
                    validated=True,
                    source="manuale",
                )
            )
        return manual

    def merge_manual(self, auto: list[Span], manual: list[Span]) -> list[Span]:
        """Unisce span automatici e manuali, rimuovendo gli automatici coperti
        da una correzione manuale (il manuale ha priorita').
        """
        result = list(manual)
        for sp in auto:
            covered = any(
                m.start < sp.end and sp.start < m.end for m in manual
            )
            if not covered:
                result.append(sp)
        result.sort(key=lambda s: s.start)
        return result

    def exclude_spans(
        self, spans: list[Span], excluded_ranges: list[tuple[int, int]]
    ) -> list[Span]:
        """Rimuove gli span coperti da un range escluso (ripristino manuale:
        l'utente chiede di NON offuscare un'entita' gia' rilevata)."""
        if not excluded_ranges:
            return spans
        return [
            sp for sp in spans
            if not any(r0 < sp.end and sp.start < r1 for r0, r1 in excluded_ranges)
        ]

    def render(
        self,
        text: str,
        spans: list[Span],
        mode: str = "placeholder",
        enabled_severities: set[Severity] | None = None,
    ) -> RenderResult:
        """Fase economica: applica modalita'/gravita' su span gia' rilevati."""
        active = (
            spans
            if enabled_severities is None
            else [
                s for s in spans
                if self._catalog.severity(s.entity_type) in enabled_severities
            ]
        )
        return RenderResult(
            segments=build_segments(text, active, self._catalog),
            anonymized_text=self._anonymizer.apply(
                text, spans, mode, enabled_severities
            ),
            report=build_report(active, self._catalog),
            json_export=build_json_export(active, self._catalog),
        )