app.py

# ==============================================================================
#  داشبورد_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'<span style="background-color: #334155; color: white; border-radius: 5px; padding: 5px 10px; margin: 3px; display: inline-block;">{p}</span>' for p in metricas['palavras_relevantes']])
        st.markdown(f'<div>{tags_html}</div>', 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... ✨")