| import streamlit as st |
| import torch |
| import torch.nn as nn |
| import os |
| import re |
| import json |
| from pathlib import Path |
| from transformers import AutoTokenizer, AutoModelForTokenClassification, AutoModel |
| from collections import defaultdict |
| from huggingface_hub import hf_hub_download |
|
|
| |
| MODEL_NER_PATH = "inesctec/CitiLink-XLMR-Anonymization-pt" |
| MODEL_COREF_PATH = "inesctec/CitiLink-SpanBERT-Coreference-pt" |
|
|
| LABEL_NAMES = { |
| "Name": ("Name", "Nome"), |
| "AdministrativeInformation": ("Admin", "Admin"), |
| "PositionDepartment": ("Position", "Cargo"), |
| "Address": ("Address", "Morada"), |
| "Date": ("Date", "Data"), |
| "Location": ("Location", "Local"), |
| "PersonalInformation": ("Info", "Info"), |
| "Company": ("Company", "Empresa"), |
| "ArtisticActivity": ("Artistic", "Artístico"), |
| "Degree": ("Degree", "Curso"), |
| "Time": ("Time", "Hora"), |
| "LicensePlate": ("License", "Matrícula"), |
| "Job": ("Job", "Profissão"), |
| "Vehicle": ("Vehicle", "Veículo"), |
| "Faculty": ("Faculty", "Faculdade"), |
| "Family": ("Family", "Família"), |
| "Other": ("Other", "Outro"), |
| } |
| UI = { |
| "page_title": ("Text Pseudonymization Demo", "Demo de Pseudonimização de Texto"), |
| "page_subtitle": ("Automatic text pseudonymization using NER + Coreference Resolution for municipal minutes", |
| "Pseudonimização automática de texto utilizando NER + Resolução de Correferência para atas municipais"), |
| "sidebar_header": ("⚙️ Configuration", "⚙️ Configuração"), |
| "choose_example": ("Choose an example:", "Escolha um exemplo:"), |
| "lang_toggle_label": ("Language", "Idioma"), |
| "about_header": ("### 📊 About", "### 📊 Sobre"), |
| "about_text": ( |
| "- **Anonymization (NER)** uses Token Classification to identify sensitive entities.\n" |
| "- **Linking (Coreference Resolution)** links mentions of the same entity across the document.\n" |
| "- **Model (NER)**: XLM-RoBERTa fine-tuned for Portuguese NER.\n" |
| "- **Model (Coreference Resolution)**: SpanBERT fine-tuned on Portuguese administrative texts.\n" |
| "- **Language**: Portuguese (pt-pt).", |
| "- **Anonimização (NER)** usa Classificação de Tokens para identificar entidades sensíveis.\n" |
| "- **Ligação (Resolução de Correferência)** liga menções da mesma entidade ao longo do documento.\n" |
| "- **Modelo (NER)**: XLM-RoBERTa ajustado para NER em Português.\n" |
| "- **Modelo (Resolução de Correferência)**: SpanBERT ajustado em textos administrativos portugueses.\n" |
| "- **Língua**: Português (PT-PT)." |
| ), |
| "resources_header": ("### 🔗 Resources", "### 🔗 Recursos"), |
| "how_it_works": ("🎯 How it works", "🎯 Como funciona"), |
| "how_text": ( |
| "The pseudonymization pipeline uses two specialized models:\n\n" |
| "1. **Detection (NER):** Identifies all sensitive entity spans across 17 categories.\n" |
| "2. **Linking (Coreference Resolution):** Detects when different mentions refer to the same real-world entity, assigning consistent IDs.", |
| "O pipeline de pseudonimização utiliza dois modelos especializados:\n\n" |
| "1. **Deteção (NER):** Identifica todos os spans de entidades sensíveis em 17 categorias.\n" |
| "2. **Ligação (Resolução de Correferência):** Deteta quando diferentes menções se referem à mesma entidade real, atribuindo IDs consistentes." |
| ), |
| "entities_title": ("Supported Entities:", |
| "Entidades Suportadas:"), |
| "input_label": ("INPUT:", "ENTRADA:"), |
| "output_label": ("OUTPUT:", "SAÍDA:"), |
| "col1_header": ("📝 Input Document", "📝 Documento de Entrada"), |
| "col2_header": ("🔒 Pseudonymization Results", "🔒 Resultados da Pseudonimização"), |
| "textarea_label": ("Enter your text here:", "Introduza o texto aqui:"), |
| "textarea_placeholder": ("Paste your document text here...", "Cole aqui o texto do documento..."), |
| "anonymize_btn": ("🔍 Pseudonymize", "🔍 Pseudonimizar"), |
| "spinner_text": ("Processing...", "A processar..."), |
| "tab_anon": ("📄 Pseudonymized Text", "📄 Texto Pseudonimizado"), |
| "tab_entities": ("🔍 Extracted Entities", "🔍 Entidades Extraídas"), |
| "tab_groups": ("🔗 Coreference Groups", "🔗 Grupos de Correferência"), |
| "no_groups": ("No coreference groups found (no entity appears more than once).", |
| "Nenhum grupo de correferência encontrado (nenhuma entidade aparece mais do que uma vez)."), |
| "group_label": ("Group", "Grupo"), |
| "mentions_label": ("mentions", "menções"), |
| "no_entities": ("No entities found.", "Nenhuma entidade encontrada."), |
| "json_expander": ("📋 Full Entity Report (JSON)", "📋 Relatório Completo (JSON)"), |
| "download_btn": ("📥 Download JSON Report", "📥 Transferir Relatório JSON"), |
| "no_results_info": ("Please process a document on the left to view the results.", |
| "Por favor, processe um documento à esquerda para ver os resultados."), |
| "metric_entities": ("Entities", "Entidades"), |
| "metric_categories": ("Categories", "Categorias"), |
| "metric_clusters": ("Groups", "Grupos"), |
| "example_names": ( |
| { |
| "Custom Text": "Custom Text", |
| "1º Portuguese Meeting Minute": "1º Portuguese Meeting Minute", |
| "2º Portuguese Meeting Minute": "2º Portuguese Meeting Minute", |
| "3º Portuguese Meeting Minute": "3º Portuguese Meeting Minute", |
| }, |
| { |
| "Custom Text": "Texto Livre", |
| "1º Portuguese Meeting Minute": "1ª Ata de Reunião Municipal", |
| "2º Portuguese Meeting Minute": "2ª Ata de Reunião Municipal", |
| "3º Portuguese Meeting Minute": "3ª Ata de Reunião Municipal", |
| } |
| ), |
| } |
| st.set_page_config( |
| page_title="Text Pseudonymization Demo", |
| page_icon="🛡️", |
| layout="wide", |
| initial_sidebar_state="expanded" |
| ) |
| st.markdown(""" |
| <style> |
| [data-testid="stSidebarResizer"] { display: none !important; } |
| [data-testid="stSidebar"][aria-expanded="true"] { min-width: 360px; max-width: 360px; } |
| [data-testid="stMain"] { margin-left: 0px; } |
| .small-metric-container { display: flex; justify-content: space-between; gap: 10px; margin-top: 10px; } |
| .small-metric-box { |
| flex: 1; background-color: #1e212b; border-radius: 6px; |
| padding: 5px; text-align: center; border: 1px solid #3a3d4a; |
| } |
| .metric-label { font-size: 0.65rem; color: #6c757d; text-transform: uppercase; font-weight: bold; } |
| .metric-value { font-size: 1rem; color: #a8dadc; font-weight: bold; } |
| .tab-window { |
| height: 495px; overflow-y: auto; padding: 20px; |
| border: 1px solid #444; border-radius: 0px 0px 8px 8px; |
| background-color: #262730 !important; |
| font-family: 'Consolas', 'Monaco', monospace; |
| font-size: 0.95rem; line-height: 1.6; user-select: text; |
| color: #efefef !important; margin-bottom: 20px; |
| } |
| .entity-highlight { |
| background-color: #3d3f4b; color: #a8dadc; |
| padding: 2px 6px; border-radius: 4px; border: 1px solid #555; |
| display: inline-block; line-height: 1.2; font-weight: bold; |
| } |
| .stTabs [data-baseweb="tab-list"] button [data-testid="stMarkdownContainer"] p { |
| color: #9ca3af !important; font-size: 1rem; font-weight: bold; |
| } |
| .stTabs [aria-selected="true"] [data-testid="stMarkdownContainer"] p { color: #a8dadc !important; } |
| .stTabs [data-baseweb="tab-highlight"] { background-color: #a8dadc !important; } |
| .stTabs [data-baseweb="tab"]:hover [data-testid="stMarkdownContainer"] p { color: #ffffff !important; } |
| .block-container { padding-top: 1.5rem !important; padding-bottom: 0rem !important; } |
| header[data-testid="stHeader"] { background: transparent !important; color: transparent !important; } |
| [data-testid="collapsedControl"] { color: #bdbbbb !important; visibility: visible !important; display: flex !important; } |
| div[data-testid="stTextArea"] textarea { resize: none; } |
| /* Toggle PT / EN */ |
| .lang-toggle-wrap { |
| display: flex; align-items: center; justify-content: center; |
| gap: 0; border-radius: 20px; overflow: hidden; |
| border: 1px solid #3a3d4a; width: 100%; margin-bottom: 6px; |
| } |
| .lang-btn { |
| flex: 1; padding: 6px 0; text-align: center; |
| font-size: 0.82rem; font-weight: bold; cursor: pointer; |
| border: none; transition: background 0.2s, color 0.2s; |
| } |
| .lang-btn-active { background: #a8dadc; color: #1a1a2e; } |
| .lang-btn-inactive { background: #1e212b; color: #6c757d; } |
| </style> |
| """, unsafe_allow_html=True) |
|
|
|
|
| |
| def lang_idx(): |
| return 1 if st.session_state.get("lang_pt", True) else 0 |
|
|
|
|
| |
| def t(key): |
| return UI[key][lang_idx()] |
|
|
|
|
| |
| def label_name(internal_tag): |
| entry = LABEL_NAMES.get(internal_tag) |
| if entry is None: |
| return internal_tag |
| return entry[lang_idx()] |
|
|
|
|
| |
| def bucket_distance(d): |
| buckets = [1, 2, 3, 4, 5, 8, 16, 32, 64] |
| for i, b in enumerate(buckets): |
| if d <= b: |
| return i |
| return len(buckets) |
|
|
|
|
| |
| class FFNN(nn.Module): |
| def __init__(self, input_dim, hidden_dim, output_dim, dropout=0.3): |
| super().__init__() |
| self.net = nn.Sequential( |
| nn.Linear(input_dim, hidden_dim), |
| nn.ReLU(), |
| nn.Dropout(dropout), |
| nn.Linear(hidden_dim, hidden_dim), |
| nn.ReLU(), |
| nn.Dropout(dropout), |
| nn.Linear(hidden_dim, output_dim) |
| ) |
|
|
| def forward(self, x): |
| return self.net(x) |
|
|
|
|
| |
| class CorefModel(nn.Module): |
| def __init__(self, model_dir, ffnn_dim=1024, max_span_width=52, |
| max_antecedents=50, max_cluster_size=12, dropout=0.3): |
| super().__init__() |
| self.encoder = AutoModel.from_pretrained(model_dir) |
| self.hidden = self.encoder.config.hidden_size |
| self.max_span_width = max_span_width |
| self.max_antecedents = max_antecedents |
| self.max_cluster_size = max_cluster_size |
| self.width_emb = nn.Embedding(max_span_width + 1, 20) |
| self.dist_emb = nn.Embedding(10, 20) |
| span_dim = self.hidden * 3 + 20 |
| self.mention_scorer = FFNN(span_dim, ffnn_dim, 1, dropout) |
| ant_dim = span_dim * 3 + 20 |
| self.antecedent_scorer = FFNN(ant_dim, ffnn_dim, 1, dropout) |
| self.span_attn = nn.Linear(self.hidden, 1) |
|
|
| |
| def span_repr(self, token_embs, start, end): |
| h_start = token_embs[start] |
| h_end = token_embs[min(end, len(token_embs) - 1)] |
| span_tokens = token_embs[start: min(end + 1, len(token_embs))] |
| attn_w = torch.softmax(self.span_attn(span_tokens), dim=0) |
| head = (attn_w * span_tokens).sum(0) |
| width = torch.tensor(min(end - start, self.max_span_width), dtype=torch.long) |
| w_emb = self.width_emb(width) |
| return torch.cat([h_start, h_end, head, w_emb], dim=-1) |
|
|
| |
| def forward(self, input_ids, attention_mask, spans): |
| out = self.encoder(input_ids=input_ids, attention_mask=attention_mask) |
| token_embs = out.last_hidden_state[0] |
| span_vecs = [] |
| for (s, e) in spans: |
| span_vecs.append(self.span_repr(token_embs, s, e)) |
| if not span_vecs: |
| return [] |
| span_vecs = torch.stack(span_vecs) |
| m_scores = self.mention_scorer(span_vecs).squeeze(-1) |
| n = len(spans) |
| pairs = [] |
| for i in range(n): |
| for j in range(max(0, i - self.max_antecedents), i): |
| d = i - j |
| d_idx = torch.tensor(bucket_distance(d), dtype=torch.long) |
| d_emb = self.dist_emb(d_idx) |
| v = torch.cat([span_vecs[i], span_vecs[j], |
| span_vecs[i] * span_vecs[j], d_emb], dim=-1) |
| score = (m_scores[i] + m_scores[j] |
| + self.antecedent_scorer(v).squeeze()) |
| pairs.append((i, j, score.item())) |
| return pairs |
|
|
|
|
| |
| def find(parent, x): |
| if parent[x] != x: |
| parent[x] = find(parent, parent[x]) |
| return parent[x] |
|
|
|
|
| |
| def union(parent, rank, a, b): |
| ra, rb = find(parent, a), find(parent, b) |
| if ra == rb: |
| return |
| if rank[ra] < rank[rb]: |
| ra, rb = rb, ra |
| parent[rb] = ra |
| if rank[ra] == rank[rb]: |
| rank[ra] += 1 |
|
|
|
|
| @st.cache_resource |
| |
| def load_models(): |
| try: |
| tokenizer_ner = AutoTokenizer.from_pretrained(MODEL_NER_PATH, add_prefix_space=True) |
| model_ner = AutoModelForTokenClassification.from_pretrained(MODEL_NER_PATH) |
| model_ner.eval() |
|
|
| coref_cfg = { |
| "ffnn_dim": 1024, |
| "max_span_width": 52, |
| "max_antecedents": 50, |
| "max_cluster_size": 12, |
| "dropout": 0.3, |
| "max_tokens": 512, |
| "stride": 128, |
| "epsilon": 0.0001, |
| "min_cluster_score": 0.3, |
| "ner_union_margin": 0.5, |
| } |
|
|
| model_coref = CorefModel( |
| model_dir=MODEL_COREF_PATH, |
| ffnn_dim=coref_cfg["ffnn_dim"], |
| max_span_width=coref_cfg["max_span_width"], |
| max_antecedents=coref_cfg["max_antecedents"], |
| max_cluster_size=coref_cfg["max_cluster_size"], |
| dropout=coref_cfg["dropout"], |
| ) |
| model_coref = CorefModel( |
| model_dir=MODEL_COREF_PATH, |
| ffnn_dim=coref_cfg["ffnn_dim"], |
| max_span_width=coref_cfg["max_span_width"], |
| max_antecedents=coref_cfg["max_antecedents"], |
| max_cluster_size=coref_cfg["max_cluster_size"], |
| dropout=coref_cfg["dropout"], |
| ) |
|
|
| cabecas_path = hf_hub_download( |
| repo_id=MODEL_COREF_PATH, |
| filename="cabecas_coref.pt" |
| ) |
| cabecas = torch.load(cabecas_path, map_location="cpu", weights_only=False) |
| model_coref.load_state_dict(cabecas, strict=False) |
|
|
| model_coref.eval() |
|
|
| tokenizer_coref = AutoTokenizer.from_pretrained(MODEL_COREF_PATH) |
| return tokenizer_ner, model_ner, model_coref, tokenizer_coref, coref_cfg, None |
| except Exception as e: |
| return None, None, None, None, None, str(e) |
|
|
| |
| def aggregate_token_embeddings(model_coref, tokenizer_coref, tokens, max_tokens=512, stride=128, device="cpu"): |
| emb_sum = torch.zeros(len(tokens), model_coref.hidden) |
| emb_count = torch.zeros(len(tokens)) |
| encoded = tokenizer_coref( |
| tokens, is_split_into_words=True, |
| return_offsets_mapping=False, add_special_tokens=False, return_tensors=None |
| ) |
| word_ids_full = encoded.word_ids() |
| subword_ids = encoded["input_ids"] |
| total_subwords = len(subword_ids) |
| start = 0 |
| while start < total_subwords: |
| end = min(start + max_tokens - 2, total_subwords) |
| chunk_ids = [tokenizer_coref.cls_token_id] + subword_ids[start:end] + [tokenizer_coref.sep_token_id] |
| chunk_tensor = torch.tensor([chunk_ids], dtype=torch.long).to(device) |
| attn_mask = torch.ones_like(chunk_tensor) |
| with torch.no_grad(): |
| out = model_coref.encoder(input_ids=chunk_tensor, attention_mask=attn_mask) |
| hidden = out.last_hidden_state[0] |
| for sub_pos, (sub_id, h) in enumerate(zip(range(start, end), hidden[1:-1])): |
| word_idx = word_ids_full[sub_id] |
| if word_idx is not None: |
| emb_sum[word_idx] += h.cpu() |
| emb_count[word_idx] += 1 |
| if end >= total_subwords: |
| break |
| start += max_tokens - 2 - stride |
| count_safe = emb_count.unsqueeze(-1).clamp(min=1) |
| return emb_sum / count_safe |
|
|
|
|
| |
| def predict_coref_clusters(model_coref, tokenizer_coref, tokens, ner_spans, coref_cfg, device="cpu"): |
| if not ner_spans: |
| return [] |
| max_tokens = coref_cfg.get("max_tokens", 512) |
| stride = coref_cfg.get("stride", 128) |
| epsilon = coref_cfg.get("epsilon", 0.0001) |
| min_cluster_score = coref_cfg.get("min_cluster_score", 0.3) |
| ner_union_margin = coref_cfg.get("ner_union_margin", 0.5) |
| threshold = max(epsilon, min_cluster_score) + ner_union_margin |
| token_embs = aggregate_token_embeddings(model_coref, tokenizer_coref, tokens, max_tokens, stride, device) |
| spans = [(s, e) for (s, e, _) in ner_spans] |
| span_labels = [lbl for (_, _, lbl) in ner_spans] |
| model_coref.eval() |
| with torch.no_grad(): |
| span_vecs = [] |
| for (s, e) in spans: |
| sv = model_coref.span_repr(token_embs, s, e) |
| span_vecs.append(sv) |
| if not span_vecs: |
| return [] |
| span_vecs_t = torch.stack(span_vecs) |
| m_scores = model_coref.mention_scorer(span_vecs_t).squeeze(-1) |
| n = len(spans) |
| parent = list(range(n)) |
| rank = [0] * n |
| for i in range(n): |
| for j in range(max(0, i - model_coref.max_antecedents), i): |
| if span_labels[i] != span_labels[j]: |
| continue |
| d = i - j |
| d_idx = torch.tensor(bucket_distance(d), dtype=torch.long) |
| d_emb = model_coref.dist_emb(d_idx) |
| v = torch.cat([span_vecs_t[i], span_vecs_t[j], |
| span_vecs_t[i] * span_vecs_t[j], d_emb], dim=-1) |
| score = (m_scores[i] + m_scores[j] |
| + model_coref.antecedent_scorer(v).squeeze()).item() |
| if score > threshold: |
| union(parent, rank, i, j) |
| cluster_map = defaultdict(list) |
| for i, (s, e) in enumerate(spans): |
| root = find(parent, i) |
| cluster_map[root].append((s, e)) |
| return [v for v in cluster_map.values() if len(v) > 1] |
|
|
|
|
| |
| def process_anonymization(text, tokenizer_ner, model_ner, model_coref, tokenizer_coref, coref_cfg): |
| if not text.strip(): |
| return "Por favor, insira um texto.", [] |
| |
| id2label = model_ner.config.id2label |
| inputs = tokenizer_ner( |
| text, truncation=True, max_length=512, stride=164, |
| return_overflowing_tokens=True, return_offsets_mapping=True, |
| padding=True, return_tensors="pt" |
| ) |
| offset_mapping_all = inputs.pop("offset_mapping") |
| inputs.pop("overflow_to_sample_mapping") |
| input_ids_all = inputs["input_ids"] |
| with torch.no_grad(): |
| logits = model_ner(**inputs).logits |
| all_predictions = torch.argmax(logits, dim=2).tolist() |
| SPACE_PREFIXES = [" ", "▁", "Ġ"] |
| entidades_brutas = [] |
| for window_idx, (predictions, offsets) in enumerate(zip(all_predictions, offset_mapping_all)): |
| temp_entity = {"tokens": [], "start": None, "end": None, "label": None} |
| for idx, (pred_id, offset) in enumerate(zip(predictions, offsets)): |
| label_name_str = id2label[pred_id] |
| start_char, end_char = int(offset[0]), int(offset[1]) |
| if start_char == end_char: |
| continue |
| tag_base = label_name_str.split('-', 1)[1] if '-' in label_name_str else None |
| token_id = input_ids_all[window_idx][idx].item() |
| token = tokenizer_ner.convert_ids_to_tokens(token_id) |
| comeca_nova_palavra = any(token.startswith(p) for p in SPACE_PREFIXES) |
| if label_name_str.startswith("B-"): |
| if temp_entity["label"]: |
| chars_to_remove = ".,;:!?" if temp_entity["label"] != "PERSONAL-PositionDepartment" else ",;:!?" |
| while (temp_entity["end"] - temp_entity["start"]) > 0 and text[temp_entity["end"] - 1] in chars_to_remove: |
| temp_entity["end"] -= 1 |
| entidades_brutas.append(temp_entity) |
| temp_entity = {"start": start_char, "end": end_char, "label": tag_base, "tokens": [token]} |
| elif label_name_str.startswith("I-") and temp_entity["label"] == tag_base: |
| temp_entity["tokens"].append(token) |
| temp_entity["end"] = end_char |
| elif not comeca_nova_palavra and temp_entity["label"] is not None: |
| temp_entity["tokens"].append(token) |
| temp_entity["end"] = end_char |
| else: |
| if temp_entity["label"]: |
| chars_to_remove = ".,;:!?" |
| while (temp_entity["end"] - temp_entity["start"]) > 0 and text[temp_entity["end"] - 1] in chars_to_remove: |
| temp_entity["end"] -= 1 |
| entidades_brutas.append(temp_entity) |
| temp_entity = {"tokens": [], "start": None, "end": None, "label": None} |
| if temp_entity["label"]: |
| entidades_brutas.append(temp_entity) |
| entidades_brutas.sort(key=lambda x: x["start"]) |
| entidades_finais = [] |
| for atual in entidades_brutas: |
| if not entidades_finais: |
| entidades_finais.append(atual) |
| continue |
| ultima = entidades_finais[-1] |
| distancia = atual["start"] - ultima["end"] |
| if distancia <= 1 and atual["label"] == ultima["label"]: |
| ultima["end"] = atual["end"] |
| ultima["tokens"].extend(atual["tokens"]) |
| else: |
| adicionar = True |
| for i, selecionada in enumerate(entidades_finais): |
| interseccao = max(0, min(atual["end"], selecionada["end"]) - max(atual["start"], selecionada["start"])) |
| if interseccao > 0: |
| if (atual["end"] - atual["start"]) > (selecionada["end"] - selecionada["start"]): |
| entidades_finais[i] = atual |
| adicionar = False |
| break |
| if adicionar: |
| entidades_finais.append(atual) |
| SINAIS_REMOVER = ".-,;:_^~´`/*+!?\"'" |
| for ent in entidades_finais: |
| while (ent["end"] - ent["start"]) > 0 and text[ent["end"] - 1] in SINAIS_REMOVER: |
| ent["end"] -= 1 |
| |
| tokens_doc = text.split() |
| ner_spans_coref = [] |
| char_to_token = {} |
| pos = 0 |
| for tok_idx, tok in enumerate(tokens_doc): |
| for c in range(pos, pos + len(tok)): |
| char_to_token[c] = tok_idx |
| pos += len(tok) + 1 |
| for ent in entidades_finais: |
| s_tok = char_to_token.get(ent["start"]) |
| e_tok = char_to_token.get(min(ent["end"] - 1, len(text) - 1)) |
| if s_tok is not None and e_tok is not None: |
| lbl = ent["label"].replace("PERSONAL-", "") if ent["label"] else "Other" |
| ner_spans_coref.append((s_tok, e_tok, lbl)) |
| clusters = predict_coref_clusters(model_coref, tokenizer_coref, tokens_doc, ner_spans_coref, coref_cfg) |
| |
| tok_span_to_ent_char = {} |
| for ent in entidades_finais: |
| s_tok = char_to_token.get(ent["start"]) |
| e_tok = char_to_token.get(min(ent["end"] - 1, len(text) - 1)) |
| if s_tok is not None and e_tok is not None: |
| tok_span_to_ent_char[(s_tok, e_tok)] = ent["start"] |
| char_start_to_cluster_id = {} |
| cluster_id_counter = defaultdict(int) |
| for cluster in clusters: |
| cluster_label = None |
| cluster_char_starts = [] |
| for (cs, ce) in cluster: |
| char_s = tok_span_to_ent_char.get((cs, ce)) |
| if char_s is not None: |
| cluster_char_starts.append(char_s) |
| if cluster_label is None: |
| for ent in entidades_finais: |
| if ent["start"] == char_s: |
| cluster_label = ent["label"].replace("PERSONAL-", "") if ent["label"] else "Other" |
| break |
| if not cluster_label or not cluster_char_starts: |
| continue |
| cluster_id_counter[cluster_label] += 1 |
| cid = cluster_id_counter[cluster_label] |
| for char_s in cluster_char_starts: |
| char_start_to_cluster_id[char_s] = cid |
| entity_id_counters = defaultdict(int) |
| text_to_id = defaultdict(dict) |
| entidades_finais.sort(key=lambda x: x["start"]) |
| for ent in entidades_finais: |
| tag = ent["label"].replace("PERSONAL-", "") if ent["label"] else "Other" |
| texto_ent = text[ent["start"]:ent["end"]].strip().lower() |
| if ent["start"] in char_start_to_cluster_id: |
| ent["entity_id"] = char_start_to_cluster_id[ent["start"]] |
| elif texto_ent in text_to_id[tag]: |
| ent["entity_id"] = text_to_id[tag][texto_ent] |
| else: |
| entity_id_counters[tag] += 1 |
| new_id = entity_id_counters[tag] |
| ent["entity_id"] = new_id |
| text_to_id[tag][texto_ent] = new_id |
| |
| entidades_para_substituir = sorted(entidades_finais, key=lambda x: x["start"], reverse=True) |
| texto_anon = text |
| relatorio_json = [] |
| for ent in entidades_para_substituir: |
| tag_interno = ent["label"].replace("PERSONAL-", "") if ent["label"] else "Other" |
| eid = ent.get("entity_id", "") |
| id_part = f"-{eid}" if eid else "" |
| texto_original = text[ent["start"]:ent["end"]].strip() |
| relatorio_json.append({ |
| "category": ent.get("label", "Other"), |
| "text": texto_original, |
| "start": ent["start"], |
| "end": ent["end"], |
| "id": eid, |
| "tag_interno": tag_interno, |
| }) |
| tag_display = label_name(tag_interno) |
| placeholder = f' <span class="entity-highlight"><b><{tag_display}{id_part}></b></span> ' |
| texto_anon = texto_anon[:ent["start"]] + placeholder + texto_anon[ent["end"]:] |
| relatorio_json.reverse() |
| return re.sub(r' +', ' ', texto_anon).strip(), relatorio_json |
|
|
|
|
| |
| |
| |
|
|
| @st.cache_data |
| |
| def load_example_texts(): |
| json_path = os.path.join(os.path.dirname(__file__), 'example_text.json') |
| try: |
| with open(json_path, 'r', encoding='utf-8') as f: |
| return json.load(f) |
| except Exception: |
| return { |
| "Custom Text": "", |
| "1º Portuguese Meeting Minute": "", |
| "2º Portuguese Meeting Minute": "", |
| "3º Portuguese Meeting Minute": "" |
| } |
|
|
|
|
| |
| |
| |
|
|
| |
| def main(): |
| if "lang_pt" not in st.session_state: |
| st.session_state["lang_pt"] = False |
| li = lang_idx() |
| st.markdown( |
| f'<p style="font-size:60px;font-weight:bold;color:#bdbbbb;text-align:center;margin-bottom:4px;">' |
| f'🛡️ PID: {UI["page_title"][li]}</p>', |
| unsafe_allow_html=True |
| ) |
| st.markdown( |
| f'<p style="text-align:center;color:#666;">{UI["page_subtitle"][li]}</p>', |
| unsafe_allow_html=True |
| ) |
| tokenizer_ner, model_ner, model_coref, tokenizer_coref, coref_cfg, error = load_models() |
| if error: |
| st.error(f"Erro ao carregar modelos: {error}") |
| st.stop() |
| st.sidebar.header(t("sidebar_header")) |
| st.sidebar.markdown("---") |
| st.sidebar.markdown(f"**🌐 {t('lang_toggle_label')}**") |
| col_en, col_pt = st.sidebar.columns(2) |
| with col_en: |
| if st.button( |
| "English (EN)", |
| use_container_width=True, |
| type="secondary" if st.session_state["lang_pt"] else "primary", |
| key="btn_en" |
| ): |
| st.session_state["lang_pt"] = False |
| st.rerun() |
| with col_pt: |
| if st.button( |
| "Portuguese (PT)", |
| use_container_width=True, |
| type="primary" if st.session_state["lang_pt"] else "secondary", |
| key="btn_pt" |
| ): |
| st.session_state["lang_pt"] = True |
| st.rerun() |
| st.sidebar.markdown("---") |
| example_texts = load_example_texts() |
| name_map = UI["example_names"][lang_idx()] |
| options_keys = list(example_texts.keys()) |
| options_display = [name_map.get(k, k) for k in options_keys] |
| selected_display = st.sidebar.selectbox( |
| t("choose_example"), |
| options=options_display, |
| index=0 |
| ) |
| display_to_key = {v: k for k, v in name_map.items()} |
| selected_example = display_to_key.get(selected_display, selected_display) |
| st.sidebar.markdown("<br>", unsafe_allow_html=True) |
| st.sidebar.markdown("---") |
| st.sidebar.markdown(t("about_header")) |
| st.sidebar.info(t("about_text")) |
| st.sidebar.markdown("") |
| st.sidebar.markdown(t("resources_header")) |
| st.sidebar.markdown(""" |
| - [📖 Model Card (NER)](https://huggingface.co/inesctec/CitiLink-XLMR-Anonymization-pt) |
| - [📖 Model Card (Coreference)](https://huggingface.co/inesctec/CitiLink-SpanBERT-Coreference-pt) |
| - [💾 GitHub Repository](https://github.com/LIAAD/citilink_pseudonymization) |
| """) |
| st.write("") |
| with st.expander(t("how_it_works"), expanded=False): |
| st.markdown(t("how_text")) |
| ent_cols = [ |
| ["Name", "AdministrativeInformation", "PositionDepartment", "Address", "Date"], |
| ["Location", "PersonalInformation", "Company", "ArtisticActivity"], |
| ["Degree", "Time", "LicensePlate", "Job"], |
| ["Vehicle", "Faculty", "Family", "Other"], |
| ] |
| grid_html = "<div style='display:grid;grid-template-columns:repeat(4,1fr);gap:10px;font-size:0.85em;color:white;'>" |
| for col in ent_cols: |
| grid_html += "<div>" + "".join(f"• {label_name(e)}<br>" for e in col) + "</div>" |
| grid_html += "</div>" |
| st.markdown(f""" |
| <div style="background-color:#262730;padding:20px;border-radius:10px; |
| border-left:5px solid #3b82f6;margin-bottom:25px;"> |
| <p style="color:#3b82f6;font-weight:bold;margin-top:0;margin-bottom:10px;">{t("entities_title")}</p> |
| {grid_html} |
| </div> |
| """, unsafe_allow_html=True) |
| st.markdown(f"**{t('input_label')}**") |
| st.markdown(''' |
| <div class="tab-window" style="height:auto;padding:15px;border-top:4px solid #666;margin-bottom:10px;"> |
| O interessado Dr. João Silva submeteu o processo administrativo 5597/2023 no dia 20/05/2023, |
| relativo ao imóvel localizado na Rua das Flores n.º 10 conforme o solicitado. |
| </div> |
| ''', unsafe_allow_html=True) |
| st.markdown(f"**{t('output_label')}**") |
| ex_tags = { |
| "PositionDepartment": label_name("PositionDepartment"), |
| "Name": label_name("Name"), |
| "AdministrativeInformation": label_name("AdministrativeInformation"), |
| "Date": label_name("Date"), |
| "Address": label_name("Address"), |
| } |
| st.markdown(f''' |
| <div class="tab-window" style="height:auto;padding:15px;border-top:4px solid #a8dadc;"> |
| O interessado |
| <span class="entity-highlight"><b><{ex_tags["PositionDepartment"]}-1></b></span> |
| <span class="entity-highlight"><b><{ex_tags["Name"]}-1></b></span> |
| submeteu o processo administrativo |
| <span class="entity-highlight"><b><{ex_tags["AdministrativeInformation"]}-1></b></span> |
| no dia <span class="entity-highlight"><b><{ex_tags["Date"]}-1></b></span>, |
| relativo ao imóvel localizado na |
| <span class="entity-highlight"><b><{ex_tags["Address"]}-1></b></span> conforme o solicitado. |
| </div> |
| ''', unsafe_allow_html=True) |
| st.write("") |
| col1, col2 = st.columns([1, 1]) |
| with col1: |
| st.subheader(t("col1_header")) |
| if st.session_state.get("last_example") != selected_example: |
| st.session_state["last_example"] = selected_example |
| st.session_state["result_text"] = None |
| st.session_state["result_report"] = None |
| input_text = st.text_area( |
| t("textarea_label"), |
| value=example_texts[selected_example], |
| height=400, |
| key=f"input_area_{selected_example}_{li}", |
| placeholder=t("textarea_placeholder") |
| ) |
| process_btn = st.button(t("anonymize_btn"), type="primary", use_container_width=True) |
| if process_btn and input_text: |
| with st.spinner(t("spinner_text")): |
| texto_final, relatorio = process_anonymization( |
| input_text, tokenizer_ner, model_ner, |
| model_coref, tokenizer_coref, coref_cfg |
| ) |
| st.session_state["result_text"] = texto_final |
| st.session_state["result_report"] = relatorio |
| total_entidades = len(relatorio) |
| tipos_unicos = len(set(ent["category"] for ent in relatorio)) |
| contagem_grupos = defaultdict(int) |
| for ent in relatorio: |
| if ent.get("id"): |
| contagem_grupos[(ent["category"], ent["id"])] += 1 |
| total_grupos = sum(1 for c in contagem_grupos.values() if c > 1) |
| st.markdown(f""" |
| <div class="small-metric-container"> |
| <div class="small-metric-box"> |
| <div class="metric-label">{t("metric_entities")}</div> |
| <div class="metric-value">{total_entidades}</div> |
| </div> |
| <div class="small-metric-box"> |
| <div class="metric-label">{t("metric_categories")}</div> |
| <div class="metric-value">{tipos_unicos}</div> |
| </div> |
| <div class="small-metric-box"> |
| <div class="metric-label">{t("metric_clusters")}</div> |
| <div class="metric-value">{total_grupos}</div> |
| </div> |
| </div> |
| """, unsafe_allow_html=True) |
| with col2: |
| st.subheader(t("col2_header")) |
| resultado_disponivel = ( |
| st.session_state.get("result_text") is not None |
| and st.session_state.get("result_report") is not None |
| ) |
| if resultado_disponivel: |
| texto_final = st.session_state["result_text"] |
| relatorio = st.session_state["result_report"] |
| texto_base = st.session_state.get("result_input", input_text) |
| if process_btn: |
| st.session_state["result_input"] = input_text |
| texto_base = input_text |
| texto_rebuilt = texto_base if texto_base else "" |
| for ent in sorted(relatorio, key=lambda x: x["start"], reverse=True): |
| tag_interno = ent.get("tag_interno", ent["category"].replace("PERSONAL-", "")) |
| tag_display = label_name(tag_interno) |
| eid = ent.get("id", "") |
| id_part = f"-{eid}" if eid else "" |
| placeholder = f' <span class="entity-highlight"><b><{tag_display}{id_part}></b></span> ' |
| texto_rebuilt = texto_rebuilt[:ent["start"]] + placeholder + texto_rebuilt[ent["end"]:] |
| texto_rebuilt = re.sub(r' +', ' ', texto_rebuilt).strip() |
| tab_text, tab_entities, tab_groups = st.tabs([t("tab_anon"), t("tab_entities"), t("tab_groups")]) |
| with tab_text: |
| st.markdown(f''' |
| <div class="tab-window" style="border-top:4px solid #2162bf;"> |
| {texto_rebuilt} |
| </div> |
| ''', unsafe_allow_html=True) |
| with tab_entities: |
| agrupado = {} |
| for item in relatorio: |
| tag_interno = item.get("tag_interno", item["category"].replace("PERSONAL-", "")) |
| cat_display = label_name(tag_interno) |
| if cat_display not in agrupado: |
| agrupado[cat_display] = [] |
| lbl = item["text"] |
| agrupado[cat_display].append(lbl) |
| html_content = "" |
| for cat, lista in agrupado.items(): |
| count = len(lista) |
| html_content += f"<div style='margin-bottom:20px;'>" |
| html_content += f"<b style='color:#a8dadc;font-size:1.1rem;'>{cat}</b> " |
| html_content += f"<span style='color:#666;'>({count})</span><br>" |
| for item_label in lista: |
| html_content += f"<div style='color:#ffffff;margin-left:15px;margin-top:3px;'>- {item_label}</div>" |
| html_content += "</div>" |
| st.markdown(f''' |
| <div class="tab-window" style="border-top:4px solid #2162bf;"> |
| {html_content if html_content else t("no_entities")} |
| </div> |
| ''', unsafe_allow_html=True) |
| with tab_groups: |
| grupos = defaultdict(list) |
| for item in relatorio: |
| if item.get("id"): |
| tag_int = item.get("tag_interno", item["category"].replace("PERSONAL-", "")) |
| grupos[(tag_int, item["id"])].append(item["text"]) |
| grupos_multi = {k: v for k, v in grupos.items() if len(v) > 1} |
| if not grupos_multi: |
| grupos_html = f'<div style="color:#9ca3af;padding:10px;">{t("no_groups")}</div>' |
| else: |
| por_classe = defaultdict(list) |
| for (tag_int, gid), mencoes in grupos_multi.items(): |
| por_classe[label_name(tag_int)].append((gid, mencoes)) |
| grupos_html = "" |
| for classe in sorted(por_classe.keys()): |
| clusters_classe = sorted(por_classe[classe], key=lambda x: x[0]) |
| grupos_html += f"<div style='margin-bottom:18px;'>" |
| grupos_html += f"<b style='color:#a8dadc;font-size:1.15rem;'>{classe}</b> " |
| grupos_html += f"<span style='color:#666;'>({len(clusters_classe)})</span>" |
| for gid, mencoes in clusters_classe: |
| grupos_html += ( |
| f"<div style='margin:8px 0 4px 12px;padding:8px 12px;" |
| f"background-color:#1e212b;border-left:3px solid #3b82f6;border-radius:6px;'>" |
| ) |
| grupos_html += ( |
| f"<div style='color:#60a5fa;font-weight:bold;margin-bottom:6px;'>" |
| f"<{classe}-{gid}> " |
| f"<span style='color:#6c757d;font-weight:normal;font-size:0.8rem;'>" |
| f"· {len(mencoes)} {t('mentions_label')}</span></div>" |
| ) |
| for m in mencoes: |
| grupos_html += ( |
| f"<div style='color:#ffffff;margin-left:10px;margin-top:2px;font-size:0.9rem;'>" |
| f"↳ {m}</div>" |
| ) |
| grupos_html += "</div>" |
| grupos_html += "</div>" |
| st.markdown(f''' |
| <div class="tab-window" style="border-top:4px solid #2162bf;"> |
| {grupos_html} |
| </div> |
| ''', unsafe_allow_html=True) |
| with st.expander(t("json_expander")): |
| download_data = { |
| "full_text": texto_base, |
| "personal_info": relatorio |
| } |
| json_string = json.dumps(download_data, indent=4, ensure_ascii=False) |
| st.download_button( |
| label=t("download_btn"), |
| data=json_string, |
| file_name="anonymization_report.json", |
| mime="application/json", |
| use_container_width=True |
| ) |
| st.json(relatorio) |
| else: |
| st.markdown('<div style="margin-top:30px;"></div>', unsafe_allow_html=True) |
| st.info(t("no_results_info")) |
| st.markdown("---") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
