# app.py # NFL Sentiment Lab — Gradio Upload App # Upload CSV (player + text) -> preprocess (NLTK) -> VADER sentiment -> aggregate by player # Outputs: CSVs + PNG graphs + ZIP bundle, all downloadable in the UI. import re import zipfile import tempfile from pathlib import Path from typing import Optional, Tuple import numpy as np import pandas as pd import matplotlib.pyplot as plt import gradio as gr # ----------------------------- # NLTK bootstrap (includes punkt_tab) # ----------------------------- def ensure_nltk(): import nltk from nltk.data import find nltk_data_dir = Path(tempfile.gettempdir()) / "nltk_data" nltk_data_dir.mkdir(parents=True, exist_ok=True) if str(nltk_data_dir) not in nltk.data.path: nltk.data.path.append(str(nltk_data_dir)) # Include punkt_tab (some environments require it) resources = [ ("tokenizers/punkt", "punkt"), ("tokenizers/punkt_tab", "punkt_tab"), # <-- key fix ("corpora/stopwords", "stopwords"), ("corpora/wordnet", "wordnet"), ("sentiment/vader_lexicon", "vader_lexicon"), ] for res_path, res_name in resources: try: find(res_path) except LookupError: nltk.download(res_name, download_dir=str(nltk_data_dir), quiet=True) return nltk # Download at startup so tokenization doesn't fail mid-run ensure_nltk() # ----------------------------- # Lab preprocessing + sentiment # ----------------------------- def build_preprocess_components(): nltk = ensure_nltk() from nltk.corpus import stopwords from nltk.stem import WordNetLemmatizer import unicodedata stop_words = set(stopwords.words("english")) lemmatizer = WordNetLemmatizer() def remove_non_ascii(words): new_words = [] for word in words: new_word = ( unicodedata.normalize("NFKD", word) .encode("ascii", "ignore") .decode("utf-8", "ignore") ) new_words.append(new_word) return new_words def to_lowercase(words): return [w.lower() for w in words] def remove_punctuation(words): new_words = [] for word in words: new_word = re.sub(r"[^\w\s]", "", word) if new_word != "": new_words.append(new_word) return new_words def remove_stopwords(words): return [w for w in words if w not in stop_words] def lemmatize_list(words): return [lemmatizer.lemmatize(w, pos="v") for w in words] def normalize(text) -> str: # ---- CRITICAL FIX: coerce any non-string (floats/NaN) safely ---- if text is None or (isinstance(text, float) and np.isnan(text)): text = "" elif not isinstance(text, str): # handles ints, floats, pandas NA, etc. try: if pd.isna(text): text = "" else: text = str(text) except Exception: text = str(text) # preserve_line=True avoids sentence tokenization path that can be brittle words = nltk.word_tokenize(text, preserve_line=True) words = remove_non_ascii(words) words = to_lowercase(words) words = remove_punctuation(words) words = remove_stopwords(words) words = lemmatize_list(words) return " ".join(words) return normalize def build_sentiment_components(): ensure_nltk() from nltk.sentiment import SentimentIntensityAnalyzer sia = SentimentIntensityAnalyzer() def get_sentiment(text) -> str: # also safe-coerce here (belt + suspenders) if text is None or (isinstance(text, float) and np.isnan(text)): text = "" elif not isinstance(text, str): try: if pd.isna(text): text = "" else: text = str(text) except Exception: text = str(text) scores = sia.polarity_scores(text) compound = scores["compound"] pos = scores["pos"] neg = scores["neg"] if compound >= 0.05 and pos > neg: return "Positive" elif compound <= -0.05 and neg > pos: return "Negative" else: return "Neutral" return get_sentiment # ----------------------------- # Helpers # ----------------------------- def get_uploaded_path(file_obj) -> Path: if file_obj is None: raise gr.Error("Upload a CSV first.") if isinstance(file_obj, str): return Path(file_obj) name = getattr(file_obj, "name", None) if name: return Path(name) if isinstance(file_obj, dict) and "name" in file_obj: return Path(file_obj["name"]) raise gr.Error("Could not read uploaded file path. Please re-upload the CSV.") def detect_columns(df: pd.DataFrame) -> Tuple[Optional[str], Optional[str]]: text_col = next((c for c in ["text", "comment", "body", "content"] if c in df.columns), None) player_col = next((c for c in ["player", "Player", "player_name", "name"] if c in df.columns), None) return player_col, text_col def save_top25_plot(top_25: pd.DataFrame, out_path: Path, label_name: str): plt.figure(figsize=(10, 6)) plot_df = top_25.sort_values("overall_sentiment_score", ascending=True) plt.barh(plot_df.index.astype(str), plot_df["overall_sentiment_score"].values) plt.title("Top 25 Players by Overall Sentiment Score") plt.xlabel("Overall Sentiment Score") plt.ylabel(label_name) plt.tight_layout() plt.savefig(out_path, dpi=200) plt.close() def save_sentiment_distribution_plot(df: pd.DataFrame, out_path: Path): counts = df["sentiment"].value_counts().reindex(["Positive", "Neutral", "Negative"]).fillna(0) plt.figure(figsize=(7, 5)) plt.bar(counts.index.astype(str), counts.values) plt.title("Sentiment Distribution (All Comments)") plt.xlabel("Sentiment") plt.ylabel("Count") plt.tight_layout() plt.savefig(out_path, dpi=200) plt.close() def save_wordcloud_positive(df: pd.DataFrame, raw_text_series: pd.Series, out_path: Path): try: from wordcloud import WordCloud positive_text = " ".join(raw_text_series[df["sentiment"] == "Positive"].astype(str).tolist()) if positive_text.strip() == "": raise ValueError("No positive comments available for word cloud.") wc = WordCloud(width=1200, height=800, background_color="white").generate(positive_text) plt.figure(figsize=(10, 6)) plt.imshow(wc, interpolation="bilinear") plt.axis("off") plt.title("Word Cloud (Positive Comments)") plt.tight_layout() plt.savefig(out_path, dpi=200) plt.close() except Exception as e: plt.figure(figsize=(10, 4)) plt.text(0.02, 0.5, f"WordCloud not generated.\nReason: {e}", fontsize=12) plt.axis("off") plt.tight_layout() plt.savefig(out_path, dpi=200) plt.close() def zip_results(folder: Path, zip_path: Path): with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zf: for file_path in folder.rglob("*"): if file_path.is_file(): zf.write(file_path, arcname=file_path.relative_to(folder)) # ----------------------------- # Core lab analysis # ----------------------------- def run_lab_analysis(file_obj, player_col_choice: str, text_col_choice: str): csv_path = get_uploaded_path(file_obj) try: df = pd.read_csv(csv_path) except Exception as e: raise gr.Error(f"Could not read CSV: {e}") if df.empty: raise gr.Error("CSV loaded, but it’s empty.") detected_player, detected_text = detect_columns(df) player_col = player_col_choice or detected_player text_col = text_col_choice or detected_text if not player_col or player_col not in df.columns: raise gr.Error("Could not find the PLAYER column. Select it from the dropdown.") if not text_col or text_col not in df.columns: raise gr.Error("Could not find the TEXT column. Select it from the dropdown.") out_dir = Path(tempfile.mkdtemp(prefix="nfl_sentiment_lab_")) normalize = build_preprocess_components() get_sentiment = build_sentiment_components() work = df.copy() # ---- CRITICAL FIX: clean + coerce columns BEFORE apply() ---- work[player_col] = work[player_col].fillna("").astype(str) work[text_col] = work[text_col].fillna("").astype(str) # Lab steps: clean_text -> sentiment work["clean_text"] = work[text_col].apply(normalize) work["sentiment"] = work["clean_text"].apply(get_sentiment) # Row-level output row_csv = out_dir / "NFL_reddit_sentiment_analysis.csv" work.to_csv(row_csv, index=False) # Player aggregation player_sentiment = ( work.groupby(player_col)["sentiment"] .value_counts() .unstack(fill_value=0) ) # Ensure the 3 categories exist for col in ["Positive", "Neutral", "Negative"]: if col not in player_sentiment.columns: player_sentiment[col] = 0 totals = player_sentiment[["Positive", "Neutral", "Negative"]].sum(axis=1).replace(0, np.nan) player_sentiment["percent_positive"] = (player_sentiment["Positive"] / totals).fillna(0.0) player_sentiment["overall_sentiment_score"] = ( (player_sentiment["Positive"] * 1 + player_sentiment["Neutral"] * 0 + player_sentiment["Negative"] * -1) / totals ).fillna(0.0) player_sentiment_sorted = player_sentiment.sort_values("overall_sentiment_score", ascending=False) player_csv = out_dir / "player_sentiment_results.csv" player_sentiment_sorted.to_csv(player_csv) # Top 25 top_25 = player_sentiment_sorted.head(25) top25_csv = out_dir / "top_25_players.csv" top_25.to_csv(top25_csv) # Plots top25_png = out_dir / "top_25_players.png" dist_png = out_dir / "sentiment_distribution.png" wc_png = out_dir / "positive_wordcloud.png" save_top25_plot(top_25, top25_png, label_name=player_col) save_sentiment_distribution_plot(work, dist_png) save_wordcloud_positive(work, work[text_col], wc_png) # Zip bundle zip_path = out_dir / "analysis_results.zip" zip_results(out_dir, zip_path) # Summary n_rows = len(work) n_players = work[player_col].nunique(dropna=True) sent_counts = work["sentiment"].value_counts().to_dict() summary_md = f""" ### Lab Analysis Summary - Rows (comments): **{n_rows:,}** - Unique players: **{n_players:,}** - Sentiment counts: - Positive: **{sent_counts.get("Positive", 0):,}** - Neutral: **{sent_counts.get("Neutral", 0):,}** - Negative: **{sent_counts.get("Negative", 0):,}** """ top25_display = top_25.reset_index().rename(columns={player_col: "player"}) return ( summary_md, top25_display, str(top25_png), str(dist_png), str(wc_png), str(row_csv), str(player_csv), str(top25_csv), str(zip_path), ) def load_csv_columns(file_obj): if file_obj is None: return ( gr.update(choices=[], value=None), gr.update(choices=[], value=None), "Upload a CSV to populate column lists.", ) csv_path = get_uploaded_path(file_obj) try: peek = pd.read_csv(csv_path, nrows=50) except Exception as e: raise gr.Error(f"Could not read CSV: {e}") cols = list(peek.columns) detected_player, detected_text = detect_columns(peek) note = f"Detected → player: **{detected_player}**, text: **{detected_text}** (change if needed)." return ( gr.update(choices=cols, value=(detected_player if detected_player in cols else None)), gr.update(choices=cols, value=(detected_text if detected_text in cols else None)), note, ) # ----------------------------- # Gradio UI # ----------------------------- with gr.Blocks(title="NFL Sentiment Lab — CSV Upload App") as demo: gr.Markdown( """ # NFL Sentiment Lab (CSV Upload) Upload a CSV with at least: - a **player** column - a **text** column Click **Run Lab Analysis** to: - preprocess text → VADER sentiment → aggregate by player - show graphs - download CSVs + ZIP bundle """ ) file_in = gr.File(label="Drag & drop CSV here", file_types=[".csv"]) with gr.Row(): # ---- FIX: value=None avoids warning when choices=[] ---- player_col_dd = gr.Dropdown(label="Player column", choices=[], value=None) text_col_dd = gr.Dropdown(label="Text column", choices=[], value=None) detect_note = gr.Markdown("Upload a CSV to populate column lists.") file_in.change( fn=load_csv_columns, inputs=[file_in], outputs=[player_col_dd, text_col_dd, detect_note], ) run_btn = gr.Button("Run Lab Analysis") summary = gr.Markdown() top25_table = gr.Dataframe(label="Top 25 Players (by overall sentiment score)", interactive=False) with gr.Row(): img_top25 = gr.Image(label="Top 25 Players Plot", type="filepath") img_dist = gr.Image(label="Sentiment Distribution", type="filepath") img_wc = gr.Image(label="Positive Word Cloud", type="filepath") gr.Markdown("## Downloads") with gr.Row(): out_row_csv = gr.File(label="Row-level sentiment CSV") out_player_csv = gr.File(label="Player sentiment summary CSV") out_top25_csv = gr.File(label="Top-25 players CSV") out_zip = gr.File(label="ZIP (all results + graphs)") run_btn.click( fn=run_lab_analysis, inputs=[file_in, player_col_dd, text_col_dd], outputs=[ summary, top25_table, img_top25, img_dist, img_wc, out_row_csv, out_player_csv, out_top25_csv, out_zip, ], ) if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860)