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 / "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,
kernel_name="python3",
)
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 retail analytics app focused on e-commerce return prediction and review intelligence.
The user asks questions about product reviews, return risk, customer complaints, sentiment, rating distribution, and product/category return patterns. You have access to pre-computed artifacts from a Python analysis pipeline.
AVAILABLE ARTIFACTS:
{artifacts_json}
KPI SUMMARY:
{kpis_json}
YOUR JOB:
1. Answer the user's question using the KPIs and available artifacts.
2. At the END of your response, output a JSON block fenced with ```json ... ``` using this exact shape:
{{"show": "figure"|"table"|"none", "scope": "python", "filename": "..."}}
RULES:
- If the user asks about return rate, return risk, returned products, or high-risk categories, show category_return_rate.csv if available.
- If the user asks about sentiment or reviews, show sentiment_counts_sampled.csv or sentiment_distribution.png.
- If the user asks about ratings, show rating_distribution.png or rating_distribution.csv.
- If the user asks for dashboard overview, show df_dashboard.csv.
- If no artifact is relevant, use "show": "none".
- Keep the answer concise and business-focused.
"""
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") if isinstance(directive, dict) else "none"
fname = directive.get("filename", "") if isinstance(directive, dict) else ""
chart_name = directive.get("chart", "") if isinstance(directive, dict) else ""
# Interactive chart builders keyed by name. Old aliases are included so
# the app will still work if an LLM/webhook returns an older chart name.
chart_builders = {
"sales": build_monthly_return_chart,
"returns": build_monthly_return_chart,
"monthly_returns": build_monthly_return_chart,
"sentiment": build_sentiment_chart,
"top_sellers": build_top_return_categories_chart,
"top_returns": build_top_return_categories_chart,
"rating": build_rating_chart,
"ratings": build_rating_chart,
}
if chart_name and chart_name in chart_builders:
chart_out = chart_builders[chart_name]()
elif show == "figure" and fname:
low_fname = fname.lower()
if "sentiment" in low_fname:
chart_out = build_sentiment_chart()
elif "rating" in low_fname:
chart_out = build_rating_chart()
elif "category" in low_fname or "top" in low_fname or "risk" in low_fname:
chart_out = build_top_return_categories_chart()
elif "return" in low_fname or "dashboard" in low_fname or "monthly" in low_fname:
chart_out = build_monthly_return_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]:
"""Retail return/review 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, then come back here to explore the results.",
{"show": "none"},
)
reviews_rows = kpis.get("reviews_rows", "?")
returns_rows = kpis.get("returns_rows", "?")
n_products = kpis.get("n_products", kpis.get("n_titles", "?"))
return_rate = kpis.get("estimated_return_rate", None)
if isinstance(return_rate, (int, float)):
return_rate_text = f"{return_rate:.1%}"
else:
return_rate_text = "not available"
kpi_text = (
f"Quick summary: **{reviews_rows} reviews**, **{returns_rows} return records**, "
f"and **{n_products} products/classes** analyzed. Estimated return rate: **{return_rate_text}**."
)
if any(w in msg_lower for w in ["return", "returned", "returns", "highest return", "return rate", "risk"]):
return (
f"Here are the highest return-risk products/categories. {kpi_text}",
{"show": "table", "scope": "python", "filename": "category_return_rate.csv"},
)
if any(w in msg_lower for w in ["complaint", "complaints", "problem", "issues", "review", "reviews"]):
return (
f"Here is the review intelligence summary. {kpi_text}",
{"show": "table", "scope": "python", "filename": "sentiment_counts_sampled.csv"},
)
if any(w in msg_lower for w in ["sentiment", "positive", "negative", "neutral"]):
return (
f"Here is the sentiment breakdown from customer reviews. {kpi_text}",
{"show": "figure", "chart": "sentiment"},
)
if any(w in msg_lower for w in ["rating", "ratings", "stars"]):
return (
f"Here is the rating distribution. {kpi_text}",
{"show": "figure", "scope": "python", "filename": "rating_distribution.png"},
)
if any(w in msg_lower for w in ["dashboard", "overview", "summary", "kpi"]):
return (
f"Dashboard overview: {kpi_text}\n\nAsk me about return rates, review complaints, sentiment, ratings, or high-risk products.",
{"show": "table", "scope": "python", "filename": "df_dashboard.csv"},
)
return (
f"I can help analyze e-commerce returns and review intelligence. {kpi_text}\n\n"
"Try asking about: **highest return-rate categories**, **review complaints**, "
"**sentiment**, **ratings**, 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.
'
'
'
)
used = set()
for key, icon, label, colour in kpi_config:
val = aliases.get(key)
if val is None:
continue
used.add(key)
html += card(icon, label, fmt_value(key, val), colour)
# Extra KPIs not in config
for key, val in kpis.items():
if key in used or key in {"n_titles", "n_months", "total_units_sold", "total_revenue"}:
continue
label = key.replace("_", " ").title()
html += card("đ", label, fmt_value(key, 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_monthly_return_chart() -> go.Figure:
path = PY_TAB_DIR / "df_dashboard.csv"
if not path.exists():
return _empty_chart("Monthly Return Overview â run the pipeline first")
df = pd.read_csv(path)
if df.empty:
return _empty_chart("df_dashboard.csv is empty")
date_col = next((c for c in df.columns if "month" in c.lower() or "date" in c.lower() or "time" in c.lower()), None)
val_cols = [c for c in df.columns if c != date_col and pd.api.types.is_numeric_dtype(df[c])]
if not val_cols:
# Try converting numeric-looking columns
for c in df.columns:
if c != date_col:
df[c] = pd.to_numeric(df[c], errors="coerce")
val_cols = [c for c in df.columns if c != date_col and pd.api.types.is_numeric_dtype(df[c])]
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):
y_format = ":.1%" if "rate" in col.lower() or "risk" in col.lower() else ":,.0f"
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=5),
hovertemplate=f"{col.replace('_',' ').title()}
%{{x|%b %Y}}: %{{y{y_format}}}",
))
fig.update_layout(**_styled_layout(
height=450,
hovermode="x unified",
title=dict(text="Monthly Return 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
# Backwards-compatible name used by older template code
build_sales_chart = build_monthly_return_chart
def build_rating_chart() -> go.Figure:
path = PY_TAB_DIR / "rating_distribution.csv"
if not path.exists():
return _empty_chart("Rating Distribution â run the pipeline first")
df = pd.read_csv(path)
if df.empty:
return _empty_chart("rating_distribution.csv is empty")
rating_col = next((c for c in df.columns if "rating" in c.lower()), df.columns[0])
count_col = next((c for c in df.columns if "count" in c.lower() or "number" in c.lower()), df.columns[-1])
fig = go.Figure(go.Bar(
x=df[rating_col].astype(str),
y=df[count_col],
marker_color="#7c5cbf",
hovertemplate="Rating %{x}
Reviews: %{y:,.0f}",
))
fig.update_layout(**_styled_layout(
height=420,
title=dict(text="Distribution of Customer Ratings"),
showlegend=False,
))
fig.update_xaxes(title="Rating")
fig.update_yaxes(title="Number of Reviews")
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)
if df.empty:
return _empty_chart("sentiment_counts_sampled.csv is empty")
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 sentiment_counts_sampled.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].astype(str),
x=df[col],
orientation="h",
marker_color=colors.get(col, "#888"),
hovertemplate=f"{col.title()}: %{{x:,.0f}}",
))
fig.update_layout(**_styled_layout(
height=max(420, len(df) * 30),
barmode="stack",
title=dict(text="Sentiment Distribution by Product/Class"),
))
fig.update_xaxes(title="Number of Reviews")
fig.update_yaxes(autorange="reversed")
return fig
def build_top_return_categories_chart() -> go.Figure:
path = PY_TAB_DIR / "category_return_rate.csv"
if not path.exists():
path = PY_TAB_DIR / "top_titles_by_units_sold.csv" # backwards-compatible fallback
if not path.exists():
return _empty_chart("Highest Return-Risk Categories â run the pipeline first")
df = pd.read_csv(path).head(15)
if df.empty:
return _empty_chart(f"{path.name} is empty")
category_col = next(
(c for c in df.columns if "category" in c.lower() or "class" in c.lower() or "title" in c.lower() or "product" in c.lower()),
df.columns[0],
)
value_col = next(
(c for c in df.columns if "return" in c.lower() or "risk" in c.lower() or "rate" in c.lower() or "unit" in c.lower()),
df.columns[-1],
)
df[value_col] = pd.to_numeric(df[value_col], errors="coerce")
fig = go.Figure(go.Bar(
y=df[category_col].astype(str),
x=df[value_col],
orientation="h",
marker=dict(color=df[value_col], colorscale=[[0, "#c5b4f0"], [1, "#7c5cbf"]]),
hovertemplate="%{y}
Return rate/risk: %{x:.2%}",
))
fig.update_layout(**_styled_layout(
height=max(420, len(df) * 32),
title=dict(text="Highest Return-Risk Products / Categories"),
showlegend=False,
))
fig.update_yaxes(autorange="reversed")
fig.update_xaxes(title="Return Rate / Risk Score")
return fig
# Backwards-compatible name used by older template code
build_top_sellers_chart = build_top_return_categories_chart
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 Return Overview")
chart_sentiment = gr.Plot(label="Sentiment Distribution")
chart_top = gr.Plot(label="Highest Return Risk")
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. Which products have the highest return rate? / What are the main complaints? / Sentiment analysis",
lines=1,
)
gr.Examples(
examples=[
"Which products have the highest return rate?",
"What are the main complaints in the reviews?",
"What does the sentiment look like?",
"Show me the rating distribution",
"Give me a dashboard overview",
"Which categories are highest risk?",
],
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)])