import re from recognizers import analyzer_full, analyzer_ner_only from config import MODES, LABEL_IT # Simboli valuta da preservare quando si offusca un IMPORTO _CURRENCY_PREFIX = re.compile(r"^(€|EUR|\$|£)\s*", flags=re.IGNORECASE) # Pattern per riclassificare entità generiche del NER in entità specifiche appalti. # Es. il modello può etichettare un CIG come NUMERO_DOCUMENTO o N_LICENZA — qui # verifichiamo se rientra in un nostro pattern e rinominiamo l'etichetta. _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][0-9A-Z]{14}$"), "REA": re.compile(r"^[A-Z]{2}-\d{4,7}$"), } # Entità generiche del NER candidate alla riclassificazione _GENERIC_NER_ENTITIES: set[str] = { "NUMERO_DOCUMENTO", "N_LICENZA", "N_SENTENZA", "NUMERO_CONTO", } def _apply_mode(s: str, mode: str, label: str) -> str: """Apply the chosen anonymization mode to a string.""" n = len(s) if mode == "placeholder": return f"[{label}]" elif mode == "last4": keep = min(4, n) return "*" * (n - keep) + s[-keep:] elif mode == "stars": return "*" * n elif mode == "first": return s[0] + "*" * (n - 1) if n > 1 else s return f"[{label}]" def _replace(original: str, entity_type: str, mode: str) -> str: """Return the anonymized string for a single entity span.""" label = LABEL_IT.get(entity_type, entity_type) # IMPORTO/VALUTA → preserva il simbolo valuta se presente if entity_type in ("IMPORTO_GARA", "VALUTA"): match = _CURRENCY_PREFIX.match(original) if match: prefix = match.group(0) rest = original[match.end():] return prefix + _apply_mode(rest, mode, label) if rest else prefix return _apply_mode(original, mode, label) def _reclassify(results: list, text: str) -> list: """ Per ogni rilevazione con etichetta NER generica, verifica se il testo matcha un pattern specifico (CIG/CUP/REA): in tal caso sostituisce l'entity_type. """ for r in results: if r.entity_type in _GENERIC_NER_ENTITIES: span = text[r.start:r.end].strip() for specific_label, pattern in _RECLASSIFY_PATTERNS.items(): if pattern.fullmatch(span): r.entity_type = specific_label break return results def _remove_overlaps(results: list) -> list: """Keep highest-score results when spans overlap.""" sorted_results = sorted(results, key=lambda r: (-r.score, -(r.end - r.start))) kept = [] for r in sorted_results: if not any(r.start < k.end and k.start < r.end for k in kept): kept.append(r) return kept def anonymize(text: str, mode_label: str, min_score: float, use_regex: bool = True) -> tuple[str, str]: """Detect PII entities, filter by min_score, and anonymize. Args: text: input text mode_label: human-readable mode label from MODES min_score: minimum confidence score use_regex: if True usa NER + regex appalti; se False solo NER """ if not text or not text.strip(): return "", "⚠️ Nessun testo fornito." mode = MODES.get(mode_label, "placeholder") analyzer = analyzer_full if use_regex else analyzer_ner_only results = analyzer.analyze(text=text, language="it") results = [r for r in results if r.score >= min_score] results = _reclassify(results, text) results = _remove_overlaps(results) if not results: return text, "ℹ️ Nessuna entità rilevata sopra lo score minimo." # Replace from end → start to preserve char offsets results_desc = sorted(results, key=lambda r: r.start, reverse=True) anon_text = text report_lines: list[tuple[int, str]] = [] for r in results_desc: original = text[r.start:r.end] replacement = _replace(original, r.entity_type, mode) anon_text = anon_text[:r.start] + replacement + anon_text[r.end:] label = LABEL_IT.get(r.entity_type, r.entity_type) report_lines.append( (r.start, f" [{label}] '{original}' → '{replacement}' (score: {r.score:.2f})") ) report = "\n".join(line for _, line in sorted(report_lines, key=lambda x: x[0])) return anon_text, report