import gradio as gr import pandas as pd import numpy as np from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.cluster import KMeans, AgglomerativeClustering from sklearn.decomposition import TruncatedSVD from sklearn.manifold import TSNE import plotly.graph_objects as go import tempfile import os def run_clustering(file_obj, raw_text_area, num_clusters, algorithm, reduction_method, data_type): if data_type == "Paste Raw Text (One Document Per Line)": if not raw_text_area or len(raw_text_area.strip()) < 10: return "Please enter multiple lines of text to cluster.", None, None, None documents = [line.strip() for line in raw_text_area.split('\n') if line.strip()] if len(documents) < num_clusters: return f"Number of documents ({len(documents)}) must be greater than or equal to number of clusters ({num_clusters}).", None, None, None df = pd.DataFrame({"Document": documents}) text_col = "Document" else: if file_obj is None: return "Please upload a CSV or Excel file.", None, None, None try: if file_obj.name.endswith('.csv'): df = pd.read_csv(file_obj.name) else: df = pd.read_excel(file_obj.name) except Exception as e: return f"Error reading file: {str(e)}", None, None, None # Find text column text_col = None for col in df.columns: if col.lower() in ['text', 'document', 'content', 'body', 'sentence']: text_col = col break if not text_col: # Fallback to first string/object column string_cols = df.select_dtypes(include=['object']).columns if len(string_cols) > 0: text_col = string_cols[0] else: return "Could not find any text column in the uploaded dataset. Ensure it has a column with textual content.", None, None, None df = df.dropna(subset=[text_col]) documents = df[text_col].astype(str).tolist() if len(documents) < num_clusters: return f"Number of documents ({len(documents)}) must be greater than or equal to number of clusters ({num_clusters}).", None, None, None # 1. TF-IDF Embedding try: vectorizer = TfidfVectorizer(stop_words='english', max_features=1000) X = vectorizer.fit_transform(documents).toarray() except Exception as e: return f"Error building TF-IDF vectors: {str(e)}. Try using longer or more diverse texts.", None, None, None # 2. Cluster Allocation if algorithm == "K-Means": clusterer = KMeans(n_clusters=num_clusters, random_state=42, n_init=10) labels = clusterer.fit_predict(X) else: # Hierarchical Agglomerative clusterer = AgglomerativeClustering(n_clusters=num_clusters) labels = clusterer.fit_predict(X) df['Cluster'] = labels + 1 # 3. Dimensions Reduction (2D Projection) if reduction_method == "PCA (SVD)": reducer = TruncatedSVD(n_components=2, random_state=42) X_2d = reducer.fit_transform(X) else: # t-SNE # t-SNE perplexity must be smaller than the number of samples perp = min(30, max(2, len(documents) // 2)) reducer = TSNE(n_components=2, perplexity=perp, random_state=42, init='random') X_2d = reducer.fit_transform(X) df['x'] = X_2d[:, 0] df['y'] = X_2d[:, 1] # Shorten texts for hover label hover_texts = [d[:75] + "..." if len(d) > 75 else d for d in documents] df['HoverText'] = hover_texts # 4. Generate Interactive Plotly Chart fig = go.Figure() # Premium color palette for clusters colors = ['#ff7043', '#4db6ac', '#9575cd', '#ffd54f', '#64b5f6', '#f06292', '#81c784', '#ffffff', '#a1887f', '#ba68c8'] for c_id in sorted(df['Cluster'].unique()): sub_df = df[df['Cluster'] == c_id] color = colors[(c_id - 1) % len(colors)] fig.add_trace(go.Scatter( x=sub_df['x'], y=sub_df['y'], mode='markers', marker=dict(size=12, color=color, line=dict(width=1, color='#16100c')), name=f"Cluster {c_id}", text=sub_df['HoverText'], hoverinfo='text+name' )) fig.update_layout( title="Document Cluster Projection", paper_bgcolor='#16100c', plot_bgcolor='#16100c', font_color='#f4eee6', xaxis=dict(showgrid=True, gridcolor='rgba(255,255,255,0.05)', showticklabels=False), yaxis=dict(showgrid=True, gridcolor='rgba(255,255,255,0.05)', showticklabels=False), margin=dict(l=40, r=40, t=50, b=40) ) # 5. Extract Cluster Keywords (Top terms inside each cluster) cluster_keywords = [] feature_names = np.array(vectorizer.get_feature_names_out()) for c_id in sorted(df['Cluster'].unique()): sub_idx = np.where(labels == (c_id - 1))[0] sub_X = X[sub_idx] mean_tfidf = sub_X.mean(axis=0) top_indices = mean_tfidf.argsort()[::-1][:6] top_words = ", ".join(feature_names[top_indices]) cluster_keywords.append({ "Cluster ID": c_id, "Document Count": len(sub_idx), "Representative Keywords": top_words }) df_keywords = pd.DataFrame(cluster_keywords) # Output file out_csv = tempfile.mktemp(suffix=".csv") df.drop(columns=['x', 'y', 'HoverText'], errors='ignore').to_csv(out_csv, index=False) # Build clean previews dataframe df_preview = df[[text_col, 'Cluster']].head(100) return "", fig, df_keywords, df_preview, out_csv theme = gr.themes.Default( primary_hue="orange", neutral_hue="stone" ).set( body_background_fill="#0d0907", body_text_color="#c4bbae", block_background_fill="#16100c", block_border_width="1px", block_label_text_color="#f4eee6" ) with gr.Blocks(theme=theme, title="Clustering Analyzer") as demo: gr.Markdown( """ # 🧩 Document & Artifact Clustering Analyzer ### Group similar documents, articles, or textual transcripts automatically using unsupervised machine learning. Visualize groupings in 2D vector space instantly. """ ) error_msg = gr.Markdown("", visible=False) with gr.Row(): with gr.Column(scale=1): data_type = gr.Radio( choices=["Upload Dataset Sheet", "Paste Raw Text (One Document Per Line)"], value="Paste Raw Text (One Document Per Line)", label="Data Input Mode" ) with gr.Group(visible=False) as upload_group: file_obj = gr.File(label="Upload CSV or Excel sheet", file_types=[".csv", ".xlsx"]) gr.Markdown("💡 **Tip**: Make sure your dataset contains a textual column (e.g., **Text**, **Content**).") with gr.Group(visible=True) as text_group: raw_text_area = gr.Textbox( label="Input Text Documents (one per line)", placeholder="Romeo loves Juliet\nShakespeare wrote plays\nVerification is important in coding\nNetworks model nodes and edges\nAI assistants write python scripts", lines=10 ) with gr.Row(): num_clusters = gr.Slider(minimum=2, maximum=10, value=3, step=1, label="Number of Clusters (k)") algorithm = gr.Dropdown(choices=["K-Means", "Hierarchical Agglomerative"], value="K-Means", label="Cluster Algorithm") reduction_method = gr.Radio(choices=["PCA (SVD)", "t-SNE"], value="PCA (SVD)", label="Vector Dimensionality Reduction") btn = gr.Button("Calculate and Visualize Clusters", variant="primary") with gr.Column(scale=2): with gr.Tabs(): with gr.TabItem("Interactive 2D Cluster Map"): plot_box = gr.Plot() with gr.TabItem("Representative Cluster Keywords"): table_keywords = gr.Dataframe(headers=["Cluster ID", "Document Count", "Representative Keywords"]) with gr.TabItem("Labeled Documents Table"): table_docs = gr.Dataframe(max_rows=15) download_btn = gr.File(label="Download Complete Labeled CSV") def update_visibility(mode): if mode == "Upload Dataset Sheet": return gr.update(visible=True), gr.update(visible=False) else: return gr.update(visible=False), gr.update(visible=True) data_type.change( update_visibility, inputs=[data_type], outputs=[upload_group, text_group] ) def process(file_obj, text_area, clusters, algo, reduction, mode): err, plot, keywords, docs, csv_path = run_clustering(file_obj, text_area, clusters, algo, reduction, mode) if err: return gr.update(value=err, visible=True), None, None, None, gr.update(visible=False) return gr.update(visible=False), plot, keywords, docs, gr.update(value=csv_path, visible=True) btn.click( process, inputs=[file_obj, raw_text_area, num_clusters, algorithm, reduction_method, data_type], outputs=[error_msg, plot_box, table_keywords, table_docs, download_btn] ) if __name__ == "__main__": demo.launch()