streamlit_app.py

# MIT License
# Copyright (c) 2026 Sai Venkata Ganesh Bandaluppi

import os
from typing import Any

import httpx
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
import streamlit as st

API_BASE_URL = os.environ.get("API_BASE_URL", "http://localhost:8008")
REQUEST_TIMEOUT = 15.0

st.set_page_config(
    page_title="Audio Intelligence Pipeline",
    page_icon="mic",
    layout="wide",
    initial_sidebar_state="expanded",
)

_NAVY = "#0D1B2A"
_GOLD = "#FFD700"

st.markdown(
    f"""
    <style>
    .stApp {{ background-color: {_NAVY}; color: white; }}
    .stSidebar {{ background-color: #0a1520; }}
    .metric-card {{
        background-color: #122030;
        border: 1px solid {_GOLD}33;
        border-radius: 8px;
        padding: 16px;
        margin: 4px;
    }}
    h1, h2, h3 {{ color: {_GOLD}; }}
    .stButton > button {{
        background-color: {_GOLD};
        color: {_NAVY};
        font-weight: bold;
        border: none;
    }}
    </style>
    """,
    unsafe_allow_html=True,
)

if "token" not in st.session_state:
    st.session_state["token"] = None
if "username" not in st.session_state:
    st.session_state["username"] = None


def _headers() -> dict[str, str]:
    return {"Authorization": f"Bearer {st.session_state['token']}"}


def _api_get(path: str) -> Any | None:
    try:
        with httpx.Client(timeout=REQUEST_TIMEOUT) as client:
            resp = client.get(f"{API_BASE_URL}{path}", headers=_headers())
            if resp.status_code == 200:
                return resp.json()
            st.error(f"API error {resp.status_code}: {resp.json().get('detail', 'Unknown error')}")
            return None
    except httpx.RequestError as exc:
        st.error(f"Could not reach API: {exc}")
        return None


def _api_post(path: str, **kwargs) -> Any | None:
    try:
        with httpx.Client(timeout=REQUEST_TIMEOUT) as client:
            resp = client.post(f"{API_BASE_URL}{path}", headers=_headers(), **kwargs)
            if resp.status_code in (200, 201):
                return resp.json()
            st.error(f"API error {resp.status_code}: {resp.json().get('detail', 'Unknown error')}")
            return None
    except httpx.RequestError as exc:
        st.error(f"Could not reach API: {exc}")
        return None


def _api_delete(path: str) -> bool:
    try:
        with httpx.Client(timeout=REQUEST_TIMEOUT) as client:
            resp = client.delete(f"{API_BASE_URL}{path}", headers=_headers())
            return resp.status_code == 204
    except httpx.RequestError as exc:
        st.error(f"Could not reach API: {exc}")
        return False


def render_login() -> None:
    st.title("Audio Intelligence Pipeline")
    st.subheader("Sign in to continue")

    tab_login, tab_register = st.tabs(["Login", "Register"])

    with tab_login:
        with st.form("login_form"):
            username = st.text_input("Username")
            password = st.text_input("Password", type="password")
            submitted = st.form_submit_button("Login")
        if submitted:
            try:
                with httpx.Client(timeout=REQUEST_TIMEOUT) as client:
                    resp = client.post(
                        f"{API_BASE_URL}/auth/login",
                        json={"username": username, "password": password},
                    )
                if resp.status_code == 200:
                    st.session_state["token"] = resp.json()["access_token"]
                    st.session_state["username"] = username
                    st.rerun()
                else:
                    st.error("Invalid credentials. Please try again.")
            except httpx.RequestError as exc:
                st.error(f"Could not reach API: {exc}")

    with tab_register:
        with st.form("register_form"):
            new_username = st.text_input("Username", key="reg_username")
            new_email = st.text_input("Email", key="reg_email")
            new_password = st.text_input("Password", type="password", key="reg_password")
            reg_submitted = st.form_submit_button("Create Account")
        if reg_submitted:
            try:
                with httpx.Client(timeout=REQUEST_TIMEOUT) as client:
                    resp = client.post(
                        f"{API_BASE_URL}/auth/register",
                        json={"username": new_username, "email": new_email, "password": new_password},
                    )
                if resp.status_code == 201:
                    st.success("Account created. Please log in.")
                else:
                    detail = resp.json().get("detail", "Registration failed")
                    st.error(detail)
            except httpx.RequestError as exc:
                st.error(f"Could not reach API: {exc}")


def page_process_call() -> None:
    st.header("Process Call")
    st.write("Submit a call transcript or upload an audio recording for full AI analysis.")

    call_type = st.selectbox(
        "Call Type",
        options=["billing", "technical", "sales", "complaint", "general"],
    )

    input_method = st.radio("Input Method", ["Paste Transcript", "Upload Audio File"], horizontal=True)

    if input_method == "Paste Transcript":
        transcript = st.text_area("Call Transcript", height=250, placeholder="Paste the full call transcript here...")
        if st.button("Analyse Call", disabled=not transcript.strip()):
            with st.spinner("Running intelligence pipeline..."):
                result = _api_post(
                    "/calls/process",
                    data={"call_type": call_type, "transcript": transcript},
                )
            if result:
                _display_call_result(result)
    else:
        audio_file = st.file_uploader("Upload Audio File", type=["wav", "mp3", "m4a"])
        if st.button("Analyse Call", disabled=audio_file is None) and audio_file:
            with st.spinner("Transcribing and analysing..."):
                result = _api_post(
                    "/calls/process",
                    data={"call_type": call_type},
                    files={"audio_file": (audio_file.name, audio_file.getvalue(), audio_file.type)},
                )
            if result:
                _display_call_result(result)


def _display_call_result(result: dict[str, Any]) -> None:
    st.success("Analysis complete")
    st.divider()

    col1, col2, col3, col4 = st.columns(4)
    with col1:
        st.metric("Sentiment Score", f"{result['sentiment_score']:.1f} / 10")
    with col2:
        st.metric("Compliance Score", f"{result['compliance_score']:.1f} / 100")
    with col3:
        st.metric("Violations", result["violations_count"])
    with col4:
        st.metric("Escalation", "Yes" if result["escalation_detected"] else "No")

    col_left, col_right = st.columns(2)
    with col_left:
        st.subheader("Trajectory")
        traj = result["sentiment_trajectory"].capitalize()
        st.write(traj)
    with col_right:
        st.subheader("Status")
        st.write(result["status"].capitalize())

    if result["violations_detail"]:
        st.subheader("Compliance Violations")
        for v in result["violations_detail"]:
            severity = v.get("severity", "unknown")
            col = {"critical": "red", "high": "orange", "medium": "yellow", "low": "blue"}.get(severity, "gray")
            st.markdown(f":{col}[**{severity.upper()}**] {v.get('description', 'No description')}")

    st.subheader("Coaching Recommendations")
    st.write(result["coaching_recommendations"])

    st.subheader("Trend Insights")
    st.write(result["trend_insights"])

    with st.expander("Full Transcript"):
        st.text(result["transcript"])


def page_call_library() -> None:
    st.header("Call Library")

    col_filter, col_esc = st.columns(2)
    with col_filter:
        type_filter = st.selectbox(
            "Filter by Call Type",
            ["All", "billing", "technical", "sales", "complaint", "general"],
        )
    with col_esc:
        esc_filter = st.selectbox("Escalation", ["All", "Escalated Only", "No Escalation"])

    path = "/calls/" if type_filter == "All" else f"/calls/?call_type={type_filter}"
    records = _api_get(path)
    if records is None:
        return
    if not records:
        st.info("No calls found. Process your first call to get started.")
        return

    if esc_filter == "Escalated Only":
        records = [r for r in records if r["escalation_detected"]]
    elif esc_filter == "No Escalation":
        records = [r for r in records if not r["escalation_detected"]]

    df = pd.DataFrame([
        {
            "ID": r["id"][:8] + "...",
            "Full ID": r["id"],
            "Call Type": r["call_type"].capitalize(),
            "Sentiment": f"{r['sentiment_score']:.1f}",
            "Compliance": f"{r['compliance_score']:.1f}",
            "Trajectory": r["sentiment_trajectory"].capitalize(),
            "Escalated": "Yes" if r["escalation_detected"] else "No",
            "Violations": r["violations_count"],
            "Date": r["created_at"][:10],
        }
        for r in records
    ])

    selected_rows = st.dataframe(
        df.drop(columns=["Full ID"]),
        use_container_width=True,
        hide_index=True,
        on_select="rerun",
        selection_mode="single-row",
    )

    if selected_rows and selected_rows.get("selection", {}).get("rows"):
        row_idx = selected_rows["selection"]["rows"][0]
        selected_id = df.iloc[row_idx]["Full ID"]

        st.subheader("Selected Call Details")
        detail = _api_get(f"/calls/{selected_id}")
        if detail:
            _display_call_result(detail)

            if st.button("Delete This Call", type="secondary") and _api_delete(
                f"/calls/{selected_id}"
            ):
                st.success("Call deleted.")
                st.rerun()


def page_analytics() -> None:
    st.header("Analytics")

    stats = _api_get("/dashboard/stats")
    records = _api_get("/calls/")
    if not stats or not records:
        st.info("No data available yet. Process some calls to see analytics.")
        return

    col1, col2, col3, col4 = st.columns(4)
    with col1:
        st.metric("Total Calls", stats["total_calls"])
    with col2:
        st.metric("Avg Sentiment", f"{stats['avg_sentiment']:.2f}")
    with col3:
        st.metric("Avg Compliance", f"{stats['avg_compliance']:.1f}%")
    with col4:
        st.metric("Escalation Rate", f"{stats['escalation_rate'] * 100:.1f}%")

    st.divider()

    df = pd.DataFrame(records)
    df["date"] = pd.to_datetime(df["created_at"]).dt.date
    df["sentiment_score"] = df["sentiment_score"].astype(float)
    df["compliance_score"] = df["compliance_score"].astype(float)

    col_left, col_right = st.columns(2)

    with col_left:
        st.subheader("Call Volume Over Time")
        volume = df.groupby("date").size().reset_index(name="count")
        fig_volume = px.line(
            volume,
            x="date",
            y="count",
            title="Daily Call Volume",
            color_discrete_sequence=[_GOLD],
        )
        fig_volume.update_layout(
            paper_bgcolor=_NAVY,
            plot_bgcolor="#122030",
            font_color="#FFFFFF",
        )
        st.plotly_chart(fig_volume, use_container_width=True)

    with col_right:
        st.subheader("Compliance Score by Call Type")
        comp_by_type = df.groupby("call_type")["compliance_score"].mean().reset_index()
        fig_comp = px.bar(
            comp_by_type,
            x="call_type",
            y="compliance_score",
            title="Avg Compliance per Call Type",
            color_discrete_sequence=[_GOLD],
        )
        fig_comp.update_layout(paper_bgcolor=_NAVY, plot_bgcolor="#122030", font_color="#FFFFFF")
        st.plotly_chart(fig_comp, use_container_width=True)

    col_left2, col_right2 = st.columns(2)

    with col_left2:
        st.subheader("Sentiment Distribution")
        fig_hist = px.histogram(
            df,
            x="sentiment_score",
            nbins=10,
            range_x=[1, 10],
            title="Sentiment Score Distribution",
            color_discrete_sequence=[_GOLD],
        )
        fig_hist.update_layout(paper_bgcolor=_NAVY, plot_bgcolor="#122030", font_color="#FFFFFF")
        st.plotly_chart(fig_hist, use_container_width=True)

    with col_right2:
        st.subheader("Calls by Type")
        type_counts = df["call_type"].value_counts().reset_index()
        type_counts.columns = ["call_type", "count"]
        fig_pie = px.pie(
            type_counts,
            names="call_type",
            values="count",
            title="Call Type Distribution",
            color_discrete_sequence=px.colors.sequential.Blues_r,
        )
        fig_pie.update_layout(paper_bgcolor=_NAVY, font_color="#FFFFFF")
        st.plotly_chart(fig_pie, use_container_width=True)

    st.subheader("Top Violation Types")
    all_violations: list[str] = []
    for record in records:
        for v in record.get("violations_detail", []):
            desc = v.get("description", "Unknown")
            if desc:
                all_violations.append(desc)

    if all_violations:
        from collections import Counter
        violation_counts = Counter(all_violations).most_common(8)
        vdf = pd.DataFrame(violation_counts, columns=["Violation", "Count"])
        fig_viol = px.bar(
            vdf,
            x="Count",
            y="Violation",
            orientation="h",
            title="Most Frequent Violations",
            color_discrete_sequence=[_GOLD],
        )
        fig_viol.update_layout(paper_bgcolor=_NAVY, plot_bgcolor="#122030", font_color="#FFFFFF")
        st.plotly_chart(fig_viol, use_container_width=True)
    else:
        st.info("No violations recorded in the current call library.")


def page_coaching_hub() -> None:
    st.header("Coaching Hub")
    st.write("Aggregate coaching insights across all processed calls.")

    records = _api_get("/calls/")
    if not records:
        st.info("No calls available. Process calls to see coaching themes.")
        return

    st.subheader("Session Performance Overview")
    df = pd.DataFrame(records)
    df["sentiment_score"] = df["sentiment_score"].astype(float)
    df["compliance_score"] = df["compliance_score"].astype(float)

    escalated = df[df["escalation_detected"]].shape[0]
    perfect_comp = df[df["compliance_score"] >= 90.0].shape[0]

    col1, col2, col3 = st.columns(3)
    with col1:
        st.metric("Calls Reviewed", len(records))
    with col2:
        st.metric("Calls with Escalation", escalated)
    with col3:
        st.metric("Calls with 90+ Compliance", perfect_comp)

    st.divider()
    st.subheader("Agent Performance by Call Type")

    perf = df.groupby("call_type").agg(
        avg_sentiment=("sentiment_score", "mean"),
        avg_compliance=("compliance_score", "mean"),
        call_count=("call_type", "count"),
    ).reset_index()

    fig_perf = go.Figure()
    fig_perf.add_trace(go.Bar(
        name="Avg Sentiment x10",
        x=perf["call_type"],
        y=perf["avg_sentiment"] * 10,
        marker_color=_GOLD,
    ))
    fig_perf.add_trace(go.Bar(
        name="Avg Compliance",
        x=perf["call_type"],
        y=perf["avg_compliance"],
        marker_color="#4FC3F7",
    ))
    fig_perf.update_layout(
        barmode="group",
        title="Performance by Call Type",
        paper_bgcolor=_NAVY,
        plot_bgcolor="#122030",
        font_color="#FFFFFF",
        legend={"bgcolor": "#122030"},
    )
    st.plotly_chart(fig_perf, use_container_width=True)

    st.subheader("Coaching Themes")
    st.write("Top coaching areas identified across all calls:")

    themes = {
        "De-escalation Skills": df["escalation_detected"].sum(),
        "Compliance Adherence": len(df[df["compliance_score"] < 85]),
        "Sentiment Recovery": len(df[df["sentiment_trajectory"] == "declining"]),
        "Low Sentiment Calls": len(df[df["sentiment_score"] < 5.0]),
        "High Performing Calls": len(df[df["sentiment_score"] >= 8.0]),
    }

    for theme, count in sorted(themes.items(), key=lambda x: -x[1]):
        pct = count / len(df) * 100
        col_t, col_p, col_b = st.columns([3, 1, 4])
        with col_t:
            st.write(theme)
        with col_p:
            st.write(f"{count} calls")
        with col_b:
            st.progress(min(pct / 100, 1.0))

    st.subheader("Improvement Recommendations")
    low_comp = df[df["compliance_score"] < 70]
    declining = df[df["sentiment_trajectory"] == "declining"]
    escalated_df = df[df["escalation_detected"]]

    if not low_comp.empty:
        st.warning(
            f"{len(low_comp)} call(s) scored below 70 on compliance. "
            "Priority review of script adherence and mandatory disclosures is recommended."
        )
    if not declining.empty:
        st.warning(
            f"{len(declining)} call(s) showed declining sentiment trajectories. "
            "Focus coaching sessions on early rapport-building and recovery techniques."
        )
    if not escalated_df.empty:
        st.warning(
            f"{len(escalated_df)} call(s) involved escalation. "
            "Review de-escalation training materials and ensure all agents complete the module."
        )
    if low_comp.empty and declining.empty and escalated_df.empty:
        st.success("All calls are performing well. Continue maintaining current standards.")


def main() -> None:
    if not st.session_state["token"]:
        render_login()
        return

    with st.sidebar:
        st.title("Audio Intelligence")
        st.write(f"Signed in as **{st.session_state['username']}**")
        st.divider()

        page = st.radio(
            "Navigation",
            ["Process Call", "Call Library", "Analytics", "Coaching Hub"],
            label_visibility="collapsed",
        )

        st.divider()
        if st.button("Sign Out"):
            st.session_state["token"] = None
            st.session_state["username"] = None
            st.rerun()

    if page == "Process Call":
        page_process_call()
    elif page == "Call Library":
        page_call_library()
    elif page == "Analytics":
        page_analytics()
    elif page == "Coaching Hub":
        page_coaching_hub()


main()