import os
import time
import pandas as pd
import streamlit as st
from qa_content import filter_context_messages, analyze_customer_messages
from qa_info import analyze_data
# 设置页面配置
st.set_page_config(
page_title="AI对话分析系统",
page_icon="💬",
layout="wide",
initial_sidebar_state="expanded"
)
# 自定义CSS样式
st.markdown("""
""", unsafe_allow_html=True)
# 初始化session_state
if 'view_mode' not in st.session_state:
st.session_state.view_mode = "chat_analysis" # 默认视图模式：'chat_analysis' 或 'stats_analysis'
if 'chat_data' not in st.session_state:
st.session_state.chat_data = []
if 'selected_user' not in st.session_state:
st.session_state.selected_user = None
if 'user_list' not in st.session_state:
st.session_state.user_list = []
if 'stats_data' not in st.session_state:
st.session_state.stats_data = {}
if 'file_processed' not in st.session_state:
st.session_state.file_processed = False
# 侧边栏操作区域
with st.sidebar:
st.markdown("", unsafe_allow_html=True)
# 视图选择
st.markdown("", unsafe_allow_html=True)
view_options = ["历史对话内容", "历史对话数量"]
selected_view = st.radio("",
options=view_options,
index=0 if st.session_state.view_mode == "chat_analysis" else 1,
label_visibility="collapsed")
# 根据选择更新视图模式
if selected_view == "历史对话内容" and st.session_state.view_mode != "chat_analysis":
st.session_state.view_mode = "chat_analysis"
st.rerun()
elif selected_view == "历史对话数量" and st.session_state.view_mode != "stats_analysis":
st.session_state.view_mode = "stats_analysis"
st.rerun()
# 分割线
st.markdown("
", unsafe_allow_html=True)
# 上传文件部分 - 同时处理两种分析
st.markdown("", unsafe_allow_html=True)
# st.markdown("", unsafe_allow_html=True)
# 文件上传
uploaded_file = st.file_uploader("上传Excel文件", type=["xlsx", "xls"])
# 修改临时文件路径，确保在Docker环境中有权限写入
TEMP_FILE_DIR = "./temp_files"
os.makedirs(TEMP_FILE_DIR, exist_ok=True)
# 如果有上传文件并且未处理
if uploaded_file is not None and not st.session_state.file_processed:
# 显示处理状态
st.markdown("", unsafe_allow_html=True)
progress = st.progress(0)
# # 保存上传的文件
# with open("temp_analysis_file.xlsx", "wb") as f:
# f.write(uploaded_file.getvalue())
# 保存上传的文件
temp_file_path = os.path.join(TEMP_FILE_DIR, "temp_analysis_file.xlsx")
with open(temp_file_path, "wb") as f:
f.write(uploaded_file.getvalue())
try:
# 更新进度
progress.progress(20)
time.sleep(0.2)
# 分析对话数据
# results = analyze_customer_messages("temp_analysis_file.xlsx")
results = analyze_customer_messages(temp_file_path)
# 更新进度
progress.progress(40)
time.sleep(0.2)
# 筛选不满意对话结果
filtered_results = filter_context_messages(results)
# 更新进度
progress.progress(60)
time.sleep(0.2)
# 分析统计数据
# stats_result = analyze_data("temp_analysis_file.xlsx")
stats_result = analyze_data(temp_file_path)
# 更新进度
progress.progress(90)
time.sleep(0.2)
# 将分析结果保存到session_state中
if filtered_results:
st.session_state.chat_data = filtered_results
st.session_state.user_list = [entry["用户"] for entry in filtered_results]
st.session_state.selected_user = st.session_state.user_list[0]
st.session_state.stats_data = stats_result
# 处理完成
progress.progress(100)
time.sleep(0.3)
progress.empty()
st.session_state.file_processed = True
# 显示成功消息
if filtered_results:
chat_msg = f"✓ 已分析 {len(filtered_results)} 位用户的对话"
else:
chat_msg = "未找到符合条件的对话记录"
st.success(f"数据分析完成！\n{chat_msg}")
st.rerun()
except Exception as e:
progress.empty()
st.error(f"处理文件出错: {str(e)}")
st.session_state.file_processed = True
# 如果已有对话数据，显示用户选择列表
if st.session_state.view_mode == "chat_analysis" and st.session_state.user_list:
st.markdown("
", unsafe_allow_html=True)
st.markdown("", unsafe_allow_html=True)
selected_user = st.selectbox(
"选择要查看的用户",
options=st.session_state.user_list,
index=st.session_state.user_list.index(st.session_state.selected_user) if st.session_state.selected_user in st.session_state.user_list else 0
)
if selected_user != st.session_state.selected_user:
st.session_state.selected_user = selected_user
st.rerun()
# 重置按钮（始终显示在侧边栏底部）
st.markdown("
", unsafe_allow_html=True)
if st.button("重置分析数据", type="primary", use_container_width=True):
for key in ['chat_data', 'selected_user', 'user_list', 'file_processed', 'stats_data']:
if key in st.session_state:
del st.session_state[key]
# # 尝试删除临时文件
# try:
# if os.path.exists("temp_analysis_file.xlsx"):
# os.remove("temp_analysis_file.xlsx")
# except:
# pass
# 尝试删除临时文件
try:
temp_file_path = os.path.join(TEMP_FILE_DIR, "temp_analysis_file.xlsx")
if os.path.exists(temp_file_path):
os.remove(temp_file_path)
except:
pass
st.rerun()
# 版权信息
# st.markdown("© 2024 AI对话分析系统 v1.0
", unsafe_allow_html=True)
# 主界面 - 仅显示结果
if st.session_state.view_mode == "chat_analysis":
# 历史对话内容结果显示
st.markdown("历史对话内容
", unsafe_allow_html=True)
# 检查是否有数据可显示
if not st.session_state.chat_data or not st.session_state.selected_user:
# 显示空状态
st.markdown("", unsafe_allow_html=True)
st.markdown("
💬
", unsafe_allow_html=True)
st.markdown("
欢迎使用历史对话内容
", unsafe_allow_html=True)
st.markdown("
请在左侧边栏上传Excel文件，系统会自动分析不满意对话并在此处显示结果。
", unsafe_allow_html=True)
st.markdown("
正在查看用户 {st.session_state.selected_user} 的对话记录
", unsafe_allow_html=True)
# 找到当前选择的用户对话
selected_chat = None
for entry in st.session_state.chat_data:
if entry["用户"] == st.session_state.selected_user:
selected_chat = entry["消息列表"]
break
if selected_chat:
# 显示所有消息
for message in selected_chat:
try:
user = message.get('客户用户名', '未知')
servicer = message.get('客服用户名', '未知')
content = message.get("消息内容", "")
direction = message.get("对话方向", "")
time_str = message.get("信息时间", "")
call_type = message.get("自动回复类型", "")
call_intent = message.get('自动回复意图', '')
call_type_tag = f'{call_type}'
call_intent_tag = f'{call_intent}'
# 如果消息为空，则跳过
if not content:
continue
# 根据对话方向判断是客户还是客服
if direction == "呼入": # 客户消息
st.markdown(f"""
{user[:1]}
{time_str}{call_type_tag}
{content}
{call_intent_tag}
""", unsafe_allow_html=True)
else: # 客服消息(呼出)
# 判断是否是机器人回复
is_bot = str(message.get("机器人自动回复状态", "")) == "是"
is_ok = message.get('分类是否正确', "")
# 标签
bot_tag = '机器人' if is_bot else ''
# 根据分类是否正确显示不同颜色的标签
ok_tag = ""
if is_bot and is_ok == "满意":
ok_tag = '满意'
elif is_bot and is_ok == "不满意":
ok_tag = '不满意'
elif is_bot and is_ok == "未评价":
ok_tag = '未评价'
st.markdown(f"""
{time_str} {bot_tag} {ok_tag}
{content}
{servicer[:]}
""", unsafe_allow_html=True)
except Exception as e:
st.error(f"显示消息出错: {str(e)}")
else:
st.markdown('', unsafe_allow_html=True)
st.markdown("", unsafe_allow_html=True)
else: # stats_analysis 模式
# 历史对话数量结果显示
st.markdown("历史对话数量分析
", unsafe_allow_html=True)
# 检查是否有数据可显示
if not st.session_state.stats_data:
# 显示空状态
st.markdown("", unsafe_allow_html=True)
st.markdown("
📊
", unsafe_allow_html=True)
st.markdown("
欢迎使用历史对话数量
", unsafe_allow_html=True)
st.markdown("
请在左侧边栏上传Excel文件，系统会自动分析对话数据并在此处显示统计结果。
", unsafe_allow_html=True)
st.markdown("
", unsafe_allow_html=True)
st.markdown(f"
{st.session_state.stats_data.get('总呼出', 0)}
", unsafe_allow_html=True)
st.markdown(f"
总呼出数量
", unsafe_allow_html=True)
st.markdown("
", unsafe_allow_html=True)
st.markdown(f"
{st.session_state.stats_data.get('AI呼出', 0)}
", unsafe_allow_html=True)
st.markdown(f"
AI对话数量
", unsafe_allow_html=True)
st.markdown("
", unsafe_allow_html=True)
st.markdown(f"
{st.session_state.stats_data.get('满意率', '0%')}
", unsafe_allow_html=True)
st.markdown(f"
满意率
", unsafe_allow_html=True)
st.markdown("
", unsafe_allow_html=True)
st.markdown(f"
{st.session_state.stats_data.get('未评价率', '0%')}
", unsafe_allow_html=True)
st.markdown(f"
未评价率
", unsafe_allow_html=True)
st.markdown("
', unsafe_allow_html=True)
st.markdown('', unsafe_allow_html=True)
# 使用表格呈现
data = [
["总呼出数量", st.session_state.stats_data.get('总呼出', 0)],
["AI呼出数量", st.session_state.stats_data.get('AI呼出', 0)],
["人工呼出数量", st.session_state.stats_data.get('人呼出', 0)]
]
# 添加AI占比
total = st.session_state.stats_data.get('总呼出', 0)
ai = st.session_state.stats_data.get('AI呼出', 0)
ai_percentage = f"{(ai/total*100):.1f}%" if total > 0 else "0%"
data.append(["AI占比", ai_percentage])
# 创建DataFrame并显示
df1 = pd.DataFrame(data, columns=["指标", "数值"])
st.dataframe(df1, hide_index=True, use_container_width=True)
st.markdown('
', unsafe_allow_html=True)
with col2:
st.markdown('', unsafe_allow_html=True)
st.markdown('', unsafe_allow_html=True)
# 使用表格呈现
data = [
["满意数量", st.session_state.stats_data.get('满意数', 0)],
["不满意数量", st.session_state.stats_data.get('不满意', 0)],
["未评价数量", st.session_state.stats_data.get('未评价', 0)],
["满意率", st.session_state.stats_data.get('满意率', '0%')]
]
# 创建DataFrame并显示
df2 = pd.DataFrame(data, columns=["指标", "数值"])
st.dataframe(df2, hide_index=True, use_container_width=True)
st.markdown('
', unsafe_allow_html=True)
# 未评价详细统计
st.markdown('', unsafe_allow_html=True)
st.markdown('', unsafe_allow_html=True)
unrated_stats = st.session_state.stats_data.get('未评价统计', {})
if unrated_stats:
# 将字典转换为DataFrame
data = [[user, count] for user, count in unrated_stats.items()]
df3 = pd.DataFrame(data, columns=["客服", "未评价数量"])
# 按未评价数量降序排序
df3 = df3.sort_values(by="未评价数量", ascending=False)
# 显示表格
st.dataframe(df3, hide_index=True, use_container_width=True)
else:
st.info("没有未评价数据")
st.markdown('
', unsafe_allow_html=True)