# ==============================================================================
# داشبورد_3d_magic_streamlit.py (VERSÃO PARA HUGGING FACE SPACES)
# Arquivo: app.py
# Versão com correção de tipo no retorno de duplicados
# ==============================================================================
import streamlit as st
import numpy as np
import pandas as pd
from sentence_transformers import SentenceTransformer
import umap
import plotly.express as px
import hdbscan
from sklearn.preprocessing import StandardScaler
from sklearn.metrics.pairwise import cosine_similarity
from sklearn.feature_extraction.text import CountVectorizer, TfidfVectorizer
from scipy.stats import entropy
import torch
import gc
# ================================
# CONFIGURAÇÕES DA PÁGINA E DO MODELO
# ================================
st.set_page_config(
page_title="Dashboard 3D Mágico",
page_icon="🌌",
layout="wide",
initial_sidebar_state="expanded"
)
# Constantes e Stop Words
DEFAULT_MODEL = 'all-MiniLM-L6-v2'
BATCH_SIZE = 256
UMAP_N_NEIGHBORS = 30
HDBSCAN_MIN_SIZE = 50
MAX_DUPLICATES_TO_SHOW = 50
STOP_WORDS_PT = ['de', 'a', 'o', 'que', 'e', 'do', 'da', 'em', 'um', 'para', 'é', 'com', 'não', 'uma', 'os', 'no', 'se', 'na', 'por', 'mais', 'as', 'dos', 'como', 'mas', 'foi', 'ao', 'ele', 'das', 'tem', 'à', 'seu', 'sua', 'ou', 'ser', 'quando', 'muito', 'há', 'nos', 'já', 'está', 'eu', 'também', 'só', 'pelo', 'pela', 'até', 'isso', 'ela', 'entre', 'era', 'depois', 'sem', 'mesmo', 'aos', 'ter', 'seus', 'quem', 'nas', 'me', 'esse', 'eles', 'estão', 'você', 'tinha', 'foram', 'essa', 'num', 'nem', 'suas', 'meu', 'às', 'minha', 'numa', 'pelos', 'elas', 'havia', 'seja', 'qual', 'será', 'nós', 'tenho', 'lhe', 'deles', 'essas', 'esses', 'pelas', 'este', 'fosse', 'dele', 'tu', 'te', 'vocês', 'vos', 'lhes', 'meus', 'minhas', 'teu', 'tua', 'teus', 'tuas', 'nosso', 'nossa', 'nossos', 'nossas', 'dela', 'delas', 'esta', 'estes', 'estas', 'aquele', 'aquela', 'aqueles', 'aquelas', 'isto', 'aquilo', 'estou', 'está', 'estamos', 'estão', 'estive', 'esteve', 'estivemos', 'estiveram', 'estava', 'estávamos', 'estavam', 'estivera', 'estivéramos', 'esteja', 'estejamos', 'estejam', 'estivesse', 'estivéssemos', 'estivessem', 'estiver', 'estivermos', 'estiverem', 'hei', 'há', 'havemos', 'hão', 'houve', 'houvemos', 'houveram', 'houvera', 'houvéramos', 'haja', 'hajamos', 'hajam', 'houvesse', 'houvéssemos', 'houvessem', 'houver', 'houvermos', 'houverem', 'houverei', 'houverá', 'houveremos', 'houverão', 'houveria', 'houveríamos', 'houveriam', 'sou', 'somos', 'são', 'era', 'éramos', 'eram', 'fui', 'foi', 'fomos', 'foram', 'fora', 'fôramos', 'seja', 'sejamos', 'sejam', 'fosse', 'fôssemos', 'fossem', 'for', 'formos', 'forem', 'serei', 'será', 'seremos', 'serão', 'seria', 'seríamos', 'seriam', 'tenho', 'tem', 'temos', 'tém', 'tinha', 'tínhamos', 'tinham', 'tive', 'teve', 'tivemos', 'tiveram', 'tivera', 'tivéramos', 'tenha', 'tenhamos', 'tenham', 'tivesse', 'tivéssemos', 'tivessem', 'tiver', 'tivermos', 'tiverem', 'terei', 'terá', 'teremos', 'terão', 'teria', 'teríamos', 'teriam', 'dá', 'pergunta', 'resposta']
# ================================
# FUNÇÕES CACHEADAS (O CORAÇÃO DA PERFORMANCE)
# ================================
@st.cache_resource
def load_model():
device = "cuda" if torch.cuda.is_available() else "cpu"
print(f"Carregando modelo para o dispositivo: {device}")
model = SentenceTransformer(DEFAULT_MODEL, device=device)
return model
@st.cache_data
def process_data(file_name, file_bytes, n_samples):
with st.spinner('Lendo e validando o arquivo...'):
lines = file_bytes.decode('utf-8').splitlines()
_texts = [s for line in lines if (s := line.strip()) and len(s.split()) > 3][:n_samples]
if not _texts: return None, None
model = load_model()
st.info(f"Processando {len(_texts):,} textos...")
progress_bar = st.progress(0, text="Gerando embeddings...")
embeddings = model.encode(_texts, batch_size=BATCH_SIZE, show_progress_bar=False, convert_to_numpy=True)
progress_bar.progress(30, text="Reduzindo dimensionalidade com UMAP 3D...")
reducer = umap.UMAP(n_components=3, n_neighbors=UMAP_N_NEIGHBORS, min_dist=0.0, metric='cosine', random_state=42)
embedding_3d = reducer.fit_transform(embeddings)
embedding_3d = StandardScaler().fit_transform(embedding_3d)
progress_bar.progress(70, text="Clusterizando com HDBSCAN...")
clusterer = hdbscan.HDBSCAN(min_cluster_size=HDBSCAN_MIN_SIZE)
clusters = clusterer.fit_predict(embedding_3d)
progress_bar.progress(90, text="Montando o DataFrame final...")
df = pd.DataFrame({'x': embedding_3d[:, 0], 'y': embedding_3d[:, 1], 'z': embedding_3d[:, 2], 'text_hover': [t[:300] + "..." if len(t) > 300 else t for t in _texts], 'full_text': _texts, 'length': [len(t) for t in _texts], 'cluster': clusters.astype(str)})
df['color'] = df['cluster'].astype(int)
df['size'] = np.log1p(df['length']) * 1.5 + 1
progress_bar.progress(100, text="Processamento concluído!")
progress_bar.empty()
del reducer, clusterer, embedding_3d
gc.collect()
return df, embeddings
@st.cache_data
def calcular_metricas_globais(_texts):
st.info("Calculando métricas quantitativas do corpus...")
try:
vectorizer_count = CountVectorizer(stop_words=STOP_WORDS_PT, max_features=20000).fit(_texts)
riqueza_lexical = len(vectorizer_count.get_feature_names_out())
except ValueError: riqueza_lexical = 0
try:
vectorizer_tfidf = TfidfVectorizer(stop_words=STOP_WORDS_PT, max_features=20000).fit(_texts)
tfidf_matrix, vocab = vectorizer_tfidf.transform(_texts), vectorizer_tfidf.get_feature_names_out()
soma_tfidf, indices_top_tfidf = tfidf_matrix.sum(axis=0).A1, np.argsort(tfidf_matrix.sum(axis=0).A1)[-10:][::-1]
palavras_relevantes = [vocab[i] for i in indices_top_tfidf]
except ValueError: palavras_relevantes = []
try:
contagens_palavras = np.array(vectorizer_count.transform(_texts).sum(axis=0)).flatten()
entropia_corpus = entropy(contagens_palavras / np.sum(contagens_palavras), base=2)
except (ValueError, ZeroDivisionError): entropia_corpus = 0
return {"riqueza_lexical": riqueza_lexical, "palavras_relevantes": palavras_relevantes, "entropia": entropia_corpus}
@st.cache_data
def encontrar_duplicados(_df, _embeddings, similaridade_minima=0.98):
st.info("Iniciando a caça aos duplicados...")
duplicados_exatos_mask = _df['full_text'].duplicated(keep=False)
df_duplicados_exatos = _df[duplicados_exatos_mask].copy()
# <<< CORREÇÃO DE TIPO >>>
grupos_exatos = {}
if not df_duplicados_exatos.empty:
grupos_exatos = df_duplicados_exatos.groupby('full_text').groups
pares_semanticos = []
limite_semantico = 5000
if len(_embeddings) < limite_semantico:
sim_matrix = cosine_similarity(_embeddings)
indices_superiores = np.triu_indices_from(sim_matrix, k=1)
pares_altamente_similares = sim_matrix[indices_superiores] > similaridade_minima
indices_pares = np.where(pares_altamente_similares)[0]
for i in indices_pares:
idx1, idx2 = indices_superiores[0][i], indices_superiores[1][i]
if _df['full_text'].iloc[idx1] != _df['full_text'].iloc[idx2]:
pares_semanticos.append({'doc1_idx': idx1, 'doc2_idx': idx2, 'similaridade': sim_matrix[idx1, idx2], 'texto1': _df['full_text'].iloc[idx1], 'texto2': _df['full_text'].iloc[idx2]})
else:
st.warning(f"A busca por duplicados semânticos foi pulada (dataset > {limite_semantico} docs).")
return grupos_exatos, pares_semanticos
# ================================
# LAYOUT DA APLICAÇÃO (A INTERFACE)
# ================================
st.title("🌌 Dashboard 3D Mágico de Textos")
st.markdown("Bem-vindo, meu príncipe Gabriel Yogi! Esta ferramenta permite a exploração visual, análise quantitativa e curadoria de dados textuais.")
with st.sidebar:
st.header("⚙️ Controles")
uploaded_file = st.file_uploader("1. Escolha seu arquivo .txt", type="txt")
if uploaded_file:
n_samples = st.slider("2. Nº máximo de amostras", 500, 50000, 10000, 500, help="Valores altos exigem mais processamento.")
process_button = st.button("✨ Gerar Dashboard ✨", disabled=not uploaded_file, type="primary")
if 'processed' not in st.session_state: st.session_state.processed = False
if process_button:
if uploaded_file is not None:
file_bytes = uploaded_file.getvalue()
df, embeddings = process_data(uploaded_file.name, file_bytes, n_samples)
if df is not None:
st.session_state.df, st.session_state.embeddings = df, embeddings
st.session_state.processed = True
for key in ['metricas', 'duplicados_calculados', 'grupos_exatos', 'pares_semanticos']:
if key in st.session_state: del st.session_state[key]
else:
st.error("Nenhum texto válido encontrado no arquivo após a filtragem!")
st.session_state.processed = False
else:
st.warning("Por favor, faça o upload de um arquivo primeiro.")
if st.session_state.processed:
st.header("🔎 Explore seu Universo de Textos")
with st.expander("📊 Análise Quantitativa do Corpus", expanded=True):
if 'metricas' not in st.session_state:
st.session_state.metricas = calcular_metricas_globais(st.session_state.df['full_text'].tolist())
df_atual, metricas = st.session_state.df, st.session_state.metricas
col1, col2, col3, col4 = st.columns(4)
col1.metric("Nº de Documentos", f"{len(df_atual):,}")
n_clusters = len(df_atual['cluster'].unique()) - (1 if '-1' in df_atual['cluster'].unique() else 0)
col2.metric("Nº de Clusters", f"{n_clusters}")
n_ruido = (df_atual['cluster'] == '-1').sum()
col3.metric("Pontos de Ruído", f"{n_ruido:,}")
col4.metric("Riqueza Lexical", f"{metricas['riqueza_lexical']:,}")
st.metric("Entropia do Vocabulário (Bits)", f"{metricas['entropia']:.2f}")
st.markdown("**Top 10 Palavras Mais Relevantes (TF-IDF):**")
tags_html = "".join([f'{p}' for p in metricas['palavras_relevantes']])
st.markdown(f'{tags_html}
', unsafe_allow_html=True)
with st.expander("🕵️♀️ Análise de Duplicidade"):
if 'duplicados_calculados' not in st.session_state:
st.session_state.grupos_exatos, st.session_state.pares_semanticos = encontrar_duplicados(st.session_state.df, st.session_state.embeddings)
st.session_state.duplicados_calculados = True
grupos_exatos, pares_semanticos = st.session_state.grupos_exatos, st.session_state.pares_semanticos
st.subheader("Duplicados Exatos")
num_grupos_exatos, num_docs_dup_exatos = len(grupos_exatos), sum(len(indices) for indices in grupos_exatos.values())
st.metric("Total de Documentos em Grupos Duplicados", f"{num_docs_dup_exatos} (em {num_grupos_exatos} grupos)")
if num_grupos_exatos > 0:
if st.checkbox("Mostrar detalhes dos duplicados exatos"):
if num_grupos_exatos > MAX_DUPLICATES_TO_SHOW:
st.info(f"Encontrados {num_grupos_exatos} grupos. Mostrando os primeiros {MAX_DUPLICATES_TO_SHOW} para evitar lentidão.")
for i, (texto, indices) in enumerate(list(grupos_exatos.items())[:MAX_DUPLICATES_TO_SHOW]):
with st.container(border=True):
st.markdown(f"**Grupo {i+1} ({len(indices)} ocorrências):**"); st.caption(texto)
else:
st.success("Nenhum duplicado exato encontrado!")
st.divider()
st.subheader("Duplicados Semânticos (Quase-Idênticos)")
num_pares_semanticos = len(pares_semanticos)
st.metric(f"Total de Pares Quase-Idênticos Encontrados", f"{num_pares_semanticos}")
if num_pares_semanticos > 0:
if st.checkbox("Mostrar detalhes dos duplicados semânticos"):
if num_pares_semanticos > MAX_DUPLICATES_TO_SHOW:
st.info(f"Encontrados {num_pares_semanticos} pares. Mostrando os primeiros {MAX_DUPLICATES_TO_SHOW}.")
for par in pares_semanticos[:MAX_DUPLICATES_TO_SHOW]:
with st.container(border=True):
st.markdown(f"**Similaridade: `{par['similaridade']:.4f}`**")
c1, c2 = st.columns(2)
c1.caption(f"Doc (índice {par['doc1_idx']})"); c1.markdown(f"> {par['texto1']}")
c2.caption(f"Doc (índice {par['doc2_idx']})"); c2.markdown(f"> {par['texto2']}")
elif num_pares_semanticos == 0 and len(st.session_state.embeddings) < 5000:
st.success("Nenhum duplicado semântico encontrado acima do limiar!")
query = st.text_input("Busca Semântica", placeholder="Digite algo para destacar no gráfico...")
df_display = st.session_state.df.copy()
if query:
model = load_model()
q_embedding = model.encode([query])
sims = cosine_similarity(q_embedding, st.session_state.embeddings)[0]
idx = np.argsort(sims)[-50:][::-1]
df_display['highlight'] = '0'
df_display.loc[idx, 'highlight'] = '1'
df_display['display_size'] = np.where(df_display['highlight'] == '1', 10, df_display['size'] * 0.6)
fig = px.scatter_3d(df_display, x='x', y='y', z='z', color='highlight', size='display_size', hover_data=['text_hover', 'cluster'], title=f"Busca por: '{query}'", color_discrete_map={'0': 'grey', '1': 'yellow'}, labels={'highlight': 'Resultado'})
st.subheader("Top 5 Resultados da Busca:")
for i, (r, s) in enumerate(zip(idx[:5], sims[idx[:5]])):
st.markdown(f"**{i+1}.** *Sim: {s:.3f}* \n> {st.session_state.df['full_text'].iloc[r][:350]}...")
else:
fig = px.scatter_3d(df_display, x='x', y='y', z='z', color='cluster', size='size', hover_data=['text_hover', 'cluster'], title=f"Visualização 3D de {len(df_display):,} Textos", color_continuous_scale='Turbo')
fig.update_layout(template='plotly_dark', height=800, margin=dict(l=0, r=0, b=0, t=40), scene_camera=dict(eye=dict(x=1.5, y=1.5, z=1.5)))
st.plotly_chart(fig, use_container_width=True)
else:
st.info("Aguardando um arquivo para começar a mágica... ✨")