Spaces:
Sleeping
Sleeping
| import os | |
| import gradio as gr | |
| import pandas as pd | |
| import numpy as np | |
| import re | |
| import plotly.express as px | |
| from collections import Counter | |
| 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), "Please upload a file." | |
| 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: | |
| content = f.read() | |
| df = pd.DataFrame({'text': [content]}) | |
| else: | |
| return None, gr.update(choices=[], visible=False), "Unsupported file format. Please upload .csv, .xlsx, or .txt." | |
| 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 calculate_collocations(docs, n_gram_type, metric, min_freq, top_n): | |
| """Calculates n-gram collocations using Raw Frequency or PMI.""" | |
| # Combine texts and tokenize | |
| words = [] | |
| stopwords = {'the', 'and', 'for', 'that', 'with', 'this', 'have', 'from', 'your', 'will', 'not', 'are', 'was', 'were', 'but', 'how', 'they', 'our', 'them', 'their', 'she', 'him', 'her', 'his', 'has', 'had', 'been', 'would', 'could', 'should'} | |
| for doc in docs: | |
| cleaned = re.sub(r'[^\w\s]', ' ', doc.lower()) | |
| doc_words = [w.strip() for w in cleaned.split() if w.strip() and w.strip() not in stopwords and len(w.strip()) > 2] | |
| words.extend(doc_words) | |
| if len(words) < 5: | |
| return pd.DataFrame() | |
| n = 2 if n_gram_type == "Bigrams (2-word pairs)" else 3 | |
| # Generate n-grams | |
| ngrams_list = [] | |
| for i in range(len(words) - n + 1): | |
| ngram = tuple(words[i:i+n]) | |
| ngrams_list.append(ngram) | |
| ngram_counts = Counter(ngrams_list) | |
| # Filter by minimum frequency | |
| filtered_ngrams = {k: v for k, v in ngram_counts.items() if v >= min_freq} | |
| if not filtered_ngrams: | |
| return pd.DataFrame() | |
| results = [] | |
| if metric == "Raw Joint Frequency": | |
| for ngram, count in Counter(filtered_ngrams).most_common(top_n): | |
| results.append({ | |
| "Word Phrase": " ".join(ngram), | |
| "Score": float(count), | |
| "Measure": "Frequency" | |
| }) | |
| else: | |
| # Pointwise Mutual Information (PMI) | |
| # PMI(x, y) = log2( P(x,y) / (P(x)*P(y)) ) | |
| word_counts = Counter(words) | |
| total_words = len(words) | |
| total_ngrams = len(ngrams_list) | |
| pmi_scores = {} | |
| for ngram, count in filtered_ngrams.items(): | |
| # Joint probability | |
| p_joint = count / total_ngrams | |
| # Marginal probabilities product | |
| p_marginals = 1.0 | |
| for word in ngram: | |
| p_marginals *= (word_counts[word] / total_words) | |
| pmi = np.log2(p_joint / p_marginals) | |
| pmi_scores[ngram] = pmi | |
| sorted_pmi = sorted(pmi_scores.items(), key=lambda x: x[1], reverse=True)[:top_n] | |
| for ngram, score in sorted_pmi: | |
| results.append({ | |
| "Word Phrase": " ".join(ngram), | |
| "Score": round(float(score), 4), | |
| "Measure": "PMI Score" | |
| }) | |
| return pd.DataFrame(results) | |
| def run_analysis(file_obj, text_col, n_gram_type, metric, min_freq, top_n): | |
| if file_obj is None: | |
| return None, None, None, "Please upload a dataset first." | |
| # Re-load data | |
| df, _, _ = load_data(file_obj) | |
| if df is None: | |
| return None, None, None, "Failed to parse the file." | |
| docs = df[text_col].astype(str).fillna("").tolist() | |
| if not docs: | |
| return None, None, None, "No text documents found in the selected column." | |
| try: | |
| df_res = calculate_collocations(docs, n_gram_type, metric, min_freq, top_n) | |
| if df_res.empty: | |
| return None, None, None, "No collocations met the minimum frequency filter. Try lowering 'Min Word Co-occurrences'." | |
| # Plotly Bar Chart | |
| fig = px.bar( | |
| df_res, | |
| x="Score", | |
| y="Word Phrase", | |
| orientation="h", | |
| color="Score", | |
| title=f"Top Collocations via {metric}", | |
| template="plotly_dark", | |
| color_continuous_scale="Cividis" | |
| ) | |
| fig.update_layout(yaxis={'categoryorder': 'total ascending'}, height=450, margin=dict(l=20, r=20, t=40, b=20)) | |
| # Export CSV | |
| csv_path = "collocations_report.csv" | |
| df_res.to_csv(csv_path, index=False) | |
| status_md = f"Successfully calculated top **{len(df_res)}** collocations using {metric}." | |
| return df_res, fig, csv_path, status_md | |
| except Exception as e: | |
| return None, None, None, f"Analysis failed: {str(e)}" | |
| custom_css = """ | |
| body { | |
| background-color: #0b0f19; | |
| color: #f3f4f6; | |
| } | |
| .gradio-container { | |
| font-family: 'Inter', sans-serif !important; | |
| } | |
| h1, h2 { | |
| color: #6366f1 !important; | |
| } | |
| """ | |
| with gr.Blocks(theme=gr.themes.Default(primary_hue="indigo", secondary_hue="slate"), css=custom_css) as demo: | |
| df_state = gr.State() | |
| gr.HTML(""" | |
| <div style="text-align: center; margin-bottom: 2rem;"> | |
| <h1 style="font-size: 2.5rem; font-weight: 700; margin-bottom: 0.5rem; background: linear-gradient(to right, #6366f1, #a855f7); -webkit-background-clip: text; -webkit-text-fill-color: transparent;">Interactive Collocation Analyzer</h1> | |
| <p style="font-size: 1.1rem; color: #94a3b8; max-width: 800px; margin: 0 auto;"> | |
| Extract and analyze recurring word pairs (Bigrams) or triplets (Trigrams) that frequently co-occur in your text. | |
| Toggle between raw joint counts and Pointwise Mutual Information (PMI) to reveal locked idioms and idioms. | |
| </p> | |
| </div> | |
| """) | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| gr.Markdown("### 1. Upload Dataset") | |
| file_input = gr.File(label="Upload Dataset (.csv, .xlsx, .txt)", file_types=[".csv", ".xlsx", ".txt"]) | |
| text_column_selector = gr.Dropdown( | |
| label="Target Text Column", | |
| choices=[], | |
| visible=False, | |
| interactive=True | |
| ) | |
| status_text = gr.Markdown("No file uploaded yet.") | |
| gr.Markdown("### 2. Configure Collocations") | |
| n_gram_type = gr.Radio( | |
| choices=["Bigrams (2-word pairs)", "Trigrams (3-word triplets)"], | |
| value="Bigrams (2-word pairs)", | |
| label="Co-occurrence Type" | |
| ) | |
| metric_selector = gr.Radio( | |
| choices=["Raw Joint Frequency", "Pointwise Mutual Information (PMI)"], | |
| value="Raw Joint Frequency", | |
| label="Association Metric" | |
| ) | |
| with gr.Row(): | |
| min_freq = gr.Slider(minimum=1, maximum=50, value=3, step=1, label="Min Word Co-occurrences") | |
| top_n = gr.Slider(minimum=5, maximum=40, value=15, step=1, label="Phrases to Display") | |
| run_btn = gr.Button("Analyze Collocations", variant="primary") | |
| with gr.Column(scale=2): | |
| gr.Markdown("### 3. Collocations Results") | |
| status_markdown = gr.Markdown("Configure settings and click 'Analyze Collocations' to run.") | |
| with gr.Tabs(): | |
| with gr.TabItem("Collocations Plot"): | |
| chart_output = gr.Plot(label="Collocation Strength Plot") | |
| with gr.TabItem("Collocations Table"): | |
| table_output = gr.Dataframe( | |
| headers=["Word Phrase", "Score", "Measure"], | |
| datatype=["str", "number", "str"], | |
| interactive=False, | |
| wrap=True | |
| ) | |
| gr.Markdown("### 4. Export") | |
| download_csv = gr.File(label="Download Collocations Report (CSV)") | |
| file_input.change( | |
| fn=load_data, | |
| inputs=file_input, | |
| outputs=[df_state, text_column_selector, status_text] | |
| ) | |
| run_btn.click( | |
| fn=run_analysis, | |
| inputs=[file_input, text_column_selector, n_gram_type, metric_selector, min_freq, top_n], | |
| outputs=[table_output, chart_output, download_csv, status_markdown] | |
| ) | |
| if __name__ == "__main__": | |
| demo.launch() | |