"""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