"""
Chat components for rendering messages and handling user interactions.
"""
import streamlit as st
import pandas as pd
import base64
from typing import List, Dict, Any
from config import OUTLINE_INDIGO_USER, DARK_MODE_SLATE_AI, RETRY_BUTTON_TEXT, DOWNLOAD_BUTTON_TEXT
class ChatRenderer:
"""Handles rendering of chat messages and interactions."""
def __init__(self):
self.user_avatar = OUTLINE_INDIGO_USER
self.ai_avatar = DARK_MODE_SLATE_AI
def render_chat_history(self, messages: List[Dict[str, Any]]) -> None:
"""Render the complete chat history."""
st.markdown("
", unsafe_allow_html=True)
for i, msg in enumerate(messages):
if msg["role"] == "user":
self._render_user_message(msg, i, messages)
elif msg["role"] == "assistant":
# Skip error messages that are already shown after user messages
if not self._is_error_shown_after_user(i, messages):
self._render_assistant_message(msg)
st.markdown("
", unsafe_allow_html=True)
def _render_user_message(self, msg: Dict[str, Any], index: int, messages: List[Dict[str, Any]]) -> None:
"""Render a user message with avatar."""
st.markdown(
f'''''',
unsafe_allow_html=True
)
# Check if next message is an error and render retry option
self._render_error_retry_if_needed(index, messages)
def _render_error_retry_if_needed(self, index: int, messages: List[Dict[str, Any]]) -> None:
"""Render error message and retry button if the next message is an error."""
if (
index + 1 < len(messages)
and messages[index + 1]["role"] == "assistant"
and messages[index + 1].get("is_error")
):
error_msg = messages[index + 1]["content"]
cols = st.columns([0.3, 0.7])
with cols[1]:
col1, col2 = st.columns([0.8, 0.2])
with col1:
st.markdown(
f'{error_msg}
',
unsafe_allow_html=True
)
with col2:
if st.button(RETRY_BUTTON_TEXT, key=f"retry_{index}"):
self._handle_retry(index)
def _handle_retry(self, index: int) -> None:
"""Handle retry button click."""
messages = st.session_state["messages"]
st.session_state["messages"] = messages[:index + 1]
st.session_state["messages"].append({
"role": "assistant",
"content": "🤔 Thinking...",
"is_placeholder": True
})
st.rerun()
def _is_error_shown_after_user(self, index: int, messages: List[Dict[str, Any]]) -> bool:
"""Check if this error message is already shown after the previous user message."""
if not messages[index].get("is_error"):
return False
# Check if this is an error that follows a user message
if index > 0 and messages[index - 1]["role"] == "user":
return True
return False
def _render_assistant_message(self, msg: Dict[str, Any]) -> None:
"""Render an assistant message with avatar and optional data/charts."""
bubble_class = "error-bubble" if msg.get("is_error") else "ai-bubble"
# Render message bubble
if msg.get("is_placeholder"):
self._render_thinking_message(bubble_class)
else:
self._render_regular_message(msg["content"], bubble_class)
# Render data table if present
if msg.get("data"):
self._render_data_table(msg["data"], msg)
# Render chart if present
if msg.get("chart"):
self._render_chart(msg["chart"])
def _render_thinking_message(self, bubble_class: str) -> None:
"""Render a thinking/loading message with spinner."""
st.markdown(
f'''''',
unsafe_allow_html=True
)
def _render_regular_message(self, content: str, bubble_class: str) -> None:
"""Render a regular assistant message."""
st.markdown(
f'''''',
unsafe_allow_html=True
)
def _render_data_table(self, data: List[Dict], msg: Dict[str, Any]) -> None:
"""Render data table with download option."""
df = pd.DataFrame(data)
st.dataframe(df, use_container_width=True)
csv = df.to_csv(index=False).encode("utf-8")
st.download_button(
DOWNLOAD_BUTTON_TEXT,
csv,
"results.csv",
"text/csv",
key=f"download_csv_{id(msg)}"
)
def _render_chart(self, chart_data: str) -> None:
"""Render chart from base64 data."""
img_data = base64.b64decode(chart_data)
st.image(img_data, use_column_width=True)