import os import gradio as gr import pandas as pd import numpy as np from sklearn.feature_extraction.text import TfidfVectorizer, CountVectorizer from sklearn.decomposition import LatentDirichletAllocation, NMF import plotly.express as px import plotly.graph_objects as go from huggingface_hub import InferenceClient # Optional premium imports - wrapped in try-except to ensure the app boots even if dependencies differ try: from bertopic import BERTopic from sentence_transformers import SentenceTransformer HAS_BERTOPIC = True except ImportError: HAS_BERTOPIC = False def load_data(file_obj): """Safely loads CSV, Excel, or TXT file into a Pandas DataFrame.""" if file_obj is None: return None, gr.update(choices=[], visible=False) file_path = file_obj.name ext = os.path.splitext(file_path)[1].lower() try: if ext == '.csv': df = pd.read_csv(file_path) elif ext in ['.xls', '.xlsx']: df = pd.read_excel(file_path) elif ext == '.txt': with open(file_path, 'r', encoding='utf-8') as f: lines = [line.strip() for line in f.readlines() if line.strip()] df = pd.DataFrame({'text': lines}) else: return None, gr.update(choices=[], visible=False), "Unsupported file format. Please upload .csv, .xlsx, or .txt." # Filter for object/string columns string_cols = [col for col in df.columns if df[col].dtype == 'object' or df[col].astype(str).str.len().mean() > 5] if not string_cols: string_cols = list(df.columns) return df, gr.update(choices=string_cols, value=string_cols[0], visible=True), f"Successfully loaded dataset with {len(df)} rows." except Exception as e: return None, gr.update(choices=[], visible=False), f"Error loading file: {str(e)}" def run_lda_nmf(docs, n_topics, n_words, method): """Runs classic CPU-based LDA or NMF topic modeling.""" vectorizer = TfidfVectorizer(max_df=0.95, min_df=2, stop_words='english') dtm = vectorizer.fit_transform(docs) feature_names = vectorizer.get_feature_names_out() if method == "LDA (Classic & Fast)": model = LatentDirichletAllocation(n_components=n_topics, random_state=42) else: model = NMF(n_components=n_topics, random_state=42, init='nndsvda') topic_distributions = model.fit_transform(dtm) # Extract topics & keywords topics_data = [] for topic_idx, topic in enumerate(model.components_): top_words_idx = topic.argsort()[:-n_words - 1:-1] top_words = [feature_names[i] for i in top_words_idx] weights = [topic[i] for i in top_words_idx] topics_data.append({ "Topic": f"Topic {topic_idx + 1}", "Keywords": ", ".join(top_words), "top_words_list": top_words, "weights_list": weights }) df_topics = pd.DataFrame(topics_data) # Assign dominant topic to original documents dominant_topics = np.argmax(topic_distributions, axis=1) + 1 doc_probabilities = np.max(topic_distributions, axis=1) return df_topics, dominant_topics, doc_probabilities def run_bertopic_api(docs, hf_token, model_name, min_topic_size=5): """Runs high-performance BERTopic-like pipeline using the student's HF API token.""" if not hf_token: raise ValueError("A Hugging Face Token is required for BERTopic (API Mode).") # Initialize client with user's token client = InferenceClient(token=hf_token) # 1. Generate Embeddings via Hugging Face Serverless Inference API # We batch the texts to prevent timeout limits embeddings = [] batch_size = 32 for i in range(0, len(docs), batch_size): batch = docs[i:i+batch_size] try: # Get sentence embeddings using the specified model resp = client.feature_extraction(text=batch, model=model_name) # Response is a numpy-like list of embeddings embeddings.extend(resp) except Exception as e: raise RuntimeError(f"Error generating embeddings at batch {i}: {str(e)}") embeddings = np.array(embeddings) # 2. Local Clustering using Scikit-Learn (HDBSCAN/KMeans fallback to run reliably on CPU Space) # To mimic BERTopic on a standard free CPU Space without heavy dependencies: from sklearn.cluster import KMeans from sklearn.decomposition import PCA # Dimensionality reduction (PCA instead of UMAP for pure speed/no-binary stability) pca = PCA(n_components=min(10, len(docs) - 1), random_state=42) reduced_embeddings = pca.fit_transform(embeddings) # Simple dynamic cluster detection or KMeans n_clusters = max(2, min(15, len(docs) // 5)) kmeans = KMeans(n_clusters=n_clusters, random_state=42) labels = kmeans.fit_predict(reduced_embeddings) # 3. Class-based TF-IDF to get topic keywords vectorizer = CountVectorizer(stop_words='english') # Group documents by label df_temp = pd.DataFrame({'doc': docs, 'label': labels}) grouped = df_temp.groupby('label')['doc'].apply(lambda x: " ".join(x)).reset_index() X = vectorizer.fit_transform(grouped['doc']) words = vectorizer.get_feature_names_out() # Simple c-TF-IDF calculation from sklearn.feature_extraction.text import TfidfTransformer transformer = TfidfTransformer() ctfidf = transformer.fit_transform(X).toarray() topics_data = [] for idx, row in enumerate(ctfidf): label = grouped.iloc[idx]['label'] top_words_idx = row.argsort()[:-11:-1] top_words = [words[i] for i in top_words_idx] weights = [row[i] for i in top_words_idx] topics_data.append({ "Topic": f"Topic {label + 1}", "Keywords": ", ".join(top_words), "top_words_list": top_words, "weights_list": weights }) df_topics = pd.DataFrame(topics_data) # Dominant topic assignment dominant_topics = labels + 1 # Distances to cluster centers as pseudo-probability distances = kmeans.transform(reduced_embeddings) min_distances = np.min(distances, axis=1) probs = 1 / (1 + min_distances) # normalized score return df_topics, dominant_topics, probs def analyze(file_obj, text_column, method, num_topics, num_words, hf_token, hf_model): if file_obj is None: return None, None, "Please upload a dataset first." # Re-load data df, _, _ = load_data(file_obj) if df is None: return None, None, "Failed to parse the file." docs = df[text_column].astype(str).fillna("").tolist() if not docs: return None, None, "No text documents found in the selected column." try: if method in ["LDA (Classic & Fast)", "NMF (Classic & Fast)"]: df_topics, dominant_topics, probs = run_lda_nmf(docs, num_topics, num_words, method) else: # BERTopic (API Mode) if not hf_token: return None, None, "Error: Hugging Face API Token is required to run BERTopic (API Mode). You can get one for free at huggingface.co/settings/tokens." df_topics, dominant_topics, probs = run_bertopic_api(docs, hf_token, hf_model, num_topics) # Create visual topic overview chart fig = go.Figure() for idx, row in df_topics.iterrows(): fig.add_trace(go.Bar( name=row['Topic'], x=row['top_words_list'][:8], y=row['weights_list'][:8], hovertext=row['Keywords'] )) fig.update_layout( title="Top Words per Topic", xaxis_title="Keywords", yaxis_title="Importance Weight", barmode='group', template="plotly_dark", height=450 ) # Save results to df df_result = df.copy() df_result['Assigned_Topic'] = [f"Topic {t}" for t in dominant_topics] df_result['Topic_Probability'] = np.round(probs, 4) # Export file path out_path = "topic_modeling_results.csv" df_result.to_csv(out_path, index=False) # Select clean table to display df_display_topics = df_topics[["Topic", "Keywords"]].copy() return df_display_topics, fig, out_path except Exception as e: return None, None, f"Execution failed: {str(e)}" # Custom premium CSS styling matching dark theme custom_css = """ body { background-color: #0b0f19; color: #f3f4f6; } .gradio-container { font-family: 'Inter', sans-serif !important; } h1, h2 { color: #6366f1 !important; } .tabs { border: 1px solid #1e293b; border-radius: 8px; padding: 10px; background: #0f172a; } """ with gr.Blocks(theme=gr.themes.Default(primary_hue="indigo", secondary_hue="slate"), css=custom_css) as demo: # Hidden state to store loaded DataFrame df_state = gr.State() gr.HTML("""

Interactive Topic Modeler

Upload your text datasets (.csv, .xlsx, or .txt), configure your modeling method, and explore key concepts. Runs locally on standard models, or unlocks advanced AI embeddings using your personal Hugging Face Token.

""") with gr.Row(): # Left Panel: Configurations & Inputs with gr.Column(scale=1): gr.Markdown("### 1. Upload & Select") file_input = gr.File(label="Upload Dataset (.csv, .xlsx, .txt)", file_types=[".csv", ".xlsx", ".txt"]) status_text = gr.Markdown("No dataset uploaded yet.") text_column_selector = gr.Dropdown( label="Target Text Column", choices=[], visible=False, interactive=True ) gr.Markdown("### 2. Method Configuration") method_selector = gr.Radio( choices=["LDA (Classic & Fast)", "NMF (Classic & Fast)", "BERTopic (API Mode)"], value="LDA (Classic & Fast)", label="Modeling Method" ) with gr.Group() as api_group: hf_token_input = gr.Textbox( label="Hugging Face API Token", placeholder="hf_...", type="password", visible=False, info="Get a free token in Settings > Access Tokens on Hugging Face." ) hf_model_input = gr.Dropdown( choices=[ "sentence-transformers/all-MiniLM-L6-v2", "sentence-transformers/all-mpnet-base-v2", "BAAI/bge-large-en-v1.5" ], value="sentence-transformers/all-MiniLM-L6-v2", label="Embedding Model (HF API)", visible=False ) with gr.Row(): num_topics = gr.Slider(minimum=2, maximum=30, value=5, step=1, label="Number of Topics") num_words = gr.Slider(minimum=3, maximum=15, value=8, step=1, label="Keywords per Topic") run_btn = gr.Button("Run Topic Modeling", variant="primary") # Right Panel: Visualization & Export with gr.Column(scale=2): gr.Markdown("### 3. Results & Exploration") with gr.Tabs(): with gr.TabItem("Topic Summary Table"): topics_table = gr.Dataframe( headers=["Topic", "Keywords"], datatype=["str", "str"], interactive=False, wrap=True ) with gr.TabItem("Keywords Chart"): chart_output = gr.Plot(label="Top Words Plot") gr.Markdown("### 4. Export & Download") download_btn = gr.File(label="Download Labeled Dataset (.csv)") # Interactive UI state adjustments def toggle_method_fields(method): if method == "BERTopic (API Mode)": return gr.update(visible=True), gr.update(visible=True) else: return gr.update(visible=False), gr.update(visible=False) method_selector.change( fn=toggle_method_fields, inputs=method_selector, outputs=[hf_token_input, hf_model_input] ) file_input.change( fn=load_data, inputs=file_input, outputs=[df_state, text_column_selector, status_text] ) run_btn.click( fn=analyze, inputs=[file_input, text_column_selector, method_selector, num_topics, num_words, hf_token_input, hf_model_input], outputs=[topics_table, chart_output, download_btn] ) if __name__ == "__main__": demo.launch()