Spaces:
Configuration error
Configuration error
| import os | |
| import io | |
| import json | |
| from typing import Tuple | |
| import numpy as np | |
| import pandas as pd | |
| import gradio as gr | |
| import plotly.express as px | |
| import requests | |
| from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer | |
| APP_TITLE = "StreamSmart Recommender" | |
| APP_SUBTITLE = ( | |
| "Improve streaming recommendations by combining viewer review sentiment " | |
| "with watch-time and engagement metrics." | |
| ) | |
| analyzer = SentimentIntensityAnalyzer() | |
| REQUIRED_REVIEW_COLS = ["title", "review_text"] | |
| REQUIRED_WATCH_COLS = [ | |
| "title", | |
| "genre", | |
| "avg_watch_time", | |
| "completion_rate", | |
| "drop_off_rate", | |
| "rewatch_rate", | |
| "click_through_rate", | |
| ] | |
| def clean_text(text: str) -> str: | |
| if pd.isna(text): | |
| return "" | |
| text = str(text).strip().replace("\n", " ") | |
| return " ".join(text.split()) | |
| def compute_sentiment(text: str) -> float: | |
| return analyzer.polarity_scores(clean_text(text))["compound"] | |
| def minmax(series: pd.Series) -> pd.Series: | |
| series = pd.to_numeric(series, errors="coerce").fillna(0) | |
| min_v = series.min() | |
| max_v = series.max() | |
| if max_v == min_v: | |
| return pd.Series(np.full(len(series), 0.5), index=series.index) | |
| return (series - min_v) / (max_v - min_v) | |
| def sentiment_label(score: float) -> str: | |
| if score >= 0.2: | |
| return "Positive" | |
| if score <= -0.2: | |
| return "Negative" | |
| return "Neutral" | |
| def action_label(score: float) -> str: | |
| if score >= 80: | |
| return "Promote strongly" | |
| if score >= 65: | |
| return "Promote selectively" | |
| if score >= 45: | |
| return "Investigate mismatch" | |
| return "Reduce priority" | |
| def business_explanation(row: pd.Series) -> str: | |
| s = row["avg_sentiment"] | |
| c = row["completion_rate"] | |
| d = row["drop_off_rate"] | |
| score = row["recommendation_score"] | |
| if s >= 0.2 and c >= 0.7 and d <= 0.3: | |
| return ( | |
| f"{row['title']} has strong viewer satisfaction and high completion, so it is a good candidate " | |
| "for broader recommendation placement." | |
| ) | |
| if s >= 0.2 and c < 0.7: | |
| return ( | |
| f"{row['title']} gets positive reactions from viewers who engage with it, but completion is weaker. " | |
| "This suggests the title may perform better with more targeted audience matching." | |
| ) | |
| if s < 0.2 and c >= 0.7: | |
| return ( | |
| f"{row['title']} keeps viewers watching, but sentiment is not especially strong. This may indicate " | |
| "good initial appeal with weaker perceived quality or expectation mismatch." | |
| ) | |
| if score < 45: | |
| return ( | |
| f"{row['title']} shows weak satisfaction and engagement signals overall, so it should not be prioritized " | |
| "in recommendation slots until content positioning improves." | |
| ) | |
| return ( | |
| f"{row['title']} is a mixed case: some engagement indicators are promising, but the platform should review " | |
| "audience fit, metadata, or recommendation placement before scaling promotion." | |
| ) | |
| def validate_columns(df: pd.DataFrame, required_cols: list, name: str) -> None: | |
| missing = [c for c in required_cols if c not in df.columns] | |
| if missing: | |
| raise gr.Error(f"{name} is missing required columns: {missing}") | |
| def make_demo_data() -> Tuple[pd.DataFrame, pd.DataFrame]: | |
| reviews = pd.DataFrame( | |
| { | |
| "title": [ | |
| "Midnight City", "Midnight City", "Ocean Echoes", "Ocean Echoes", | |
| "Crimson Truth", "Crimson Truth", "Quiet Orbit", "Quiet Orbit", | |
| "Laugh Track", "Laugh Track", "Golden Hour", "Golden Hour", | |
| ], | |
| "review_text": [ | |
| "Amazing pacing and really addictive storyline.", | |
| "Loved the characters and watched it in one sitting.", | |
| "Beautiful idea but too slow in the middle.", | |
| "Strong visuals, but I almost stopped halfway.", | |
| "Suspenseful and smart, one of the best thrillers.", | |
| "Great acting and excellent ending.", | |
| "Interesting concept but not very engaging.", | |
| "Felt too long and the story did not pull me in.", | |
| "Funny and light, easy to keep watching.", | |
| "Very entertaining and rewatchable.", | |
| "Good cast but the episodes drag a bit.", | |
| "Not bad, but I expected more excitement.", | |
| ], | |
| "genre": [ | |
| "Sci-Fi", "Sci-Fi", "Drama", "Drama", "Thriller", "Thriller", | |
| "Sci-Fi", "Sci-Fi", "Comedy", "Comedy", "Drama", "Drama", | |
| ], | |
| } | |
| ) | |
| watch = pd.DataFrame( | |
| { | |
| "title": ["Midnight City", "Ocean Echoes", "Crimson Truth", "Quiet Orbit", "Laugh Track", "Golden Hour"], | |
| "genre": ["Sci-Fi", "Drama", "Thriller", "Sci-Fi", "Comedy", "Drama"], | |
| "avg_watch_time": [83, 58, 79, 41, 72, 54], | |
| "completion_rate": [0.86, 0.61, 0.81, 0.39, 0.76, 0.57], | |
| "drop_off_rate": [0.18, 0.33, 0.21, 0.48, 0.24, 0.37], | |
| "rewatch_rate": [0.31, 0.15, 0.27, 0.08, 0.25, 0.11], | |
| "click_through_rate": [0.42, 0.36, 0.39, 0.29, 0.41, 0.34], | |
| } | |
| ) | |
| return reviews, watch | |
| def run_analysis(reviews_file, watch_file, use_demo: bool): | |
| if use_demo: | |
| reviews_df, watch_df = make_demo_data() | |
| else: | |
| if reviews_file is None or watch_file is None: | |
| raise gr.Error("Upload both CSV files or use the demo dataset.") | |
| reviews_df = pd.read_csv(reviews_file.name) | |
| watch_df = pd.read_csv(watch_file.name) | |
| validate_columns(reviews_df, REQUIRED_REVIEW_COLS, "Reviews CSV") | |
| validate_columns(watch_df, REQUIRED_WATCH_COLS, "Watch-time CSV") | |
| reviews = reviews_df.copy() | |
| watch = watch_df.copy() | |
| reviews["review_text"] = reviews["review_text"].apply(clean_text) | |
| reviews["sentiment_score"] = reviews["review_text"].apply(compute_sentiment) | |
| reviews["sentiment_label"] = reviews["sentiment_score"].apply(sentiment_label) | |
| if "genre" in reviews.columns: | |
| review_agg = reviews.groupby("title", as_index=False).agg( | |
| avg_sentiment=("sentiment_score", "mean"), | |
| review_count=("sentiment_score", "count"), | |
| dominant_genre=("genre", lambda s: s.mode().iat[0] if not s.mode().empty else s.iloc[0]), | |
| ) | |
| else: | |
| review_agg = reviews.groupby("title", as_index=False).agg( | |
| avg_sentiment=("sentiment_score", "mean"), | |
| review_count=("sentiment_score", "count"), | |
| ) | |
| review_agg["dominant_genre"] = "Unknown" | |
| merged = pd.merge(watch, review_agg, on="title", how="left") | |
| merged["avg_sentiment"] = merged["avg_sentiment"].fillna(0) | |
| merged["review_count"] = merged["review_count"].fillna(0).astype(int) | |
| merged["genre"] = merged["genre"].fillna(merged["dominant_genre"]).fillna("Unknown") | |
| merged["sentiment_norm"] = minmax(merged["avg_sentiment"]) | |
| merged["completion_norm"] = minmax(merged["completion_rate"]) | |
| merged["watch_norm"] = minmax(merged["avg_watch_time"]) | |
| merged["rewatch_norm"] = minmax(merged["rewatch_rate"]) | |
| merged["ctr_norm"] = minmax(merged["click_through_rate"]) | |
| merged["dropoff_norm"] = minmax(merged["drop_off_rate"]) | |
| raw_score = ( | |
| 0.35 * merged["sentiment_norm"] | |
| + 0.30 * merged["completion_norm"] | |
| + 0.20 * merged["watch_norm"] | |
| + 0.10 * merged["rewatch_norm"] | |
| + 0.05 * merged["ctr_norm"] | |
| - 0.15 * merged["dropoff_norm"] | |
| ) | |
| merged["recommendation_score"] = (raw_score.clip(lower=0) * 100).round(2) | |
| merged["action"] = merged["recommendation_score"].apply(action_label) | |
| merged["explanation"] = merged.apply(business_explanation, axis=1) | |
| merged = merged.sort_values("recommendation_score", ascending=False).reset_index(drop=True) | |
| summary = ( | |
| f"Reviews analyzed: {len(reviews)} | Titles scored: {merged['title'].nunique()} | " | |
| f"Average sentiment: {merged['avg_sentiment'].mean():.2f} | " | |
| f"Average completion rate: {merged['completion_rate'].mean():.2f}" | |
| ) | |
| table_cols = [ | |
| "title", "genre", "avg_sentiment", "avg_watch_time", "completion_rate", | |
| "drop_off_rate", "rewatch_rate", "click_through_rate", "review_count", | |
| "recommendation_score", "action" | |
| ] | |
| top_table = merged[table_cols] | |
| top_plot = px.bar( | |
| merged.head(10), | |
| x="title", | |
| y="recommendation_score", | |
| title="Top Titles by Recommendation Score", | |
| ) | |
| scatter_plot = px.scatter( | |
| merged, | |
| x="avg_sentiment", | |
| y="completion_rate", | |
| size="avg_watch_time", | |
| hover_name="title", | |
| color="genre", | |
| title="Sentiment vs Completion Rate", | |
| ) | |
| genre_plot = px.bar( | |
| merged.groupby("genre", as_index=False)["recommendation_score"].mean().sort_values("recommendation_score", ascending=False), | |
| x="genre", | |
| y="recommendation_score", | |
| title="Average Recommendation Score by Genre", | |
| ) | |
| processed_csv = io.StringIO() | |
| top_table.to_csv(processed_csv, index=False) | |
| payload = merged.to_json(orient="records") | |
| return summary, top_table, top_plot, scatter_plot, genre_plot, payload, processed_csv.getvalue() | |
| def inspect_title(payload: str, selected_title: str): | |
| if not payload: | |
| raise gr.Error("Run the analysis first.") | |
| records = json.loads(payload) | |
| df = pd.DataFrame(records) | |
| if selected_title not in df["title"].values: | |
| raise gr.Error("Title not found.") | |
| row = df[df["title"] == selected_title].iloc[0] | |
| return ( | |
| f"Title: {row['title']}\n" | |
| f"Genre: {row['genre']}\n" | |
| f"Average sentiment: {row['avg_sentiment']:.2f}\n" | |
| f"Average watch time: {row['avg_watch_time']:.2f}\n" | |
| f"Completion rate: {row['completion_rate']:.2f}\n" | |
| f"Drop-off rate: {row['drop_off_rate']:.2f}\n" | |
| f"Recommendation score: {row['recommendation_score']:.2f}\n" | |
| f"Suggested action: {row['action']}\n\n" | |
| f"Explanation: {row['explanation']}" | |
| ) | |
| def update_title_choices(payload: str): | |
| if not payload: | |
| return gr.Dropdown(choices=[], value=None) | |
| df = pd.DataFrame(json.loads(payload)) | |
| choices = sorted(df["title"].dropna().unique().tolist()) | |
| value = choices[0] if choices else None | |
| return gr.Dropdown(choices=choices, value=value) | |
| def _post_to_webhook(env_name: str, body: dict, success_label: str) -> str: | |
| webhook_url = os.getenv(env_name, "").strip() | |
| if not webhook_url: | |
| return f"{env_name} is not set yet. Add it as a Hugging Face Space secret, then try again." | |
| response = requests.post(webhook_url, json=body, timeout=60) | |
| response.raise_for_status() | |
| try: | |
| result = response.json() | |
| return f"{success_label} ran successfully. Response: {json.dumps(result, indent=2)}" | |
| except Exception: | |
| return f"{success_label} ran successfully. Raw response: {response.text}" | |
| def send_to_processing_workflow(payload: str): | |
| if not payload: | |
| raise gr.Error("Run the analysis first.") | |
| data = json.loads(payload) | |
| return _post_to_webhook( | |
| "N8N_PROCESS_WEBHOOK_URL", | |
| {"app": APP_TITLE, "records": data, "record_count": len(data)}, | |
| "n8n processing workflow", | |
| ) | |
| def send_to_report_workflow(payload: str): | |
| if not payload: | |
| raise gr.Error("Run the analysis first.") | |
| data = json.loads(payload) | |
| top5 = data[:5] | |
| return _post_to_webhook( | |
| "N8N_REPORT_WEBHOOK_URL", | |
| {"app": APP_TITLE, "top_recommendations": top5, "all_results": data}, | |
| "n8n report workflow", | |
| ) | |
| with gr.Blocks(title=APP_TITLE) as demo: | |
| gr.Markdown(f"# {APP_TITLE}\n\n{APP_SUBTITLE}") | |
| gr.Markdown( | |
| "This app combines qualitative viewer review sentiment with quantitative watch-time metrics " | |
| "to score how strongly each title should be recommended on a streaming platform." | |
| ) | |
| payload_state = gr.State("") | |
| csv_state = gr.State("") | |
| with gr.Tab("1. Upload & Run"): | |
| use_demo = gr.Checkbox(label="Use built-in demo dataset", value=True) | |
| reviews_file = gr.File(label="Upload reviews CSV", file_types=[".csv"]) | |
| watch_file = gr.File(label="Upload watch-time CSV", file_types=[".csv"]) | |
| run_btn = gr.Button("Run Analysis", variant="primary") | |
| summary_box = gr.Textbox(label="Processing Summary", lines=2) | |
| with gr.Tab("2. Dashboard"): | |
| results_table = gr.Dataframe(label="Scored Titles") | |
| chart_1 = gr.Plot(label="Top Recommendation Scores") | |
| chart_2 = gr.Plot(label="Sentiment vs Completion") | |
| chart_3 = gr.Plot(label="Genre Performance") | |
| with gr.Tab("3. Title Drilldown"): | |
| title_dropdown = gr.Dropdown(label="Select a title", choices=[]) | |
| detail_box = gr.Textbox(label="Title Recommendation Detail", lines=10) | |
| inspect_btn = gr.Button("Explain Selected Title") | |
| with gr.Tab("4. n8n Automation"): | |
| gr.Markdown( | |
| "Set these Hugging Face Space secrets before using the buttons below:\n\n" | |
| "- `N8N_PROCESS_WEBHOOK_URL`\n" | |
| "- `N8N_REPORT_WEBHOOK_URL`" | |
| ) | |
| process_btn = gr.Button("Send Full Results to n8n Processing Workflow") | |
| process_status = gr.Textbox(label="Processing Workflow Status", lines=5) | |
| report_btn = gr.Button("Send Top Recommendations to n8n Report Workflow") | |
| report_status = gr.Textbox(label="Report Workflow Status", lines=5) | |
| with gr.Tab("5. Download"): | |
| download_file = gr.File(label="Download processed CSV") | |
| def save_csv_text(csv_text: str): | |
| path = "/tmp/processed_streamsmart_results.csv" | |
| with open(path, "w", encoding="utf-8") as f: | |
| f.write(csv_text) | |
| return path | |
| run_btn.click( | |
| fn=run_analysis, | |
| inputs=[reviews_file, watch_file, use_demo], | |
| outputs=[summary_box, results_table, chart_1, chart_2, chart_3, payload_state, csv_state], | |
| ).then( | |
| fn=update_title_choices, | |
| inputs=[payload_state], | |
| outputs=[title_dropdown], | |
| ).then( | |
| fn=save_csv_text, | |
| inputs=[csv_state], | |
| outputs=[download_file], | |
| ) | |
| inspect_btn.click( | |
| fn=inspect_title, | |
| inputs=[payload_state, title_dropdown], | |
| outputs=[detail_box], | |
| ) | |
| process_btn.click( | |
| fn=send_to_processing_workflow, | |
| inputs=[payload_state], | |
| outputs=[process_status], | |
| ) | |
| report_btn.click( | |
| fn=send_to_report_workflow, | |
| inputs=[payload_state], | |
| outputs=[report_status], | |
| ) | |
| if __name__ == "__main__": | |
| demo.launch() | |