7ddb8a2ee5
Render Configuration: - render.yaml for declarative deployment - requirements-dashboard.txt (lightweight deps for cloud) - Updated .streamlit/config.toml for production - Updated app.py to auto-detect production vs local data Production Data: - Added data/production/test-07/ with 30 real call analyses - Updated .gitignore to allow data/production/ Documentation: - Added Render.com section to DEPLOYMENT.md with step-by-step guide Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
550 lines
16 KiB
Python
550 lines
16 KiB
Python
"""
|
|
CXInsights Dashboard - Main Application
|
|
Rich visualization dashboard for call analysis results.
|
|
Following Beyond Brand Identity Guidelines v1.0
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
# Add parent directory to path for imports
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
import streamlit as st
|
|
import pandas as pd
|
|
|
|
from config import COLORS, apply_custom_css
|
|
from data_loader import (
|
|
load_batch_data,
|
|
get_available_batches,
|
|
calculate_kpis,
|
|
aggregate_drivers,
|
|
)
|
|
from components import (
|
|
render_kpi_cards,
|
|
render_outcome_chart,
|
|
render_driver_analysis,
|
|
render_driver_detail,
|
|
render_call_explorer,
|
|
render_agent_performance,
|
|
render_fcr_analysis,
|
|
render_churn_risk_analysis,
|
|
render_driver_correlation_heatmap,
|
|
render_driver_outcome_heatmap,
|
|
render_rca_sankey,
|
|
render_outcome_deep_dive,
|
|
)
|
|
from exports import render_export_section
|
|
|
|
# =============================================================================
|
|
# PAGE CONFIG
|
|
# =============================================================================
|
|
|
|
st.set_page_config(
|
|
page_title="CXInsights Dashboard | Beyond",
|
|
page_icon="📊",
|
|
layout="wide",
|
|
initial_sidebar_state="expanded",
|
|
)
|
|
|
|
# Apply Beyond brand CSS
|
|
apply_custom_css()
|
|
|
|
|
|
# =============================================================================
|
|
# MAIN APP
|
|
# =============================================================================
|
|
|
|
def main():
|
|
"""Main dashboard application."""
|
|
|
|
# -------------------------------------------------------------------------
|
|
# SIDEBAR
|
|
# -------------------------------------------------------------------------
|
|
with st.sidebar:
|
|
# Logo/Brand
|
|
st.markdown(
|
|
f"""
|
|
<div style="padding: 1rem 0; margin-bottom: 1rem;">
|
|
<span style="font-size: 24px; font-weight: 700; color: {COLORS['black']};">
|
|
beyond
|
|
</span>
|
|
<sup style="font-size: 12px; color: {COLORS['blue']};">cx</sup>
|
|
<div style="font-size: 12px; color: {COLORS['grey']}; margin-top: 4px;">
|
|
CXInsights Dashboard
|
|
</div>
|
|
</div>
|
|
""",
|
|
unsafe_allow_html=True,
|
|
)
|
|
|
|
st.markdown("---")
|
|
|
|
# Batch selector
|
|
# On Render (production), use data/production; locally use data/output
|
|
base_path = Path(__file__).parent.parent / "data"
|
|
if os.environ.get("RENDER") or (base_path / "production").exists():
|
|
# Check production first (has committed data for cloud)
|
|
prod_path = base_path / "production"
|
|
if prod_path.exists() and any(prod_path.iterdir()):
|
|
data_dir = prod_path
|
|
else:
|
|
data_dir = base_path / "output"
|
|
else:
|
|
data_dir = base_path / "output"
|
|
|
|
batches = get_available_batches(data_dir)
|
|
|
|
if not batches:
|
|
st.error("No batch data found.")
|
|
st.markdown(
|
|
"Run the pipeline first:\n"
|
|
"```bash\n"
|
|
"python cli.py run <batch_id> -i <audio_dir>\n"
|
|
"```"
|
|
)
|
|
st.stop()
|
|
|
|
selected_batch = st.selectbox(
|
|
"Select Batch",
|
|
batches,
|
|
index=len(batches) - 1, # Most recent
|
|
help="Select a completed analysis batch to visualize",
|
|
)
|
|
|
|
st.markdown("---")
|
|
|
|
# Navigation
|
|
st.markdown("### Navigation")
|
|
page = st.radio(
|
|
"Section",
|
|
[
|
|
"📊 Overview",
|
|
"📈 Outcomes",
|
|
"😞 Poor CX Analysis",
|
|
"🎯 FCR Analysis",
|
|
"⚠️ Churn Risk",
|
|
"👤 Agent Performance",
|
|
"🔍 Call Explorer",
|
|
"📥 Export Insights",
|
|
],
|
|
label_visibility="collapsed",
|
|
)
|
|
|
|
st.markdown("---")
|
|
|
|
# Metadata
|
|
st.markdown(
|
|
f"""
|
|
<div style="font-size: 11px; color: {COLORS['grey']};">
|
|
<strong>Last updated:</strong><br>
|
|
{datetime.now().strftime('%Y-%m-%d %H:%M')}<br><br>
|
|
<strong>Powered by:</strong><br>
|
|
Beyond CXInsights v1.0
|
|
</div>
|
|
""",
|
|
unsafe_allow_html=True,
|
|
)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# LOAD DATA
|
|
# -------------------------------------------------------------------------
|
|
batch_path = data_dir / selected_batch
|
|
batch_data = load_batch_data(batch_path)
|
|
|
|
if batch_data is None:
|
|
st.error(f"Failed to load batch: {selected_batch}")
|
|
st.stop()
|
|
|
|
summary = batch_data["summary"]
|
|
analyses = batch_data["analyses"]
|
|
|
|
# -------------------------------------------------------------------------
|
|
# HEADER
|
|
# -------------------------------------------------------------------------
|
|
st.markdown(
|
|
f"""
|
|
<h1 style="margin-bottom: 0.25rem;">📊 CXInsights Dashboard</h1>
|
|
<p style="color: {COLORS['grey']}; margin-bottom: 2rem;">
|
|
<strong>Batch:</strong> {selected_batch} |
|
|
<strong>Calls:</strong> {summary['summary']['total_calls']} |
|
|
<strong>Generated:</strong> {summary.get('generated_at', 'N/A')[:10]}
|
|
</p>
|
|
""",
|
|
unsafe_allow_html=True,
|
|
)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# PAGE ROUTING
|
|
# -------------------------------------------------------------------------
|
|
if page == "📊 Overview":
|
|
render_overview_page(summary, analyses)
|
|
|
|
elif page == "📈 Outcomes":
|
|
render_outcomes_page(summary, analyses)
|
|
|
|
elif page == "😞 Poor CX Analysis":
|
|
render_poor_cx_page(summary, analyses)
|
|
|
|
elif page == "🎯 FCR Analysis":
|
|
render_fcr_page(summary, analyses)
|
|
|
|
elif page == "⚠️ Churn Risk":
|
|
render_churn_page(summary, analyses)
|
|
|
|
elif page == "👤 Agent Performance":
|
|
render_agent_page(analyses)
|
|
|
|
elif page == "🔍 Call Explorer":
|
|
render_call_explorer(analyses)
|
|
|
|
elif page == "📥 Export Insights":
|
|
render_export_page(summary, analyses, selected_batch)
|
|
|
|
|
|
# =============================================================================
|
|
# PAGE RENDERS
|
|
# =============================================================================
|
|
|
|
def render_overview_page(summary: dict, analyses: list[dict]):
|
|
"""Render overview page with executive summary."""
|
|
|
|
# KPI Cards
|
|
render_kpi_cards(summary, analyses)
|
|
|
|
st.markdown("---")
|
|
|
|
# Two column layout
|
|
col1, col2 = st.columns(2)
|
|
|
|
with col1:
|
|
st.markdown("### Call Outcomes Distribution")
|
|
render_outcome_chart(summary, height=350)
|
|
|
|
with col2:
|
|
st.markdown("### Top Poor CX Drivers")
|
|
render_driver_analysis(summary, "poor_cx", limit=5)
|
|
|
|
st.markdown("---")
|
|
|
|
# Second row
|
|
col1, col2 = st.columns(2)
|
|
|
|
with col1:
|
|
st.markdown("### First Call Resolution")
|
|
render_fcr_analysis(analyses, compact=True)
|
|
|
|
with col2:
|
|
st.markdown("### Churn Risk Distribution")
|
|
render_churn_risk_analysis(analyses, compact=True)
|
|
|
|
# Executive Summary Box
|
|
st.markdown("---")
|
|
st.markdown("### Executive Summary")
|
|
|
|
kpis = calculate_kpis(summary, analyses)
|
|
|
|
# Generate insights
|
|
insights = []
|
|
|
|
if kpis["poor_cx_rate"] > 30:
|
|
insights.append(
|
|
f"⚠️ **High Poor CX Rate:** {kpis['poor_cx_rate']:.1f}% of calls show "
|
|
f"customer experience issues requiring attention."
|
|
)
|
|
|
|
if kpis["churn_risk_rate"] > 20:
|
|
insights.append(
|
|
f"⚠️ **Elevated Churn Risk:** {kpis['churn_risk_rate']:.1f}% of customers "
|
|
f"show elevated churn risk signals."
|
|
)
|
|
|
|
if kpis["fcr_rate"] < 70:
|
|
insights.append(
|
|
f"📉 **FCR Below Target:** First call resolution at {kpis['fcr_rate']:.1f}% "
|
|
f"suggests process improvement opportunities."
|
|
)
|
|
|
|
top_drivers = summary.get("poor_cx", {}).get("top_drivers", [])
|
|
if top_drivers:
|
|
top = top_drivers[0]
|
|
insights.append(
|
|
f"🔍 **Top Driver:** {top['driver_code']} detected in "
|
|
f"{top['occurrences']} calls ({top.get('call_rate', 0)*100:.0f}% of total)."
|
|
)
|
|
|
|
if insights:
|
|
for insight in insights:
|
|
st.markdown(insight)
|
|
else:
|
|
st.success("✅ No critical issues detected. Performance within expected parameters.")
|
|
|
|
st.caption(
|
|
f"Source: CXInsights Analysis | Generated: {summary.get('generated_at', 'N/A')}"
|
|
)
|
|
|
|
|
|
def render_outcomes_page(summary: dict, analyses: list[dict]):
|
|
"""Render detailed outcome analysis page."""
|
|
|
|
st.markdown("## 📈 Outcome Analysis")
|
|
st.markdown(
|
|
"Understanding call outcomes helps identify resolution patterns and opportunities."
|
|
)
|
|
|
|
st.markdown("---")
|
|
|
|
col1, col2 = st.columns([2, 1])
|
|
|
|
with col1:
|
|
render_outcome_chart(summary, height=450)
|
|
|
|
with col2:
|
|
st.markdown("### Outcome Breakdown")
|
|
outcomes = summary.get("outcomes", {})
|
|
total = sum(outcomes.values())
|
|
|
|
for outcome, count in sorted(outcomes.items(), key=lambda x: -x[1]):
|
|
pct = (count / total * 100) if total > 0 else 0
|
|
st.metric(
|
|
label=outcome,
|
|
value=f"{count}",
|
|
delta=f"{pct:.1f}%",
|
|
)
|
|
|
|
st.markdown("---")
|
|
|
|
# Calls by outcome table
|
|
st.markdown("### Calls by Outcome")
|
|
|
|
outcome_filter = st.multiselect(
|
|
"Filter outcomes",
|
|
list(summary.get("outcomes", {}).keys()),
|
|
default=list(summary.get("outcomes", {}).keys()),
|
|
)
|
|
|
|
filtered = [a for a in analyses if a.get("outcome") in outcome_filter]
|
|
|
|
if filtered:
|
|
df = pd.DataFrame([
|
|
{
|
|
"Call ID": a["call_id"],
|
|
"Outcome": a["outcome"],
|
|
"FCR Status": a.get("fcr_status", "N/A"),
|
|
"Churn Risk": a.get("churn_risk", "N/A"),
|
|
"Agent": a.get("agent_classification", "N/A"),
|
|
"CX Issues": len(a.get("poor_cx_drivers", [])),
|
|
}
|
|
for a in filtered
|
|
])
|
|
st.dataframe(df, use_container_width=True, hide_index=True)
|
|
else:
|
|
st.info("No calls match the selected filters.")
|
|
|
|
# ---------------------------------------------------------------------
|
|
# DEEP DIVE SECTION
|
|
# ---------------------------------------------------------------------
|
|
st.markdown("---")
|
|
st.markdown("## Deep Dive: Outcome Analysis")
|
|
|
|
outcomes_list = list(summary.get("outcomes", {}).keys())
|
|
if outcomes_list:
|
|
# Default to the most problematic outcome (not RESOLVED/POSITIVE)
|
|
problematic = [o for o in outcomes_list if "UNRESOLVED" in o or "COMPLAINT" in o]
|
|
default_idx = outcomes_list.index(problematic[0]) if problematic else 0
|
|
|
|
selected_outcome = st.selectbox(
|
|
"Select an outcome to analyze in depth",
|
|
outcomes_list,
|
|
index=default_idx,
|
|
help="Choose an outcome to see root causes, driver correlation, and duration analysis.",
|
|
)
|
|
|
|
render_outcome_deep_dive(analyses, selected_outcome)
|
|
|
|
|
|
def render_poor_cx_page(summary: dict, analyses: list[dict]):
|
|
"""Render detailed Poor CX analysis page."""
|
|
|
|
st.markdown("## 😞 Poor CX Driver Analysis")
|
|
st.markdown(
|
|
"Root cause analysis of customer experience issues detected across calls."
|
|
)
|
|
|
|
st.markdown("---")
|
|
|
|
# Summary metrics
|
|
poor_cx_data = summary.get("poor_cx", {})
|
|
total_drivers = poor_cx_data.get("total_drivers_found", 0)
|
|
unique_drivers = len(poor_cx_data.get("top_drivers", []))
|
|
|
|
col1, col2 = st.columns(2)
|
|
with col1:
|
|
st.metric("Total Driver Instances", total_drivers)
|
|
with col2:
|
|
st.metric("Unique Driver Types", unique_drivers)
|
|
|
|
st.markdown("---")
|
|
|
|
# RCA Sankey Diagram
|
|
st.markdown("### Root Cause Analysis Flow")
|
|
st.markdown(
|
|
"Visual flow showing how Poor CX drivers lead to outcomes and churn risk. "
|
|
"Wider bands indicate more frequent paths."
|
|
)
|
|
render_rca_sankey(analyses)
|
|
|
|
st.markdown("---")
|
|
|
|
# Driver chart
|
|
st.markdown("### Driver Frequency")
|
|
render_driver_analysis(summary, "poor_cx", limit=None)
|
|
|
|
st.markdown("---")
|
|
|
|
# Correlation heatmaps
|
|
st.markdown("### Driver Correlation Analysis")
|
|
st.markdown(
|
|
"Identify patterns where certain drivers frequently appear together "
|
|
"(e.g., 'LONG_WAIT' always with 'POOR_EMPATHY')."
|
|
)
|
|
|
|
tab1, tab2 = st.tabs(["Driver Co-occurrence", "Driver by Outcome"])
|
|
|
|
with tab1:
|
|
render_driver_correlation_heatmap(analyses, "poor_cx_drivers")
|
|
|
|
with tab2:
|
|
render_driver_outcome_heatmap(analyses)
|
|
|
|
st.markdown("---")
|
|
|
|
# Detailed evidence explorer
|
|
st.markdown("### Driver Evidence Explorer")
|
|
render_driver_detail(analyses, "poor_cx_drivers")
|
|
|
|
|
|
def render_fcr_page(summary: dict, analyses: list[dict]):
|
|
"""Render FCR analysis page."""
|
|
|
|
st.markdown("## 🎯 First Call Resolution Analysis")
|
|
st.markdown(
|
|
"Analyzing resolution efficiency and identifying callbacks drivers."
|
|
)
|
|
|
|
st.markdown("---")
|
|
|
|
render_fcr_analysis(analyses, compact=False)
|
|
|
|
st.markdown("---")
|
|
|
|
# FCR failure drivers
|
|
st.markdown("### FCR Failure Root Causes")
|
|
|
|
fcr_drivers = aggregate_drivers(analyses, "fcr_failure_drivers")
|
|
|
|
if fcr_drivers:
|
|
df = pd.DataFrame([
|
|
{
|
|
"Driver": code,
|
|
"Instances": data["count"],
|
|
"Calls Affected": data["call_count"],
|
|
"Avg Confidence": f"{data['avg_confidence']:.0%}",
|
|
}
|
|
for code, data in sorted(fcr_drivers.items(), key=lambda x: -x[1]["count"])
|
|
])
|
|
st.dataframe(df, use_container_width=True, hide_index=True)
|
|
|
|
st.markdown("---")
|
|
|
|
# Evidence
|
|
st.markdown("### Evidence & Recommendations")
|
|
render_driver_detail(analyses, "fcr_failure_drivers")
|
|
else:
|
|
st.success("✅ No FCR failures detected. Excellent first-call resolution!")
|
|
|
|
|
|
def render_churn_page(summary: dict, analyses: list[dict]):
|
|
"""Render churn risk analysis page."""
|
|
|
|
st.markdown("## ⚠️ Churn Risk Analysis")
|
|
st.markdown(
|
|
"Identifying customers at risk of churning based on conversation signals."
|
|
)
|
|
|
|
st.markdown("---")
|
|
|
|
render_churn_risk_analysis(analyses, compact=False)
|
|
|
|
st.markdown("---")
|
|
|
|
# High risk calls
|
|
st.markdown("### High Risk Customer Calls")
|
|
|
|
high_risk = [
|
|
a for a in analyses
|
|
if a.get("churn_risk") in ["HIGH", "AT_RISK"]
|
|
]
|
|
|
|
if high_risk:
|
|
st.warning(
|
|
f"⚠️ {len(high_risk)} calls show elevated churn risk requiring follow-up."
|
|
)
|
|
|
|
for analysis in high_risk:
|
|
with st.expander(
|
|
f"📞 {analysis['call_id']} — Risk: {analysis.get('churn_risk', 'N/A')}"
|
|
):
|
|
st.markdown(f"**Outcome:** {analysis.get('outcome', 'N/A')}")
|
|
|
|
drivers = analysis.get("churn_risk_drivers", [])
|
|
if drivers:
|
|
st.markdown("**Risk Drivers:**")
|
|
for d in drivers:
|
|
st.markdown(
|
|
f"- **{d.get('driver_code')}** "
|
|
f"({d.get('confidence', 0):.0%}): "
|
|
f"{d.get('reasoning', 'N/A')}"
|
|
)
|
|
|
|
if d.get("corrective_action"):
|
|
st.success(f"Action: {d['corrective_action']}")
|
|
else:
|
|
st.success("✅ No high churn risk calls detected.")
|
|
|
|
|
|
def render_agent_page(analyses: list[dict]):
|
|
"""Render agent performance page."""
|
|
|
|
st.markdown("## 👤 Agent Performance Analysis")
|
|
st.markdown(
|
|
"Evaluating agent skills and identifying coaching opportunities."
|
|
)
|
|
|
|
st.markdown("---")
|
|
|
|
render_agent_performance(analyses)
|
|
|
|
|
|
def render_export_page(summary: dict, analyses: list[dict], batch_id: str):
|
|
"""Render export insights page."""
|
|
|
|
st.markdown("## 📥 Export Insights")
|
|
st.markdown(
|
|
"Download analysis results in multiple formats for reporting and integration."
|
|
)
|
|
|
|
st.markdown("---")
|
|
|
|
render_export_section(summary, analyses, batch_id)
|
|
|
|
|
|
# =============================================================================
|
|
# RUN
|
|
# =============================================================================
|
|
|
|
if __name__ == "__main__":
|
|
main()
|