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
from plotly.subplots import make_subplots
# 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 / "py" / "figures"
PY_TAB_DIR = ART_DIR / "py" / "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 BASE_DIR.glob("*.csv")]
return f"OK {log}\n\nCSVs now in /app:\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 (web scraping + synthetic data)")
logs.append("=" * 50)
logs.append(run_datacreation())
logs.append("")
logs.append("=" * 50)
logs.append("STEP 2/2: Python Analysis (sentiment, ARIMA, dashboard)")
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():
"""Called when user clicks Refresh on Gallery tab."""
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 book-sales analytics app.
The user asks questions or requests about their data. You have access to pre-computed
artifacts from a Python analysis pipeline.
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 sales trends or forecasting by title, show sales_trends or arima figures.
- If the user asks about sentiment, show sentiment figure or sentiment_counts table.
- If the user asks about forecast accuracy or ARIMA, show arima figures.
- If the user asks about top sellers, show top_titles_by_units_sold.csv.
- If the user asks a general data question, pick the most relevant artifact.
- 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:
"""Strip the JSON directive block from the displayed response."""
return JSON_BLOCK_RE.sub("", text).strip()
def _n8n_call(msg: str) -> Tuple[str, Dict]:
"""Call the student's n8n webhook and return (reply, directive)."""
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):
"""Chat function for the AI Dashboard tab."""
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", "")
# Interactive chart builders keyed by name
chart_builders = {
"sales": build_sales_chart,
"sentiment": build_sentiment_chart,
"top_sellers": build_top_sellers_chart,
}
if chart_name and chart_name in chart_builders:
chart_out = chart_builders[chart_name]()
elif show == "figure" and fname:
# Fallback: try to match filename to a chart builder
if "sales_trend" in fname:
chart_out = build_sales_chart()
elif "sentiment" in fname:
chart_out = build_sentiment_chart()
elif "arima" in fname or "forecast" in fname:
chart_out = build_sales_chart() # closest interactive equivalent
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]:
"""Simple keyword matcher when LLM is unavailable."""
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_units_sold", 0)
kpi_text = (
f"Quick summary: **{kpis.get('n_titles', '?')}** book titles across "
f"**{kpis.get('n_months', '?')}** months, with **{total:,.0f}** total units sold."
)
if any(w in msg_lower for w in ["trend", "sales trend", "monthly sale"]):
return (
f"Here are the sales trends. {kpi_text}",
{"show": "figure", "chart": "sales"},
)
if any(w in msg_lower for w in ["sentiment", "review", "positive", "negative"]):
return (
f"Here is the sentiment distribution across sampled book titles. {kpi_text}",
{"show": "figure", "chart": "sentiment"},
)
if any(w in msg_lower for w in ["arima", "forecast", "predict"]):
return (
f"Here are the sales trends and forecasts. {kpi_text}",
{"show": "figure", "chart": "sales"},
)
if any(w in msg_lower for w in ["top", "best sell", "popular", "rank"]):
return (
f"Here are the top-selling titles by units sold. {kpi_text}",
{"show": "table", "scope": "python", "filename": "top_titles_by_units_sold.csv"},
)
if any(w in msg_lower for w in ["price", "pricing", "decision"]):
return (
f"Here are the pricing decisions. {kpi_text}",
{"show": "table", "scope": "python", "filename": "pricing_decisions.csv"},
)
if any(w in msg_lower for w in ["dashboard", "overview", "summary", "kpi"]):
return (
f"Dashboard overview: {kpi_text}\n\nAsk me about sales trends, sentiment, forecasts, "
"pricing, or top sellers to see specific visualizations.",
{"show": "table", "scope": "python", "filename": "df_dashboard.csv"},
)
# Default
return (
f"I can show you various analyses. {kpi_text}\n\n"
"Try asking about: **sales trends**, **sentiment**, **ARIMA forecasts**, "
"**pricing decisions**, **top sellers**, or **dashboard overview**.",
{"show": "none"},
)
# =========================================================
# KPI CARDS (BubbleBusters style)
# =========================================================
def render_kpi_cards() -> str:
kpis = load_kpis()
if not kpis:
return (
''
'
📊
'
'
No data yet
'
'
'
'Run the pipeline to populate these cards.
'
'
'
)
for key, icon, label, colour in kpi_config:
val = kpis.get(key)
if val is None:
continue
if isinstance(val, (int, float)) and val > 100:
val = f"{val:,.0f}"
html += card(icon, label, str(val), colour)
# Extra KPIs not in config
known = {k for k, *_ in kpi_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 (BubbleBusters style)
# =========================================================
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_sales_chart() -> go.Figure:
path = PY_TAB_DIR / "df_dashboard.csv"
if not path.exists():
return _empty_chart("Sales Trends — run the pipeline first")
df = pd.read_csv(path)
date_col = next((c for c in df.columns if "month" in c.lower() or "date" in c.lower()), None)
val_cols = [c for c in df.columns if c != date_col and df[c].dtype in ("float64", "int64")]
if not date_col or not val_cols:
return _empty_chart("Could not auto-detect columns in df_dashboard.csv")
df[date_col] = pd.to_datetime(df[date_col], errors="coerce")
fig = go.Figure()
for i, col in enumerate(val_cols):
fig.add_trace(go.Scatter(
x=df[date_col], y=df[col], name=col.replace("_", " ").title(),
mode="lines+markers", line=dict(color=CHART_PALETTE[i % len(CHART_PALETTE)], width=2),
marker=dict(size=4),
hovertemplate=f"{col.replace('_',' ').title()}
%{{x|%b %Y}}: %{{y:,.0f}}",
))
fig.update_layout(**_styled_layout(height=450, hovermode="x unified",
title=dict(text="Monthly Overview")))
fig.update_xaxes(gridcolor="rgba(124,92,191,0.15)", showgrid=True)
fig.update_yaxes(gridcolor="rgba(124,92,191,0.15)", showgrid=True)
return fig
def build_sentiment_chart() -> go.Figure:
path = PY_TAB_DIR / "sentiment_counts_sampled.csv"
if not path.exists():
return _empty_chart("Sentiment Distribution — run the pipeline first")
df = pd.read_csv(path)
title_col = df.columns[0]
sent_cols = [c for c in ["negative", "neutral", "positive"] if c in df.columns]
if not sent_cols:
return _empty_chart("No sentiment columns found in CSV")
colors = {"negative": "#e8537a", "neutral": "#5e8fef", "positive": "#2ec4a0"}
fig = go.Figure()
for col in sent_cols:
fig.add_trace(go.Bar(
name=col.title(), y=df[title_col], x=df[col],
orientation="h", marker_color=colors.get(col, "#888"),
hovertemplate=f"{col.title()}: %{{x}}",
))
fig.update_layout(**_styled_layout(
height=max(400, len(df) * 28), barmode="stack",
title=dict(text="Sentiment Distribution by Product"),
))
fig.update_xaxes(title="Number of Reviews")
fig.update_yaxes(autorange="reversed")
return fig
def build_top_sellers_chart() -> go.Figure:
path = PY_TAB_DIR / "top_titles_by_units_sold.csv"
if not path.exists():
return _empty_chart("Top Sellers — run the pipeline first")
df = pd.read_csv(path).head(15)
title_col = next((c for c in df.columns if "title" in c.lower()), df.columns[0])
val_col = next((c for c in df.columns if "unit" in c.lower() or "sold" in c.lower()), df.columns[-1])
fig = go.Figure(go.Bar(
y=df[title_col], x=df[val_col], orientation="h",
marker=dict(color=df[val_col], colorscale=[[0, "#c5b4f0"], [1, "#7c5cbf"]]),
hovertemplate="%{y}
Units: %{x:,.0f}",
))
fig.update_layout(**_styled_layout(
height=max(400, len(df) * 30),
title=dict(text="Top Products by Demand"), showlegend=False,
))
fig.update_yaxes(autorange="reversed")
fig.update_xaxes(title="Total Units Sold")
return fig
# ---- Data loaders (cached at import) -------------------------------------
def _load_lookup_data():
"""Load Camille's analyzed Amazon dataset + time-series, cached on first call."""
main_path = BASE_DIR / "final_analyzed_amazon_dataset.csv"
ts_path = BASE_DIR / "synthetic_historical_sales_prices.csv"
if not main_path.exists() or not ts_path.exists():
return None, None
df = pd.read_csv(main_path)
ts = pd.read_csv(ts_path)
ts["date"] = pd.to_datetime(ts["date"])
# Group rare categories under "Other" — long tail is 6 categories with 1-2 products
cat_counts = df["main_category"].value_counts()
main_cats = cat_counts[cat_counts >= 10].index.tolist()
df["display_category"] = df["main_category"].where(
df["main_category"].isin(main_cats), "Other"
)
return df, ts
_LOOKUP_DF, _LOOKUP_TS = _load_lookup_data()
# ---- Helpers -------------------------------------------------------------
def _short_name(name, n=70):
s = str(name)
return s if len(s) <= n else s[: n - 1] + "…"
def _lookup_categories():
if _LOOKUP_DF is None:
return ["All categories"]
return ["All categories"] + sorted(_LOOKUP_DF["display_category"].unique().tolist())
def _lookup_products(category):
if _LOOKUP_DF is None:
return []
sub = _LOOKUP_DF if category == "All categories" else _LOOKUP_DF[_LOOKUP_DF["display_category"] == category]
return [f"{_short_name(r['product_name'])} | {r['product_id']}" for _, r in sub.iterrows()]
def _parse_pid(choice):
if not choice or "|" not in choice:
return None
return choice.split("|")[-1].strip()
# ---- Recommendation badge (matches BubbleBusters glass-morphism style) ----
REC_STYLES = {
"increase_price": ("#2ec4a0", "↑ INCREASE PRICE",
"Demand signals support a price increase"),
"maintain_price": ("#7c5cbf", "→ MAINTAIN PRICE",
"Current pricing is well-aligned with demand"),
"decrease_price": ("#e8537a", "↓ DECREASE PRICE",
"Demand softness suggests a price reduction"),
}
def _render_lookup_recommendation(rec):
color, label, sub = REC_STYLES.get(rec, ("#9d8fc4", "—", ""))
return f"""
"""
def _render_lookup_kpi_cards(row):
cards = [
("Current price", f"₹{row['discounted_price']:,.0f}",
f"List ₹{row['actual_price']:,.0f}", "#a48de8"),
("Competitor price", f"₹{row['competitor_price']:,.0f}",
f"{((row['discounted_price'] - row['competitor_price']) / row['competitor_price'] * 100):+.1f}% vs us", "#7aa6f8"),
("Customer rating", f"★ {row['rating']:.1f}",
f"{int(row['rating_count']):,} reviews", "#6ee7c7"),
("Monthly sales", f"{int(row['monthly_sales']):,}",
f"Demand idx {row['demand_index']:.0f}", "#3dcba8"),
("Sentiment", f"{row['sentiment_score']:+.2f}",
str(row['sentiment_label']).title(), "#e8a230"),
("Customer segment", str(row['customer_segment']).replace('_', ' ').title(),
f"Return rate {row['return_rate']*100:.1f}%", "#c45ea8"),
]
html = (
''
)
for label, value, sub, accent in cards:
html += f"""
"""
html += "
Customer voice
{text}{truncated}
"""
# ---- Charts (use existing _styled_layout + CHART_PALETTE) ----------------
def build_lookup_history(product_id):
"""18-month history: sales (left axis) + price (right axis)."""
if _LOOKUP_TS is None:
return _empty_chart("Place CSVs at Space root to enable Product Lookup")
sub = _LOOKUP_TS[_LOOKUP_TS["product_id"] == product_id].sort_values("date")
if sub.empty:
return _empty_chart("No history for this product")
fig = make_subplots(specs=[[{"secondary_y": True}]])
fig.add_trace(
go.Scatter(
x=sub["date"], y=sub["historical_sales"], name="Sales (units)",
mode="lines+markers", line=dict(color="#7c5cbf", width=2.5),
marker=dict(size=5),
hovertemplate="Sales: %{y:,.0f}
%{x|%b %Y}",
),
secondary_y=False,
)
fig.add_trace(
go.Scatter(
x=sub["date"], y=sub["historical_price"], name="Price (₹)",
mode="lines+markers", line=dict(color="#e8a230", width=2, dash="dot"),
marker=dict(size=4),
hovertemplate="Price: ₹%{y:,.0f}
%{x|%b %Y}",
),
secondary_y=True,
)
fig.update_layout(**_styled_layout(
height=400, hovermode="x unified",
title=dict(text="18-month history: sales and price"),
))
fig.update_xaxes(gridcolor="rgba(124,92,191,0.15)")
fig.update_yaxes(title_text="Sales (units)", secondary_y=False,
gridcolor="rgba(124,92,191,0.15)")
fig.update_yaxes(title_text="Price (₹)", secondary_y=True, showgrid=False)
return fig
def build_lookup_forecast(product_id):
"""6-month sales projection with confidence band (trend-based)."""
if _LOOKUP_TS is None:
return _empty_chart("Place CSVs at Space root to enable Product Lookup")
sub = _LOOKUP_TS[_LOOKUP_TS["product_id"] == product_id].sort_values("date").copy()
if sub.empty or len(sub) < 6:
return _empty_chart("Not enough history to forecast")
recent = sub.tail(6)["historical_sales"].values
trend = (recent[-1] - recent[0]) / 5
last_val = recent[-1]
last_date = sub["date"].iloc[-1]
future_dates = pd.date_range(last_date, periods=7, freq="ME")[1:]
future_vals = [max(0, last_val + trend * (i + 1)) for i in range(6)]
std = sub["historical_sales"].tail(12).std()
upper = [v + std for v in future_vals]
lower = [max(0, v - std) for v in future_vals]
fig = go.Figure()
fig.add_trace(go.Scatter(
x=sub["date"], y=sub["historical_sales"], name="Historical",
mode="lines", line=dict(color="#7c5cbf", width=2.5),
hovertemplate="Historical: %{y:,.0f}",
))
fig.add_trace(go.Scatter(x=future_dates, y=upper, mode="lines",
line=dict(width=0), showlegend=False, hoverinfo="skip"))
fig.add_trace(go.Scatter(
x=future_dates, y=lower, mode="lines", fill="tonexty",
fillcolor="rgba(46,196,160,0.18)", line=dict(width=0),
name="Forecast confidence", hoverinfo="skip",
))
fig.add_trace(go.Scatter(
x=future_dates, y=future_vals, name="Forecast",
mode="lines+markers", line=dict(color="#2ec4a0", width=2.5, dash="dash"),
marker=dict(size=6),
hovertemplate="Forecast: %{y:,.0f}",
))
fig.update_layout(**_styled_layout(
height=400, hovermode="x unified",
title=dict(text="6-month sales forecast"),
))
fig.update_xaxes(gridcolor="rgba(124,92,191,0.15)")
fig.update_yaxes(title_text="Sales (units)", gridcolor="rgba(124,92,191,0.15)")
return fig
# ---- Update callback (called on dropdown changes) ------------------------
def update_product_lookup(category, product_choice):
if _LOOKUP_DF is None:
msg = (
''
'
⚙️
'
'
Data not loaded
'
'
'
'Place final_analyzed_amazon_dataset.csv and '
'synthetic_historical_sales_prices.csv at the Space root.
'
'Select a product to see details
')
return empty, "", "", _empty_chart(""), _empty_chart("")
row = _LOOKUP_DF[_LOOKUP_DF["product_id"] == pid].iloc[0]
return (
_render_lookup_recommendation(row["pricing_recommendation"]),
_render_lookup_kpi_cards(row),
_render_lookup_review(row),
build_lookup_history(pid),
build_lookup_forecast(pid),
)
def update_product_choices(category):
"""When category changes, refresh the product dropdown."""
products = _lookup_products(category)
return gr.update(
choices=products,
value=products[0] if products else None,
)
# ============================================================================
def refresh_dashboard():
return render_kpi_cards(), build_sales_chart(), build_sentiment_chart(), build_top_sellers_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="AIBDM 2026 Workshop App") as demo:
gr.Markdown(
"# SE21 App Template\n"
"*This is an app template for SE21 students*",
elem_id="escp_title",
)
# ===========================================================
# TAB 1 -- Pipeline Runner
# ===========================================================
with gr.Tab("Pipeline Runner"):
gr.Markdown()
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 (KPIs + Interactive Charts + Gallery)
# ===========================================================
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_sales = gr.Plot(label="Monthly Overview")
chart_sentiment = gr.Plot(label="Sentiment Distribution")
chart_top = gr.Plot(label="Top Products")
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_sales, chart_sentiment, chart_top,
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 interactive 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 me sales trends / Which products perform best? / Customer sentiment analysis",
lines=1,
)
gr.Examples(
examples=[
"Show me the sales trends",
"What does customer sentiment look like across products?",
"Which products perform best?",
"Show the sales forecast for top products",
"What pricing decisions are recommended based on demand and sentiment?",
"Give me a dashboard overview",
],
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],
)
with gr.Tab("Product Lookup"):
gr.Markdown(
"### Single-product pricing deep dive\n"
"Pick any product to see its pricing recommendation, key metrics, "
"customer voice, 18-month history, and 6-month forecast."
)
with gr.Row():
with gr.Column(scale=1):
lookup_cat = gr.Dropdown(
choices=_lookup_categories(),
value="All categories",
label="Filter by category",
)
lookup_prod = gr.Dropdown(
choices=_lookup_products("All categories"),
value=(
_lookup_products("All categories")[0]
if _lookup_products("All categories") else None
),
label="Select a product",
)
with gr.Column(scale=2):
lookup_rec = gr.HTML()
lookup_kpis = gr.HTML()
lookup_review = gr.HTML()
with gr.Row():
lookup_history = gr.Plot()
lookup_forecast = gr.Plot()
lookup_cat.change(
fn=update_product_choices,
inputs=lookup_cat,
outputs=lookup_prod,
)
lookup_prod.change(
fn=update_product_lookup,
inputs=[lookup_cat, lookup_prod],
outputs=[lookup_rec, lookup_kpis, lookup_review,
lookup_history, lookup_forecast],
)
demo.load(
fn=update_product_lookup,
inputs=[lookup_cat, lookup_prod],
outputs=[lookup_rec, lookup_kpis, lookup_review,
lookup_history, lookup_forecast],
)
demo.launch(css=load_css(), allowed_paths=[str(BASE_DIR)])