Upload 10 files

app.py

@@ -275,4 +275,4 @@ with gr.Blocks(
     pdf_btn.click(handle_pdf,  inputs=[pdf_in, *inputs_common], outputs=outputs)
     demo_btn.click(lambda: DEMO_TEXT, inputs=None, outputs=txt_in)
 
-demo.launch()
+demo.launch()

config.py

@@ -8,7 +8,7 @@ MODES: dict[str, str] = {
     "Solo prima lettera  →  J***":        "first",
 }
 
-DEFAULT_MIN_SCORE: float = 0.85
+DEFAULT_MIN_SCORE: float = 0.65
 
 # ---------------------------------------------------------------------------
 # Entity type → Italian placeholder label

gliner_recognizer.py

@@ -1,366 +1,117 @@
-"""Pipeline a 3 livelli: Regex (opz.) → NER → GLiNER, chunking, agreement boost."""
-import hashlib
+"""GLiNER recognizer Presidio-compatible — Livello 3 della pipeline."""
 import logging
-import re
-from typing import Optional
+from typing import List, Optional
 
-from presidio_analyzer import RecognizerResult
-
-from recognizers import analyzer_full, analyzer_ner_only, POST_BOOST_PATTERNS
-from gliner_recognizer import GlinerRecognizer
-from span_resolver import resolve_overlapping_spans
-from config import MODES, LABEL_IT
+from presidio_analyzer import EntityRecognizer, RecognizerResult
 
 logger = logging.getLogger(__name__)
 
-# Configurazione
-_POST_BOOST_DELTA: float = 0.30
-_POST_BOOST_MAX:   float = 1.0
-_AGREEMENT_DELTA:  float = 0.15
-_AGREEMENT_MAX:    float = 1.0
-_CHUNK_SIZE:       int   = 1500   # ~400 token BERT
-_CHUNK_OVERLAP:    int   = 200
-
-_REGEX_RECOGNIZER_NAMES: set[str] = {
-    "cig", "cup", "rea", "num_gara", "pec", "inps",
-    "polizza_full", "polizza_alt",
-    "cpv", "nuts", "ateco", "ribasso", "protocollo",
-    "atto", "anac", "lotto", "siogg",
-    "cf_azienda", "societa", "albo",
-}
-
-_PROCUREMENT_PREFIX: dict[str, str] = {
-    "RUP":                 "RUP",
-    "STAZIONE_APPALTANTE": "SA",
-    "OPERATORE_ECONOMICO": "OE",
-    "SUBAPPALTATORE":      "SUB",
-    "COMMISSARIO_GARA":    "COMM",
-    "DIRETTORE_LAVORI":    "DL",
-    "CODICE_CPV":          "CPV",
-    "CODICE_NUTS":         "NUTS",
-    "CODICE_ATECO":        "ATECO",
-    "NUMERO_GARA":         "NUMGARA",
-    "IMPORTO_BASE_ASTA":   "IMPORTO_BASE",
-    "PROCUREMENT_ENTITY":  "PROCUREMENT",
-}
-
-_RECLASSIFY_PATTERNS: dict[str, re.Pattern] = {
-    "CIG": re.compile(r"^(?:\d{7}[0-9A-F]{3}|[A-Z][0-9A-F]{9})$"),
-    "CUP": re.compile(r"^[A-Z]\d{2}[A-Z][A-Z0-9]{2}\d{6}[A-Z0-9]{3}$"),
-    "REA": re.compile(r"^[A-Z]{2}[\s\-/\.]\d{4,7}$"),
-}
-
-_GENERIC_NER_ENTITIES: set[str] = {
-    "NUMERO_DOCUMENTO", "N_LICENZA", "N_SENTENZA", "NUMERO_CONTO",
+PROCUREMENT_LABELS: list[str] = [
+    "CIG",
+    "CUP",
+    "RUP",
+    "stazione appaltante",
+    "responsabile del procedimento",
+    "operatore economico",
+    "subappaltatore",
+    "importo a base d'asta",
+    "codice CPV",
+    "codice NUTS",
+    "codice ATECO",
+    "numero gara",
+    "commissario di gara",
+    "direttore dei lavori",
+]
+
+_LABEL_TO_ENTITY: dict[str, str] = {
+    "CIG":                           "CIG",
+    "CUP":                           "CUP",
+    "RUP":                           "RUP",
+    "stazione appaltante":           "STAZIONE_APPALTANTE",
+    "responsabile del procedimento": "RUP",
+    "operatore economico":           "OPERATORE_ECONOMICO",
+    "subappaltatore":                "SUBAPPALTATORE",
+    "importo a base d'asta":         "IMPORTO_BASE_ASTA",
+    "codice CPV":                    "CODICE_CPV",
+    "codice NUTS":                   "CODICE_NUTS",
+    "codice ATECO":                  "CODICE_ATECO",
+    "numero gara":                   "NUMERO_GARA",
+    "commissario di gara":           "COMMISSARIO_GARA",
+    "direttore dei lavori":          "DIRETTORE_LAVORI",
 }
 
-_CURRENCY_PREFIX = re.compile(r"^(€|EUR|\$|£)\s*", flags=re.IGNORECASE)
-
-_gliner: Optional[GlinerRecognizer] = None
-
-
-# ---------------------------------------------------------------------------
-# Helpers – singleton, utils
-# ---------------------------------------------------------------------------
-def _get_gliner(threshold: float = 0.65) -> GlinerRecognizer:
-    global _gliner
-    if _gliner is None:
-        _gliner = GlinerRecognizer(threshold=threshold)
-    return _gliner
-
-
-def _is_regex_result(r: RecognizerResult) -> bool:
-    if r.analysis_explanation is None:
-        return False
-    return r.analysis_explanation.pattern_name in _REGEX_RECOGNIZER_NAMES
-
-
-def _reclassify(results: list, text: str) -> list:
-    """Rinomina entità NER generiche → CIG/CUP/REA se span matcha pattern."""
-    for r in results:
-        if r.entity_type in _GENERIC_NER_ENTITIES:
-            span = text[r.start:r.end].strip()
-            for specific, pattern in _RECLASSIFY_PATTERNS.items():
-                if pattern.fullmatch(span):
-                    r.entity_type = specific
-                    break
-    return results
+SUPPORTED_ENTITIES: list[str] = sorted(set(_LABEL_TO_ENTITY.values()))
 
 
-def _scale_amount(original: str, scale_pct: float = 0.20) -> str:
-    """Scala importo ±scale_pct deterministicamente (hash MD5)."""
-    match = re.search(r"[\d\.,]+", original)
-    if not match:
-        return original
-    raw = match.group()
-    try:
-        value = float(raw.replace(".", "").replace(",", "."))
-    except ValueError:
-        return original
-    h = int(hashlib.md5(original.encode("utf-8")).hexdigest(), 16) % 10_000
-    factor = 1 + scale_pct * (h / 5_000 - 1)
-    scaled = value * factor
-    integer = int(scaled)
-    decimals = int(round((scaled - integer) * 100))
-    integer_str = f"{integer:,}".replace(",", ".")
-    return original[:match.start()] + f"{integer_str},{decimals:02d}" + original[match.end():]
-
-
-def _apply_mode(s: str, mode: str, label: str) -> str:
-    n = len(s)
-    if mode == "placeholder":
-        return f"[{label}]"
-    if mode == "last4":
-        keep = min(4, n)
-        return "*" * (n - keep) + s[-keep:]
-    if mode == "stars":
-        return "*" * n
-    if mode == "first":
-        return s[0] + "*" * (n - 1) if n > 1 else s
-    return f"[{label}]"
-
-
-# ---------------------------------------------------------------------------
-# Session – placeholder numerati coerenti per documento
-# ---------------------------------------------------------------------------
-class Session:
-    def __init__(self):
-        self._map: dict[tuple[str, str], str] = {}
-        self._counters: dict[str, int] = {}
-
-    def get_numbered(self, original: str, entity_type: str, prefix: str) -> str:
-        key = (original.strip(), entity_type)
-        if key in self._map:
-            return self._map[key]
-        self._counters[prefix] = self._counters.get(prefix, 0) + 1
-        placeholder = f"[{prefix}_{self._counters[prefix]:03d}]"
-        self._map[key] = placeholder
-        return placeholder
-
-
-def _replace(original: str, entity_type: str, mode: str, priority: int, session: Session) -> str:
-    if entity_type == "IMPORTO_BASE_ASTA" and mode == "placeholder":
-        return _scale_amount(original)
-    if priority == 2 and mode == "placeholder":
-        prefix = _PROCUREMENT_PREFIX.get(entity_type, entity_type)
-        return session.get_numbered(original, entity_type, prefix)
-    if entity_type in ("IMPORTO_GARA", "VALUTA"):
-        m = _CURRENCY_PREFIX.match(original)
-        if m:
-            sym  = m.group(0)
-            rest = original[m.end():]
-            label = LABEL_IT.get(entity_type, entity_type)
-            return sym + _apply_mode(rest, mode, label) if rest else sym
-    label = LABEL_IT.get(entity_type, entity_type)
-    return _apply_mode(original, mode, label)
-
-
-# ---------------------------------------------------------------------------
-# Ottimizzazione 1 – Chunking
-# ---------------------------------------------------------------------------
-def _chunk_text(text: str) -> list[tuple[int, str]]:
-    if len(text) <= _CHUNK_SIZE:
-        return [(0, text)]
-    chunks: list[tuple[int, str]] = []
-    start = 0
-    step  = _CHUNK_SIZE - _CHUNK_OVERLAP
-    while start < len(text):
-        end = min(start + _CHUNK_SIZE, len(text))
-        chunks.append((start, text[start:end]))
-        if end == len(text):
-            break
-        start += step
-    return chunks
-
-
-def _dedup_by_position(results: list[RecognizerResult]) -> list[RecognizerResult]:
-    seen: set[tuple[int, int, str]] = set()
-    unique: list[RecognizerResult] = []
-    for r in sorted(results, key=lambda x: -x.score):
-        key = (r.start, r.end, r.entity_type)
-        if key not in seen:
-            seen.add(key)
-            unique.append(r)
-    return unique
+class GlinerRecognizer(EntityRecognizer):
+    """
+    Wrapper Presidio per GLiNER zero-shot.
 
+    IMPORTANTE: self._model viene inizializzato PRIMA di super().__init__()
+    perché Presidio chiama self.load() durante l'inizializzazione del parent.
+    """
 
-def _run_analyzer_chunked(analyzer, text: str) -> list[RecognizerResult]:
-    chunks = _chunk_text(text)
-    if len(chunks) == 1:
-        return analyzer.analyze(text=text, language="it")
-    all_results: list[RecognizerResult] = []
-    for offset, chunk in chunks:
+    def __init__(
+        self,
+        threshold: float = 0.65,
+        labels: Optional[list[str]] = None,
+        model_name: str = "DeepMount00/GLiNER_PII_ITA",
+    ):
+        # ── DEVE essere prima di super().__init__() ────────────────────────
+        self._model     = None   # None = non caricato, False = caricamento fallito
+        self.threshold  = threshold
+        self.labels     = labels or PROCUREMENT_LABELS
+        self.model_name = model_name
+        # ──────────────────────────────────────────────────────────────────
+        super().__init__(
+            supported_entities=SUPPORTED_ENTITIES,
+            supported_language="it",
+            name="GlinerRecognizer",
+        )
+
+    def load(self) -> None:
+        """Carica il modello GLiNER (idempotente)."""
+        if not hasattr(self, "_model"):
+            self._model = None
+        if self._model is not None:
+            return
         try:
-            for r in analyzer.analyze(text=chunk, language="it"):
-                r.start += offset
-                r.end   += offset
-                all_results.append(r)
+            from gliner import GLiNER
+            self._model = GLiNER.from_pretrained(self.model_name)
+            logger.info("[GLiNER] Modello %s caricato", self.model_name)
         except Exception as exc:
-            logger.error("[chunk@%d] Presidio: %s", offset, exc)
-    return _dedup_by_position(all_results)
+            logger.error("[GLiNER] Caricamento fallito: %s", exc)
+            self._model = False
+
+    def analyze(self, text: str, entities: List[str], nlp_artifacts=None) -> List[RecognizerResult]:
+        if not text or not text.strip():
+            return []
 
+        if not hasattr(self, "_model") or self._model is None:
+            self.load()
+        if self._model is False:
+            return []
 
-def _run_gliner_chunked(gliner: GlinerRecognizer, text: str) -> list[RecognizerResult]:
-    chunks = _chunk_text(text)
-    if len(chunks) == 1:
-        return gliner.analyze(text=text, entities=[])
-    all_results: list[RecognizerResult] = []
-    for offset, chunk in chunks:
         try:
-            for r in gliner.analyze(text=chunk, entities=[]):
-                r.start += offset
-                r.end   += offset
-                all_results.append(r)
+            predictions = self._model.predict_entities(
+                text, self.labels, threshold=self.threshold,
+            )
         except Exception as exc:
-            logger.error("[chunk@%d] GLiNER: %s", offset, exc)
-    return _dedup_by_position(all_results)
-
-
-# ---------------------------------------------------------------------------
-# Ottimizzazione 2 – Cross-layer agreement boost
-# ---------------------------------------------------------------------------
-def _cross_layer_boost(layers: list[list[RecognizerResult]]) -> list[list[RecognizerResult]]:
-    """Score +0.15 se 2+ livelli rilevano lo stesso span (Jaccard ≥ 80%)."""
-    flat = [(i, r) for i, layer in enumerate(layers) for r in layer]
-
-    def _jaccard(a: RecognizerResult, b: RecognizerResult) -> float:
-        inter = max(0, min(a.end, b.end) - max(a.start, b.start))
-        union = max(a.end, b.end) - min(a.start, b.start)
-        return inter / union if union > 0 else 0.0
-
-    for i, r in flat:
-        agreeing: set[int] = {i}
-        for j, other in flat:
-            if i != j and _jaccard(r, other) >= 0.8:
-                agreeing.add(j)
-        if len(agreeing) >= 2:
-            r.score = min(_AGREEMENT_MAX, r.score + _AGREEMENT_DELTA)
-            if r.recognition_metadata is None:
-                r.recognition_metadata = {}
-            r.recognition_metadata["cross_layer_agreement"] = len(agreeing)
-    return layers
-
-
-# ---------------------------------------------------------------------------
-# Post-boost regex
-# ---------------------------------------------------------------------------
-def _post_boost_check(entities: list, text: str) -> list:
-    """Score +0.30 e flag post_boost=True se span matcha POST_BOOST_PATTERNS."""
-    for r in entities:
-        pattern = POST_BOOST_PATTERNS.get(r.entity_type)
-        if pattern and pattern.fullmatch(text[r.start:r.end].strip()):
-            r.score = min(_POST_BOOST_MAX, r.score + _POST_BOOST_DELTA)
-            if r.recognition_metadata is None:
-                r.recognition_metadata = {}
-            r.recognition_metadata["post_boost"] = True
-    return entities
-
-
-# ---------------------------------------------------------------------------
-# Filtro falsi positivi da re-parsing
-# ---------------------------------------------------------------------------
-def _filter_noise(entities: list, text: str) -> list:
-    """
-    Rimuove falsi positivi prodotti da re-parsing di testo già offuscato.
-
-    Casi gestiti:
-     • Span con >50% asterischi → già mascherato, salta
-     • VALUTA/IMPORTO senza cifre → simbolo isolato (€, EUR), salta
-     • FREQUENZA con score < 0.80 → falso positivo comune, salta
-     • Span che inizia/finisce con parentesi aperta → boundary NER rotto, salta
-    """
-    clean: list = []
-    for r in entities:
-        span = text[r.start:r.end]
-
-        # Span quasi interamente asterischi (testo già mascherato)
-        if len(span) > 1 and span.count("*") / len(span) > 0.5:
-            continue
-
-        # VALUTA/IMPORTO senza nessuna cifra → simbolo isolato
-        if r.entity_type in ("VALUTA", "IMPORTO_GARA"):
-            if not re.search(r"\d", span):
-                continue
-
-        # FREQUENZA generica senza context sufficiente
-        if r.entity_type == "FREQUENZA" and r.score < 0.80:
-            continue
-
-        # Boundary NER rotto: span inizia o finisce con parentesi aperta
-        stripped = span.strip()
-        if stripped.startswith("(") or stripped.endswith("("):
-            continue
-
-        clean.append(r)
-    return clean
-
-
-# ---------------------------------------------------------------------------
-# Pipeline pubblica
-# ---------------------------------------------------------------------------
-def detect(text: str, min_score: float = 0.65, use_regex: bool = True) -> list[RecognizerResult]:
-    """Rilevazione a 3 livelli con chunking + cross-layer boost + post-boost + noise filter."""
-    if not text or not text.strip():
-        return []
-
-    analyzer = analyzer_full if use_regex else analyzer_ner_only
-    try:
-        presidio_results = _run_analyzer_chunked(analyzer, text)
-    except Exception as exc:
-        logger.error("[pipeline] Presidio: %s", exc)
-        presidio_results = []
-
-    if use_regex:
-        regex_results = [r for r in presidio_results if _is_regex_result(r)]
-        ner_results   = [r for r in presidio_results if not _is_regex_result(r)]
-    else:
-        regex_results = []
-        ner_results   = presidio_results
-
-    ner_results = _reclassify(ner_results, text)
-
-    try:
-        gliner_results = _run_gliner_chunked(_get_gliner(threshold=min_score), text)
-    except Exception as exc:
-        logger.error("[pipeline] GLiNER: %s", exc)
-        gliner_results = []
-
-    regex_results  = [r for r in regex_results  if r.score >= min_score]
-    ner_results    = [r for r in ner_results    if r.score >= min_score]
-    gliner_results = [r for r in gliner_results if r.score >= min_score]
-
-    regex_results, ner_results, gliner_results = _cross_layer_boost(
-        [regex_results, ner_results, gliner_results]
-    )
-
-    final = resolve_overlapping_spans([regex_results, ner_results, gliner_results])
-    final = _post_boost_check(final, text)
-    final = _filter_noise(final, text)
-    return final
-
-
-def anonymize_from_entities(text: str, entities: list[RecognizerResult], mode_label: str) -> str:
-    """Sostituisce le entità nel testo secondo la modalità scelta."""
-    if not entities:
-        return text
-    mode = MODES.get(mode_label, "placeholder")
-    session = Session()
-    for r in sorted(entities, key=lambda r: r.start, reverse=True):
-        original = text[r.start:r.end]
-        priority = (r.recognition_metadata or {}).get("source_priority", -1)
-        replacement = _replace(original, r.entity_type, mode, priority, session)
-        text = text[:r.start] + replacement + text[r.end:]
-    return text
-
-
-def anonymize(
-    text: str,
-    mode_label: str,
-    min_score: float = 0.65,
-    use_regex: bool = True,
-) -> tuple[str, list[RecognizerResult]]:
-    """Pipeline completa: detect + anonimizza. Returns (testo_offuscato, entità)."""
-    if not text or not text.strip():
-        return "", []
-    entities = detect(text, min_score=min_score, use_regex=use_regex)
-    return anonymize_from_entities(text, entities, mode_label), entities
+            logger.error("[GLiNER] Predizione fallita: %s", exc)
+            return []
+
+        results: list[RecognizerResult] = []
+        for p in predictions:
+            entity_type = _LABEL_TO_ENTITY.get(p["label"], "PROCUREMENT_ENTITY")
+            results.append(RecognizerResult(
+                entity_type=entity_type,
+                start=p["start"],
+                end=p["end"],
+                score=float(p.get("score", self.threshold)),
+                analysis_explanation=None,
+                recognition_metadata={
+                    "recognizer_name": "GlinerRecognizer",
+                    "gliner_label":    p["label"],
+                },
+            ))
+        return results

pipeline.py

@@ -1,4 +1,4 @@
-"""Pipeline a 3 livelli: Regex (opz.) → NER → GLiNER, con chunking e agreement boost."""
+"""Pipeline a 3 livelli: Regex (opz.) → NER → GLiNER, chunking, agreement boost."""
 import hashlib
 import logging
 import re
@@ -60,7 +60,7 @@ _gliner: Optional[GlinerRecognizer] = None
 
 
 # ---------------------------------------------------------------------------
-# Helpers
+# Helpers – singleton, utils
 # ---------------------------------------------------------------------------
 def _get_gliner(threshold: float = 0.65) -> GlinerRecognizer:
     global _gliner
@@ -88,7 +88,7 @@ def _reclassify(results: list, text: str) -> list:
 
 
 def _scale_amount(original: str, scale_pct: float = 0.20) -> str:
-    """Scala importo numerico ±scale_pct deterministicamente (hash MD5)."""
+    """Scala importo ±scale_pct deterministicamente (hash MD5)."""
     match = re.search(r"[\d\.,]+", original)
     if not match:
         return original
@@ -121,7 +121,7 @@ def _apply_mode(s: str, mode: str, label: str) -> str:
 
 
 # ---------------------------------------------------------------------------
-# Session – placeholder numerati coerenti
+# Session – placeholder numerati coerenti per documento
 # ---------------------------------------------------------------------------
 class Session:
     def __init__(self):
@@ -244,6 +244,21 @@ def _cross_layer_boost(layers: list[list[RecognizerResult]]) -> list[list[Recogn
 # ---------------------------------------------------------------------------
 # Post-boost regex
 # ---------------------------------------------------------------------------
+def _post_boost_check(entities: list, text: str) -> list:
+    """Score +0.30 e flag post_boost=True se span matcha POST_BOOST_PATTERNS."""
+    for r in entities:
+        pattern = POST_BOOST_PATTERNS.get(r.entity_type)
+        if pattern and pattern.fullmatch(text[r.start:r.end].strip()):
+            r.score = min(_POST_BOOST_MAX, r.score + _POST_BOOST_DELTA)
+            if r.recognition_metadata is None:
+                r.recognition_metadata = {}
+            r.recognition_metadata["post_boost"] = True
+    return entities
+
+
+# ---------------------------------------------------------------------------
+# Filtro falsi positivi da re-parsing
+# ---------------------------------------------------------------------------
 def _filter_noise(entities: list, text: str) -> list:
     """
     Rimuove falsi positivi prodotti da re-parsing di testo già offuscato.
@@ -251,9 +266,8 @@ def _filter_noise(entities: list, text: str) -> list:
     Casi gestiti:
      • Span con >50% asterischi → già mascherato, salta
      • VALUTA/IMPORTO senza cifre → simbolo isolato (€, EUR), salta
-     • FREQUENZA con span di 1 parola e score < 0.80 → troppo generico, salta
-     • Entità che iniziano o finiscono con una parentesi aperta/chiusa
-       (artefatto di boundary NER) → salta
+     • FREQUENZA con score < 0.80 → falso positivo comune, salta
+     • Span che inizia/finisce con parentesi aperta → boundary NER rotto, salta
     """
     clean: list = []
     for r in entities:
@@ -268,33 +282,24 @@ def _filter_noise(entities: list, text: str) -> list:
             if not re.search(r"\d", span):
                 continue
 
-        # FREQUENZA con singola parola senza contesto → falso positivo comune
+        # FREQUENZA generica senza context sufficiente
         if r.entity_type == "FREQUENZA" and r.score < 0.80:
             continue
 
-        # Boundary NER rotto: span inizia o finisce con una parentesi
+        # Boundary NER rotto: span inizia o finisce con parentesi aperta
         stripped = span.strip()
         if stripped.startswith("(") or stripped.endswith("("):
             continue
 
         clean.append(r)
     return clean
-    """Score +0.30 e flag post_boost=True se span matcha POST_BOOST_PATTERNS."""
-    for r in entities:
-        pattern = POST_BOOST_PATTERNS.get(r.entity_type)
-        if pattern and pattern.fullmatch(text[r.start:r.end].strip()):
-            r.score = min(_POST_BOOST_MAX, r.score + _POST_BOOST_DELTA)
-            if r.recognition_metadata is None:
-                r.recognition_metadata = {}
-            r.recognition_metadata["post_boost"] = True
-    return entities
 
 
 # ---------------------------------------------------------------------------
 # Pipeline pubblica
 # ---------------------------------------------------------------------------
 def detect(text: str, min_score: float = 0.65, use_regex: bool = True) -> list[RecognizerResult]:
-    """Rilevazione a 3 livelli con chunking + cross-layer boost + post-boost."""
+    """Rilevazione a 3 livelli con chunking + cross-layer boost + post-boost + noise filter."""
     if not text or not text.strip():
         return []