web/services/analysis.py

"""Servizio applicativo: coordina engine + store per detect e render.

Mantiene i router sottili. Non implementa logica di dominio (delega all'engine):
si occupa di id documento, cache e costruzione DTO.
"""

from __future__ import annotations

import logging
from collections import defaultdict
from collections.abc import Sequence

from config import CATALOG
from core.output.trace import ClusterTrace
from core.support.logging_utils import get_logger
from web.dto import RenderResponse
from web.services.debug_payload import (
    _is_term,
    _llm_status,
    _parse_severities,
    build_debug_payload,
    build_llm_check,
)
from web.services.documents import StoredDocument, compute_doc_id
from web.services.registry import ServiceRegistry

logger = get_logger("analysis")


def build_methods_comparison(
    traces: Sequence[ClusterTrace],
    process_index: dict[str, tuple[str, str]],
    selected: set[str] | None,
) -> dict:
    """Confronto per-layer DERIVATO dalle tracce della singola run di analisi.

    Regola primaria "debug == analizza": il tab metodi NON ri-esegue i layer.
    Ogni contributo grezzo della traccia è già taggato col processo che l'ha
    emesso; `process_index` lo mappa al layer. Stessa forma dell'output di prima
    (`methods`, `method_labels`, `entities`, `summary`), ma senza ri-esecuzione.

    `selected`: chiavi dei layer da mostrare (None = tutti, escluso sempre l'LLM).
    Un layer selezionato che non ha catturato nulla compare comunque come colonna
    (found_by di nessuno), come nel confronto storico.
    """
    from config.detection.layers import DETECTION_LAYERS

    # Colonne = layer selezionati (escluso LLM), in ordine di priorità di fusione.
    method_labels = {
        ly.key: ly.label
        for ly in DETECTION_LAYERS
        if ly.key != "llm" and (selected is None or ly.key in selected)
    }
    active_keys = list(method_labels)

    # norm -> {layer_key -> [contributi]} dai soli contributi dei layer attivi.
    entity_map: dict[str, dict[str, list]] = defaultdict(lambda: defaultdict(list))
    for tr in traces:
        for c in tr.contributions:
            kl = process_index.get(c.process)
            if kl is None:
                continue
            key = kl[0]
            if key not in method_labels:
                continue
            norm = c.text.strip().lower()
            if not _is_term(norm):
                continue
            entity_map[norm][key].append(c)

    entries = []
    for _norm, by_method in entity_map.items():
        found_by = [k for k in active_keys if k in by_method]
        not_found_by = [k for k in active_keys if k not in by_method]
        rep = by_method[found_by[0]][0]
        item = CATALOG.get(rep.entity_type)
        by_method_info = {}
        for mk in found_by:
            mk_c = by_method[mk]
            best = max(mk_c, key=lambda s: s.score)
            mk_item = CATALOG.get(best.entity_type)
            processes = sorted({(s.process or s.source or "?") for s in mk_c})
            by_method_info[mk] = {
                "score": round(best.score, 4),
                "entity_type": best.entity_type,
                "label": mk_item.label,
                "count": len(mk_c),
                "processes": processes,
            }
        entries.append({
            "text": rep.text,
            "severity": item.severity.value,
            "found_by": found_by,
            "not_found_by": not_found_by,
            "by_method": by_method_info,
        })

    entries.sort(key=lambda e: (len(e["found_by"]), e["text"].lower()))
    found_by_all = sum(1 for e in entries if not e["not_found_by"])
    return {
        "methods": active_keys,
        "method_labels": method_labels,
        "entities": entries,
        "raw": True,
        "summary": {
            "total_unique": len(entity_map),
            "found_by_all": found_by_all,
            "partial": len(entries) - found_by_all,
        },
    }


class AnalysisService:
    def __init__(self, registry: ServiceRegistry) -> None:
        self._registry = registry

    def analyze_text(
        self,
        text: str,
        min_confidence: float | None = None,
        method: str | None = None,
        layers: list[str] | None = None,
        verify: bool = False,
        progress=None,
        approver: str | None = None,
        simulate: bool = False,
    ) -> tuple[str, RenderResponse]:
        """Rileva (con cache) e renderizza con i parametri di default.

        Se `layers` e' valorizzato, l'engine e' composto dai soli layer scelti
        (checkbox UI) e ha priorita' sul `method`. La cache e' indicizzata anche
        per selezione e per la verifica LLM: cambiare layer/metodologia/verifica
        produce un documento distinto. `progress(frazione, messaggio)` opzionale
        per l'avanzamento in UI.
        """
        if progress is not None:
            progress(0.02, "Preparazione: carico i modelli di rilevamento…")
        engine, salt, resolved = self._engine_and_salt(
            method, layers, verify, approver, min_confidence, simulate
        )
        doc_id = compute_doc_id(text.encode("utf-8"), salt=salt)
        stored = self._registry.documents.get(doc_id)
        cached = stored is not None
        if stored is None:
            stored = self._detect_and_store(
                text, doc_id, engine, resolved, min_confidence, verify, progress,
                simulate=simulate,
            )
        if logger.isEnabledFor(logging.DEBUG):
            logger.debug(
                "analyze_text: %d caratteri | selezione=%s | soglia=%s | "
                "cache=%s | doc=%s | %d span",
                len(text), layers or method or "default",
                min_confidence if min_confidence is not None else "default",
                "HIT" if cached else "MISS", doc_id[:8], len(stored.spans),
            )
        result = engine.render(stored.text, stored.spans, "placeholder")
        return doc_id, RenderResponse.build(doc_id, result)

    def _resolve_key(self, method: str | None, layers: list[str] | None) -> str:
        """Chiave risolta dei setting (layer hanno priorità), SENZA engine.

        Unica fonte della chiave: la usano sia `_select_engine` (per la coppia
        engine+chiave) sia `request_key` (per il salt PDF), così cache/salt e
        selezione engine non possono divergere."""
        if layers is not None:
            norm = sorted(set(layers)) or ["rules"]
            return "layers:" + ",".join(norm)
        return self._registry.default_method if not method else method

    def request_key(self, method: str | None, layers: list[str] | None) -> str:
        """Chiave risolta (per cache/salt) SENZA costruire l'engine.

        Usata dal flusso PDF per comporre il salt del doc_id prima del worker."""
        return self._resolve_key(method, layers)

    def engine_and_key_for_request(
        self, method: str | None, layers: list[str] | None, verify: bool = False,
        approver: str | None = None, simulate: bool = False,
    ):
        """Engine + chiave risolta per i setting di una richiesta (per il PDF)."""
        return self._select_engine(method, layers, verify, approver, simulate)

    def store_pdf_detection(
        self, doc_id: str, text: str, *, method: str | None, layers: list[str] | None,
        verify: bool, approver: str | None, min_confidence: float | None,
        pages, is_scanned: bool, raw_bytes: bytes, simulate: bool = False,
    ) -> StoredDocument:
        """Detection PDF (una passata) salvata col PDF. Riusa `_detect_and_store`
        così PDF e testo producono span+tracce nello stesso identico modo."""
        engine, resolved = self.engine_and_key_for_request(
            method, layers, verify, approver, simulate
        )
        return self._detect_and_store(
            text, doc_id, engine, resolved, min_confidence, verify, progress=None,
            pages=pages, is_pdf=True, is_scanned=is_scanned, raw_bytes=raw_bytes,
            simulate=simulate,
        )

    def _select_engine(
        self, method: str | None, layers: list[str] | None, verify: bool,
        approver: str | None = None, simulate: bool = False,
    ):
        """Sceglie l'engine (per layer o per metodologia) e la chiave risolta.

        I layer hanno priorita': se presenti, l'engine e' composto da essi.
        Unica fonte per la coppia (engine, chiave) usata da cache e debug (DRY).
        Il giudice `approver` conta solo con `verify` (engine di verifica).
        """
        if layers is not None:
            engine = self._registry.engine_for_layers(
                layers, verify=verify, approver=approver, simulate=simulate
            )
            return engine, self._resolve_key(method, layers)
        engine = self._registry.engine_for(
            method, verify=verify, approver=approver, simulate=simulate
        )
        return engine, self._resolve_key(method, layers)

    def _engine_and_salt(
        self, method: str | None, layers: list[str] | None, verify: bool = False,
        approver: str | None = None, min_confidence: float | None = None,
        simulate: bool = False,
    ):
        """Sceglie engine, salt di cache e chiave risolta.

        `verify`, il giudice (`approver`), la simulazione e `min_confidence`
        entrano nel salt: risultati con verifica/giudice/simulazione/soglia
        diversi sono documenti distinti.
        """
        engine, resolved_key = self._select_engine(
            method, layers, verify, approver, simulate
        )
        vsalt = ":verify" if verify else ""
        asalt = f":approver={approver}" if (verify and approver) else ""
        ssalt = ":sim" if (verify and simulate) else ""
        msalt = f":mc={min_confidence}" if min_confidence is not None else ""
        return engine, f"text:{resolved_key}{vsalt}{asalt}{ssalt}{msalt}", resolved_key

    def _detect_and_store(
        self, text: str, doc_id: str, engine, resolved: str,
        min_confidence: float | None, verify: bool, progress,
        *, pages=(), is_pdf=False, is_scanned=False, raw_bytes=None,
        simulate: bool = False,
    ) -> StoredDocument:
        """Esegue UNA detection tracciata (layer -> giudice -> filtro) e salva nel
        documento span + tracce + metadati. Unica sorgente per analyze/debug/PDF.
        """
        # Mappa processo→layer catturata PRIMA della detection: con la verifica
        # LLM le famiglie ML vengono scaricate dal provider durante la run.
        process_index = self._registry.process_layer_index()
        sink: dict[str, float] = {}
        spans, traces = engine.detect_with_trace(
            text, min_confidence, progress=progress, timings=sink
        )
        threshold = (
            min_confidence if min_confidence is not None
            else engine.config.min_confidence
        )
        stored = StoredDocument(
            doc_id=doc_id, text=text, spans=spans, traces=tuple(traces),
            threshold=threshold, method=resolved,
            llm_status=_llm_status(verify, engine.has_verifier, simulate),
            llm_error=(engine.verifier_error if verify else None),
            timings=tuple(_aggregate_timings(sink, process_index)),
            pages=pages, is_pdf=is_pdf, is_scanned=is_scanned, raw_bytes=raw_bytes,
        )
        self._registry.documents.save(stored)
        return stored

    def timings_for(self, doc_id: str) -> list[dict] | None:
        """Tempi per layer (+ giudice) della detection del documento, se in cache."""
        stored = self._registry.documents.get(doc_id)
        if stored is None:
            return None
        return list(stored.timings)

    def render(
        self,
        doc_id: str,
        mode: str,
        enabled_severities: list[str] | None,
        manual_ranges: list[tuple[int, int]] | None = None,
        excluded_ranges: list[tuple[int, int]] | None = None,
    ) -> RenderResponse | None:
        """Re-render economico: non riesegue il rilevamento."""
        stored = self._registry.documents.get(doc_id)
        if stored is None:
            return None
        engine = self._registry.engine
        spans = stored.spans
        if manual_ranges:
            manual = engine.build_manual_spans(stored.text, manual_ranges)
            spans = engine.merge_manual(stored.spans, manual)
        if excluded_ranges:
            spans = engine.exclude_spans(spans, excluded_ranges)
        result = engine.render(
            stored.text, spans, mode, _parse_severities(enabled_severities)
        )
        return RenderResponse.build(doc_id, result)

    def get_stored_text(self, doc_id: str) -> str | None:
        """Recupera il testo grezzo di un documento già in cache."""
        stored = self._registry.documents.get(doc_id)
        return stored.text if stored else None

    def llm_check_for(self, doc_id: str) -> dict | None:
        """Stato del giudice per un documento in cache (per l'avviso FE).

        None se il documento non è in cache. Letto dalle tracce salvate, quindi
        coerente con anteprima/report/debug (stessa unica passata)."""
        stored = self._registry.documents.get(doc_id)
        if stored is None:
            return None
        return build_llm_check(stored.traces, stored.llm_status, stored.llm_error)

    def compare_detection_methods(
        self,
        text: str,
        min_confidence: float | None = None,
        layers: list[str] | None = None,
        method: str | None = None,
        progress=None,
    ) -> dict:
        """Confronto per-layer ("metodi") DERIVATO dalla singola run di analisi.

        Regola primaria "debug == analizza": NON ri-esegue i layer. Riusa (o crea,
        se il documento non è in cache) la STESSA run tracciata di `debug_trace` /
        `analyze`, poi ricostruisce "chi ha catturato cosa" dai contributi grezzi
        già presenti nelle tracce (ogni span porta il processo emittente). Così il
        tab metodi non può mostrare layer che l'analisi non ha eseguito.

        L'LLM è sempre escluso dalle colonne (gira isolato). Restituisce TUTTE le
        entità (non solo le discrepanze).
        """
        from config.detection.layers import resolve_selected_layers

        # Stessa risoluzione/engine/cache dell'analisi: salt layer-specifico, così
        # il documento già analizzato viene RIUSATO senza una seconda passata.
        engine, salt, resolved = self._engine_and_salt(
            method, layers, False, None, min_confidence, False
        )
        doc_id = compute_doc_id(text.encode("utf-8"), salt=salt)
        stored = self._registry.documents.get(doc_id)
        if stored is None:
            stored = self._detect_and_store(
                text, doc_id, engine, resolved, min_confidence, False, progress,
            )

        resolved_layers = resolve_selected_layers(layers)
        selected = set(resolved_layers) if resolved_layers is not None else None
        process_index = self._registry.process_layer_index()
        return build_methods_comparison(stored.traces, process_index, selected)

    def debug_trace(
        self,
        text: str,
        min_confidence: float | None = None,
        method: str | None = None,
        layers: list[str] | None = None,
        verify: bool = False,
        progress=None,
        approver: str | None = None,
        simulate: bool = False,
    ) -> dict:
        """Esegue il rilevamento tracciato e restituisce, per ogni frase
        catturata, tutti i contributi (processo, layer, score), il
        rappresentante scelto, la scomposizione dei boost e l'esito.

        Riusa la cache condivisa con analyze_text (stesso salt): se il documento
        è già stato analizzato, il giudice NON viene rieseguito.
        `progress(frazione, messaggio)` opzionale per l'avanzamento in UI.
        """
        if progress is not None:
            # La costruzione dell'engine può caricare i modelli ML (prima analisi):
            # segnaliamolo subito così la UI non resta muta durante il load.
            progress(0.02, "Preparazione: carico i modelli di rilevamento…")
        engine, salt, resolved = self._engine_and_salt(
            method, layers, verify, approver, min_confidence, simulate
        )
        doc_id = compute_doc_id(text.encode("utf-8"), salt=salt)
        stored = self._registry.documents.get(doc_id)
        # Ricalcola SOLO se il documento non è mai stato analizzato. Un documento
        # analizzato senza rilevamenti ha `traces == ()` legittimamente: NON va
        # rieseguito (sarebbe una passata inutile). `_detect_and_store` è l'unico
        # costruttore di StoredDocument, quindi `stored is None` ⇔ "non analizzato".
        if stored is None:
            stored = self._detect_and_store(
                text, doc_id, engine, resolved, min_confidence, verify, progress,
                simulate=simulate,
            )
        payload = build_debug_payload(
            stored.traces, stored.threshold, stored.method, stored.llm_status
        )
        payload["timings"] = list(stored.timings)
        return payload

    def debug_for_doc(self, doc_id: str) -> dict | None:
        """Debug di un documento già analizzato, letto dalle tracce in cache.

        Nessun rilancio della detection: garantisce che il debug coincida con
        ciò che è stato oscurato (anteprima/report). None solo se il documento
        non è in cache (mai analizzato); un documento senza rilevamenti restituisce
        un payload vuoto ma valido."""
        stored = self._registry.documents.get(doc_id)
        if stored is None:
            return None
        payload = build_debug_payload(
            stored.traces, stored.threshold, stored.method, stored.llm_status
        )
        payload["timings"] = list(stored.timings)
        return payload

    def detection_layers(self) -> dict:
        """Espone i layer attivabili (checkbox UI), in ordine di elaborazione."""
        return self._registry.layer_catalog()

    def approvers(self) -> dict:
        """Espone i giudici LLM selezionabili (per il picker della UI).

        Disponibili solo se ENABLE_ML è attivo (come il verificatore): altrimenti
        la lista è vuota e la UI non mostra il picker.
        """
        from layers.llm.llm_approver import registry as approver_registry

        if not self._registry.settings.enable_ml:
            return {"default": "", "approvers": []}
        return {
            "default": approver_registry.default_key(),
            "approvers": [
                {
                    "key": s.key,
                    "label": s.label,
                    "description": s.description,
                    "model": s.repo,
                    "vram_hint": s.vram_hint,
                    "default": s.is_default,
                }
                for s in approver_registry.specs()
            ],
        }

    def detection_methods(self) -> dict:
        """Espone le metodologie abilitate + la predefinita per la UI."""
        registry = self._registry.detection_registry
        return {
            "default": self._registry.default_method,
            "methods": [
                {
                    "key": i.key,
                    "label": i.label,
                    "technique": i.technique.value,
                    "description": i.description,
                    "requires_ml": i.requires_ml,
                    "models": list(i.models),
                    "default": i.default,
                }
                for i in registry.infos()
            ],
        }


def _aggregate_timings(
    sink: dict[str, float], process_index: dict[str, tuple[str, str]]
) -> list[dict]:
    """Aggrega i secondi per-processo della pipeline in voci per LAYER.

    Ritorna voci ``{key, label, seconds}`` nell'ordine di `DETECTION_LAYERS`,
    più la voce del giudice LLM in coda (chiave speciale ``__giudice__`` del
    sink). Un processo senza layer noto (mappa incompleta) compare col proprio
    nome, così nessun tempo misurato va perso.
    """
    from config.detection.layers import DETECTION_LAYERS

    per_layer: dict[str, float] = defaultdict(float)
    orphans: dict[str, float] = defaultdict(float)
    for name, seconds in sink.items():
        if name == "__giudice__":
            continue
        if name in process_index:
            per_layer[process_index[name][0]] += seconds
        else:
            orphans[name] += seconds
    out = [
        {"key": spec.key, "label": spec.label, "seconds": round(per_layer[spec.key], 3)}
        for spec in DETECTION_LAYERS
        if spec.key in per_layer
    ]
    out.extend(
        {"key": name, "label": name, "seconds": round(seconds, 3)}
        for name, seconds in sorted(orphans.items())
    )
    if "__giudice__" in sink:
        out.append({
            "key": "giudice", "label": "Giudice LLM",
            "seconds": round(sink["__giudice__"], 3),
        })
    return out