import os
import re
import json
import time
import traceback
from pathlib import Path
from typing import Dict, Any, List, Tuple
import pandas as pd
import gradio as gr
import papermill as pm
import plotly.graph_objects as go
# Optional LLM (HuggingFace Inference API)
try:
from huggingface_hub import InferenceClient
except Exception:
InferenceClient = None
# =========================================================
# CONFIG
# =========================================================
BASE_DIR = Path(__file__).resolve().parent
NB1 = os.environ.get("NB1", "datacreation.ipynb").strip()
NB2 = os.environ.get("NB2", "pythonanalysis.ipynb").strip()
RUNS_DIR = BASE_DIR / "runs"
ART_DIR = BASE_DIR / "artifacts"
PY_FIG_DIR = ART_DIR / "figures" # notebooks write to artifacts/figures/
PY_TAB_DIR = ART_DIR / "tables" # notebooks write to artifacts/tables/
PAPERMILL_TIMEOUT = int(os.environ.get("PAPERMILL_TIMEOUT", "1800"))
MAX_PREVIEW_ROWS = int(os.environ.get("MAX_FILE_PREVIEW_ROWS", "50"))
MAX_LOG_CHARS = int(os.environ.get("MAX_LOG_CHARS", "8000"))
HF_API_KEY = os.environ.get("HF_API_KEY", "").strip()
MODEL_NAME = os.environ.get("MODEL_NAME", "deepseek-ai/DeepSeek-R1").strip()
HF_PROVIDER = os.environ.get("HF_PROVIDER", "novita").strip()
N8N_WEBHOOK_URL = os.environ.get("N8N_WEBHOOK_URL", "").strip()
LLM_ENABLED = bool(HF_API_KEY) and InferenceClient is not None
llm_client = (
InferenceClient(provider=HF_PROVIDER, api_key=HF_API_KEY)
if LLM_ENABLED
else None
)
# =========================================================
# HELPERS
# =========================================================
def ensure_dirs():
for p in [RUNS_DIR, ART_DIR, PY_FIG_DIR, PY_TAB_DIR]:
p.mkdir(parents=True, exist_ok=True)
def stamp():
return time.strftime("%Y%m%d-%H%M%S")
def tail(text: str, n: int = MAX_LOG_CHARS) -> str:
return (text or "")[-n:]
def _ls(dir_path: Path, exts: Tuple[str, ...]) -> List[str]:
if not dir_path.is_dir():
return []
return sorted(p.name for p in dir_path.iterdir() if p.is_file() and p.suffix.lower() in exts)
def _read_csv(path: Path) -> pd.DataFrame:
return pd.read_csv(path, nrows=MAX_PREVIEW_ROWS)
def _read_json(path: Path):
with path.open(encoding="utf-8") as f:
return json.load(f)
def artifacts_index() -> Dict[str, Any]:
return {
"python": {
"figures": _ls(PY_FIG_DIR, (".png", ".jpg", ".jpeg")),
"tables": _ls(PY_TAB_DIR, (".csv", ".json")),
},
}
# =========================================================
# PIPELINE RUNNERS
# =========================================================
def run_notebook(nb_name: str) -> str:
ensure_dirs()
nb_in = BASE_DIR / nb_name
if not nb_in.exists():
return f"ERROR: {nb_name} not found."
nb_out = RUNS_DIR / f"run_{stamp()}_{nb_name}"
pm.execute_notebook(
input_path=str(nb_in),
output_path=str(nb_out),
cwd=str(BASE_DIR),
log_output=True,
progress_bar=False,
request_save_on_cell_execute=True,
execution_timeout=PAPERMILL_TIMEOUT,
)
return f"Executed {nb_name}"
def run_datacreation() -> str:
try:
log = run_notebook(NB1)
csvs = [f.name for f in (ART_DIR).glob("*.csv")]
return f"OK {log}\n\nCSVs in /artifacts/:\n" + "\n".join(f" - {c}" for c in sorted(csvs))
except Exception as e:
return f"FAILED {e}\n\n{traceback.format_exc()[-2000:]}"
def run_pythonanalysis() -> str:
try:
log = run_notebook(NB2)
idx = artifacts_index()
figs = idx["python"]["figures"]
tabs = idx["python"]["tables"]
return (
f"OK {log}\n\n"
f"Figures: {', '.join(figs) or '(none)'}\n"
f"Tables: {', '.join(tabs) or '(none)'}"
)
except Exception as e:
return f"FAILED {e}\n\n{traceback.format_exc()[-2000:]}"
def run_full_pipeline() -> str:
logs = []
logs.append("=" * 50)
logs.append("STEP 1/2: Data Creation (IMDb scrape + synthetic data)")
logs.append("=" * 50)
logs.append(run_datacreation())
logs.append("")
logs.append("=" * 50)
logs.append("STEP 2/2: Python Analysis (VADER, ARIMA, Random Forest)")
logs.append("=" * 50)
logs.append(run_pythonanalysis())
return "\n".join(logs)
# =========================================================
# GALLERY LOADERS
# =========================================================
def _load_all_figures() -> List[Tuple[str, str]]:
"""Return list of (filepath, caption) for Gallery."""
items = []
for p in sorted(PY_FIG_DIR.glob("*.png")):
items.append((str(p), p.stem.replace("_", " ").title()))
return items
def _load_table_safe(path: Path) -> pd.DataFrame:
try:
if path.suffix == ".json":
obj = _read_json(path)
if isinstance(obj, dict):
return pd.DataFrame([obj])
return pd.DataFrame(obj)
return _read_csv(path)
except Exception as e:
return pd.DataFrame([{"error": str(e)}])
def refresh_gallery():
figures = _load_all_figures()
idx = artifacts_index()
table_choices = list(idx["python"]["tables"])
default_df = pd.DataFrame()
if table_choices:
default_df = _load_table_safe(PY_TAB_DIR / table_choices[0])
return (
figures if figures else [],
gr.update(choices=table_choices, value=table_choices[0] if table_choices else None),
default_df,
)
def on_table_select(choice: str):
if not choice:
return pd.DataFrame([{"hint": "Select a table above."}])
path = PY_TAB_DIR / choice
if not path.exists():
return pd.DataFrame([{"error": f"File not found: {choice}"}])
return _load_table_safe(path)
# =========================================================
# KPI LOADER
# =========================================================
def load_kpis() -> Dict[str, Any]:
for candidate in [PY_TAB_DIR / "kpis.json", PY_FIG_DIR / "kpis.json"]:
if candidate.exists():
try:
return _read_json(candidate)
except Exception:
pass
return {}
# =========================================================
# AI DASHBOARD โ LLM picks what to display
# =========================================================
DASHBOARD_SYSTEM = """You are an AI dashboard assistant for a streaming platform analytics app.
The platform uses data from IMDb combined with synthetic streaming KPIs to predict whether
each TV show should be Renewed, Cancelled, or given more Investment.
AVAILABLE ARTIFACTS (only reference ones that exist):
{artifacts_json}
KPI SUMMARY: {kpis_json}
YOUR JOB:
1. Answer the user's question conversationally using the KPIs and your knowledge of the artifacts.
2. At the END of your response, output a JSON block (fenced with ```json ... ```) that tells
the dashboard which artifact to display. The JSON must have this shape:
{{"show": "figure"|"table"|"none", "scope": "python", "filename": "..."}}
- Use "show": "figure" to display a chart image.
- Use "show": "table" to display a CSV/JSON table.
- Use "show": "none" if no artifact is relevant.
RULES:
- If the user asks about sentiment or VADER scores, show vader_sentiment_analysis.png or figure chart "sentiment".
- If the user asks about viewership trends, show viewership_trends_sampled.png or chart "platform_streams".
- If the user asks about ARIMA or forecasting, show arima_forecasts.png or chart "platform_streams".
- If the user asks about the Random Forest or model accuracy, show random_forest_results.png.
- If the user asks about renewal decisions or genre analysis, show decision_analysis.png or table "renewal_recommendations".
- If the user asks about platform overview or total streams, show platform_overview.png or chart "platform_streams".
- If the user asks about feature importances, show table feature_importances.csv.
- If the user asks about recommendations or show list, show table renewal_recommendations.csv.
- Keep your answer concise (2-4 sentences), then the JSON block.
"""
JSON_BLOCK_RE = re.compile(r"```json\s*(\{.*?\})\s*```", re.DOTALL)
FALLBACK_JSON_RE = re.compile(r"\{[^{}]*\"show\"[^{}]*\}", re.DOTALL)
def _parse_display_directive(text: str) -> Dict[str, str]:
m = JSON_BLOCK_RE.search(text)
if m:
try:
return json.loads(m.group(1))
except json.JSONDecodeError:
pass
m = FALLBACK_JSON_RE.search(text)
if m:
try:
return json.loads(m.group(0))
except json.JSONDecodeError:
pass
return {"show": "none"}
def _clean_response(text: str) -> str:
return JSON_BLOCK_RE.sub("", text).strip()
def _n8n_call(msg: str) -> Tuple[str, Dict]:
import requests as req
try:
resp = req.post(N8N_WEBHOOK_URL, json={"question": msg}, timeout=20)
data = resp.json()
answer = data.get("answer", "No response from n8n workflow.")
chart = data.get("chart", "none")
if chart and chart != "none":
return answer, {"show": "figure", "chart": chart}
return answer, {"show": "none"}
except Exception as e:
return f"n8n error: {e}. Falling back to keyword matching.", None
def ai_chat(user_msg: str, history: list):
if not user_msg or not user_msg.strip():
return history, "", None, None
idx = artifacts_index()
kpis = load_kpis()
# Priority: n8n webhook > HF LLM > keyword fallback
if N8N_WEBHOOK_URL:
reply, directive = _n8n_call(user_msg)
if directive is None:
reply_fb, directive = _keyword_fallback(user_msg, idx, kpis)
reply += "\n\n" + reply_fb
elif not LLM_ENABLED:
reply, directive = _keyword_fallback(user_msg, idx, kpis)
else:
system = DASHBOARD_SYSTEM.format(
artifacts_json=json.dumps(idx, indent=2),
kpis_json=json.dumps(kpis, indent=2) if kpis else "(no KPIs yet โ run the pipeline first)",
)
msgs = [{"role": "system", "content": system}]
for entry in (history or [])[-6:]:
msgs.append(entry)
msgs.append({"role": "user", "content": user_msg})
try:
r = llm_client.chat_completion(
model=MODEL_NAME,
messages=msgs,
temperature=0.3,
max_tokens=600,
stream=False,
)
raw = (
r["choices"][0]["message"]["content"]
if isinstance(r, dict)
else r.choices[0].message.content
)
directive = _parse_display_directive(raw)
reply = _clean_response(raw)
except Exception as e:
reply = f"LLM error: {e}. Falling back to keyword matching."
reply_fb, directive = _keyword_fallback(user_msg, idx, kpis)
reply += "\n\n" + reply_fb
# Resolve artifacts โ build interactive Plotly charts when possible
chart_out = None
tab_out = None
show = directive.get("show", "none")
fname = directive.get("filename", "")
chart_name = directive.get("chart", "")
chart_builders = {
"platform_streams": build_platform_streams_chart,
"sentiment": build_sentiment_chart,
"renewal": build_renewal_chart,
}
if chart_name and chart_name in chart_builders:
chart_out = chart_builders[chart_name]()
elif show == "figure" and fname:
if "vader" in fname or "sentiment" in fname:
chart_out = build_sentiment_chart()
elif "platform_overview" in fname or "viewership" in fname or "arima" in fname:
chart_out = build_platform_streams_chart()
elif "decision" in fname or "random_forest" in fname:
chart_out = build_renewal_chart()
else:
chart_out = _empty_chart(f"No interactive chart for {fname}")
if show == "table" and fname:
fp = PY_TAB_DIR / fname
if fp.exists():
tab_out = _load_table_safe(fp)
else:
reply += f"\n\n*(Could not find table: {fname})*"
new_history = (history or []) + [
{"role": "user", "content": user_msg},
{"role": "assistant", "content": reply},
]
return new_history, "", chart_out, tab_out
def _keyword_fallback(msg: str, idx: Dict, kpis: Dict) -> Tuple[str, Dict]:
msg_lower = msg.lower()
if not idx["python"]["figures"] and not idx["python"]["tables"]:
return (
"No artifacts found yet. Please run the pipeline first (Tab 1), "
"then come back here to explore the results.",
{"show": "none"},
)
kpi_text = ""
if kpis:
total = kpis.get("total_shows", "?")
renew = kpis.get("shows_to_renew", "?")
cancel = kpis.get("shows_to_cancel", "?")
invest = kpis.get("shows_invest_more", "?")
kpi_text = (
f"Quick summary: **{total}** shows analysed โ "
f"**{renew}** to renew, **{cancel}** to cancel, **{invest}** need more investment."
)
if any(w in msg_lower for w in ["sentiment", "vader", "review", "positive", "negative"]):
return (
f"Here is the VADER sentiment analysis by renewal decision. {kpi_text}",
{"show": "figure", "chart": "sentiment"},
)
if any(w in msg_lower for w in ["arima", "forecast", "predict", "viewership", "trend", "stream"]):
return (
f"Here are the platform viewership trends and ARIMA forecasts. {kpi_text}",
{"show": "figure", "chart": "platform_streams"},
)
if any(w in msg_lower for w in ["renew", "cancel", "decision", "genre", "random forest", "model"]):
return (
f"Here is the renewal decision breakdown. {kpi_text}",
{"show": "figure", "chart": "renewal"},
)
if any(w in msg_lower for w in ["feature", "importance", "variable"]):
return (
f"Here are the most important features from the Random Forest model. {kpi_text}",
{"show": "table", "scope": "python", "filename": "feature_importances.csv"},
)
if any(w in msg_lower for w in ["recommend", "list", "show", "top", "best", "worst"]):
return (
f"Here is the full renewal recommendations table. {kpi_text}",
{"show": "table", "scope": "python", "filename": "renewal_recommendations.csv"},
)
if any(w in msg_lower for w in ["dashboard", "overview", "summary", "kpi"]):
return (
f"Dashboard overview: {kpi_text}\n\nAsk me about sentiment, viewership trends, "
"ARIMA forecasts, renewal decisions, or feature importances.",
{"show": "figure", "chart": "renewal"},
)
# Default
return (
f"I can help you explore the streaming analytics. {kpi_text}\n\n"
"Try asking about: **sentiment analysis**, **viewership trends**, **ARIMA forecasts**, "
"**renewal decisions**, **feature importances**, or **show recommendations**.",
{"show": "none"},
)
# =========================================================
# KPI CARDS
# =========================================================
def render_kpi_cards() -> str:
kpis = load_kpis()
if not kpis:
return (
''
'
๐
'
'
No data yet
'
'
'
'Run the pipeline to populate these cards.
'
'
'
)
known = {k for k, *_ in kpi_config}
for key, icon, label, colour in kpi_config:
val = kpis.get(key)
if val is None:
continue
if isinstance(val, float):
val = f"{val:,.3f}" if val < 10 else f"{val:,.1f}"
elif isinstance(val, int) and val > 999:
val = f"{val:,}"
html += card(icon, label, str(val), colour)
# Any extra keys not in config
for key, val in kpis.items():
if key not in known:
label = key.replace("_", " ").title()
if isinstance(val, (int, float)) and val > 100:
val = f"{val:,.0f}"
html += card("๐", label, str(val), "#8fa8f8")
html += "
"
return html
# =========================================================
# INTERACTIVE PLOTLY CHARTS
# =========================================================
CHART_PALETTE = ["#7c5cbf", "#2ec4a0", "#e8537a", "#e8a230", "#5e8fef",
"#c45ea8", "#3dbacc", "#a0522d", "#6aaa3a", "#d46060"]
def _styled_layout(**kwargs) -> dict:
defaults = dict(
template="plotly_white",
paper_bgcolor="rgba(255,255,255,0.95)",
plot_bgcolor="rgba(255,255,255,0.98)",
font=dict(family="system-ui, sans-serif", color="#2d1f4e", size=12),
margin=dict(l=60, r=20, t=70, b=70),
legend=dict(
orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1,
bgcolor="rgba(255,255,255,0.92)",
bordercolor="rgba(124,92,191,0.35)", borderwidth=1,
),
title=dict(font=dict(size=15, color="#4b2d8a")),
)
defaults.update(kwargs)
return defaults
def _empty_chart(title: str) -> go.Figure:
fig = go.Figure()
fig.update_layout(
title=title, height=420, template="plotly_white",
paper_bgcolor="rgba(255,255,255,0.95)",
annotations=[dict(text="Run the pipeline to generate data",
x=0.5, y=0.5, xref="paper", yref="paper", showarrow=False,
font=dict(size=14, color="rgba(124,92,191,0.5)"))],
)
return fig
def build_platform_streams_chart() -> go.Figure:
"""Monthly platform total streams from monthly_platform_totals.csv."""
path = ART_DIR / "monthly_platform_totals.csv"
if not path.exists():
return _empty_chart("Platform Streams โ run the pipeline first")
df = pd.read_csv(path)
df["month"] = pd.to_datetime(df["month"], errors="coerce")
fig = go.Figure()
fig.add_trace(go.Scatter(
x=df["month"], y=df["total_streams_k"],
name="Monthly Streams (k)",
mode="lines+markers",
fill="tozeroy",
fillcolor="rgba(124,92,191,0.12)",
line=dict(color="#7c5cbf", width=2.5),
marker=dict(size=5),
hovertemplate="%{x|%b %Y}: %{y:,.0f}k streams",
))
rolling = df["total_streams_k"].rolling(3, center=True).mean()
fig.add_trace(go.Scatter(
x=df["month"], y=rolling,
name="3-month rolling avg",
mode="lines",
line=dict(color="#e8537a", width=2, dash="dash"),
hovertemplate="Rolling avg: %{y:,.0f}k",
))
fig.update_layout(**_styled_layout(
height=420, hovermode="x unified",
title=dict(text="Total Platform Streams Over Time"),
))
fig.update_xaxes(gridcolor="rgba(124,92,191,0.15)", showgrid=True)
fig.update_yaxes(gridcolor="rgba(124,92,191,0.15)", showgrid=True, title="Streams (thousands)")
return fig
def build_sentiment_chart() -> go.Figure:
"""VADER score distribution by renewal decision from shows_final.csv."""
path = PY_TAB_DIR / "renewal_recommendations.csv"
if not path.exists():
return _empty_chart("Sentiment Analysis โ run the pipeline first")
df = pd.read_csv(path)
if "avg_vader_score" not in df.columns or "renewal_decision" not in df.columns:
return _empty_chart("Expected columns not found in renewal_recommendations.csv")
decision_order = ["Renew", "Invest More", "Cancel"]
decision_colors = {"Renew": "#2ec4a0", "Invest More": "#e8a230", "Cancel": "#e8537a"}
fig = go.Figure()
for decision in decision_order:
subset = df[df["renewal_decision"] == decision]["avg_vader_score"].dropna()
if subset.empty:
continue
fig.add_trace(go.Box(
y=subset,
name=decision,
marker_color=decision_colors.get(decision, "#888"),
boxmean="sd",
hovertemplate=f"{decision}
VADER: %{{y:.3f}}",
))
fig.update_layout(**_styled_layout(
height=440,
title=dict(text="VADER Sentiment Score by Renewal Decision"),
))
fig.update_yaxes(title="Avg VADER Compound Score", gridcolor="rgba(124,92,191,0.15)")
return fig
def build_renewal_chart() -> go.Figure:
"""Renewal decision breakdown from renewal_recommendations.csv."""
path = PY_TAB_DIR / "renewal_recommendations.csv"
if not path.exists():
return _empty_chart("Renewal Decisions โ run the pipeline first")
df = pd.read_csv(path)
if "renewal_decision" not in df.columns:
return _empty_chart("renewal_decision column not found")
counts = df["renewal_decision"].value_counts().reindex(["Renew", "Invest More", "Cancel"], fill_value=0)
colors = ["#2ec4a0", "#e8a230", "#e8537a"]
fig = go.Figure(go.Bar(
x=counts.index,
y=counts.values,
marker_color=colors,
text=counts.values,
textposition="outside",
hovertemplate="%{x}: %{y} shows",
))
fig.update_layout(**_styled_layout(
height=400, showlegend=False,
title=dict(text="Renewal Decision Breakdown"),
))
fig.update_yaxes(title="Number of Shows", gridcolor="rgba(124,92,191,0.15)")
return fig
def refresh_dashboard():
return (
render_kpi_cards(),
build_platform_streams_chart(),
build_sentiment_chart(),
build_renewal_chart(),
)
# =========================================================
# UI
# =========================================================
ensure_dirs()
def load_css() -> str:
css_path = BASE_DIR / "style.css"
return css_path.read_text(encoding="utf-8") if css_path.exists() else ""
with gr.Blocks(title="Streaming Cancellation Risk Predictor") as demo:
gr.Markdown(
"# ๐ฌ Streaming Platform โ Cancellation Risk Predictor\n"
"*Group 36 ยท ESCP ยท AIBDM 2026*",
elem_id="escp_title",
)
# ===========================================================
# TAB 1 โ Pipeline Runner
# ===========================================================
with gr.Tab("Pipeline Runner"):
gr.Markdown(
"Run the notebooks to collect IMDb data, generate synthetic streaming KPIs, "
"and train the cancellation-risk model."
)
with gr.Row():
with gr.Column(scale=1):
btn_nb1 = gr.Button("Step 1: Data Creation", variant="secondary")
with gr.Column(scale=1):
btn_nb2 = gr.Button("Step 2: Python Analysis", variant="secondary")
with gr.Row():
btn_all = gr.Button("Run Full Pipeline (Both Steps)", variant="primary")
run_log = gr.Textbox(
label="Execution Log",
lines=18,
max_lines=30,
interactive=False,
)
btn_nb1.click(run_datacreation, outputs=[run_log])
btn_nb2.click(run_pythonanalysis, outputs=[run_log])
btn_all.click(run_full_pipeline, outputs=[run_log])
# ===========================================================
# TAB 2 โ Dashboard
# ===========================================================
with gr.Tab("Dashboard"):
kpi_html = gr.HTML(value=render_kpi_cards)
refresh_btn = gr.Button("Refresh Dashboard", variant="primary")
gr.Markdown("#### Interactive Charts")
chart_streams = gr.Plot(label="Platform Streams Over Time")
chart_sentiment = gr.Plot(label="Sentiment by Renewal Decision")
chart_renewal = gr.Plot(label="Renewal Decision Breakdown")
gr.Markdown("#### Static Figures (from notebooks)")
gallery = gr.Gallery(
label="Generated Figures",
columns=2,
height=480,
object_fit="contain",
)
gr.Markdown("#### Data Tables")
table_dropdown = gr.Dropdown(
label="Select a table to view",
choices=[],
interactive=True,
)
table_display = gr.Dataframe(
label="Table Preview",
interactive=False,
)
def _on_refresh():
kpi, c1, c2, c3 = refresh_dashboard()
figs, dd, df = refresh_gallery()
return kpi, c1, c2, c3, figs, dd, df
refresh_btn.click(
_on_refresh,
outputs=[kpi_html, chart_streams, chart_sentiment, chart_renewal,
gallery, table_dropdown, table_display],
)
table_dropdown.change(
on_table_select,
inputs=[table_dropdown],
outputs=[table_display],
)
# ===========================================================
# TAB 3 โ AI Dashboard
# ===========================================================
with gr.Tab('"AI" Dashboard'):
_ai_status = (
"Connected to your **n8n workflow**." if N8N_WEBHOOK_URL
else "**LLM active.**" if LLM_ENABLED
else "Using **keyword matching**. Upgrade options: "
"set `N8N_WEBHOOK_URL` to connect your n8n workflow, "
"or set `HF_API_KEY` for direct LLM access."
)
gr.Markdown(
"### Ask questions, get interactive visualisations\n\n"
f"Type a question and the system will pick the right chart or table. {_ai_status}"
)
with gr.Row(equal_height=True):
with gr.Column(scale=1):
chatbot = gr.Chatbot(
label="Conversation",
height=380,
)
user_input = gr.Textbox(
label="Ask about your data",
placeholder="e.g. Show sentiment analysis / Which shows should be cancelled? / ARIMA forecasts",
lines=1,
)
gr.Examples(
examples=[
"Show me the platform viewership trends",
"What does the sentiment analysis show?",
"Which shows should be cancelled?",
"Show the ARIMA forecasts",
"What are the most important features?",
"Give me the full recommendations table",
],
inputs=user_input,
)
with gr.Column(scale=1):
ai_figure = gr.Plot(label="Interactive Chart")
ai_table = gr.Dataframe(label="Data Table", interactive=False)
user_input.submit(
ai_chat,
inputs=[user_input, chatbot],
outputs=[chatbot, user_input, ai_figure, ai_table],
)
demo.launch(css=load_css(), allowed_paths=[str(BASE_DIR)])