Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| from datetime import datetime | |
| import pandas as pd | |
| import json | |
| from pathlib import Path | |
| from huggingface_hub import hf_hub_download, HfApi | |
| import streamlit as st | |
| from features import FEATURES | |
| from utils import check_password, process_olivia_data | |
| def format_seconds(seconds: int) -> str: | |
| if seconds == 1: | |
| return "1 second" | |
| elif seconds < 60: | |
| return f"{seconds} seconds" | |
| else: | |
| minutes = seconds // 60 | |
| remaining_seconds = seconds % 60 | |
| if minutes == 1: | |
| minute_str = "1 minute" | |
| else: | |
| minute_str = f"{minutes} minutes" | |
| if remaining_seconds == 1: | |
| second_str = "1 second" | |
| else: | |
| second_str = f"{remaining_seconds} seconds" | |
| return f"{minute_str} {second_str}" | |
| REPO_URL = "https://huggingface.co/datasets/trevolution/conversation-analytics-comments" | |
| api = HfApi() | |
| features_to_show = ['C: Missed Expectation : No Call Back/Follow Up', | |
| 'C: Missed Expectation - Not Informed', | |
| 'C: Missed Promises', | |
| 'C: Repeat Contact - General/Other', | |
| 'C: Repeat Contact - Previous Calls', | |
| 'C: Repeat Information', | |
| 'C: Agent Hanged Up', | |
| 'C: Disputing Charge / Chargeback', | |
| 'A: Transfer', | |
| 'A: Transfer offer', | |
| 'C: Channel Switch - Website', | |
| 'C: Objection - Competitor - Switch', | |
| 'C: Channel Switch - Webchat', | |
| 'Escalation: External - Attorney General', | |
| 'Escalation: External - BBB', | |
| 'Escalation: External - Legal', | |
| 'Escalation: Internal - Complaint', | |
| 'Escalation: Internal - Corporate', | |
| 'Escalation: Internal - Do Not Contact/Remove from list', | |
| 'Escalation: Internal - Supervisor', | |
| 'Voucher', | |
| 'Refund Voucher', | |
| 'Refund' | |
| ] | |
| style = ( | |
| 'border: 1px solid #ccc; ' | |
| 'padding: 10px; ' | |
| 'border-radius: 5px; ' | |
| 'max-height: 500px; ' # Set your desired maximum height | |
| 'overflow: auto;' # Enable vertical scrollbar if content exceeds max height | |
| ) | |
| def get_div(input): | |
| return f'<div style="{style}"><p>{input}</p></div>' | |
| def main(): | |
| if not check_password(): | |
| st.stop() | |
| comments_path = hf_hub_download( | |
| repo_id='trevolution/conversation-analytics-comments', | |
| repo_type='dataset', | |
| filename='comments_report_7.json', | |
| token=st.secrets['WRITE_TOKEN'], | |
| ) | |
| with open(comments_path, 'r') as f: | |
| comments = json.load(f) | |
| with open('transcriptions_report_7.json', 'r') as f: | |
| transcriptions = json.load(f) | |
| with open('analytics_report_7.json', 'r') as f: | |
| analytics = json.load(f) | |
| call_ids = [json.loads(_['metadata'])['call_id'] for _ in transcriptions] | |
| call_ids = list(sorted(list(set(call_ids)))) | |
| st.title('Olivia - Agent - Conversation Analytics') | |
| call_id = st.selectbox( | |
| 'Call IDs:', | |
| call_ids, | |
| format_func=lambda call_id: f'{call_ids.index(call_id) + 1}: {call_id}' | |
| ) | |
| if not st.session_state.get('selectbox'): | |
| st.session_state['selectbox'] = call_id | |
| else: | |
| if call_id != st.session_state['selectbox']: | |
| st.session_state['analyze_button'] = False | |
| st.session_state['selectbox'] = call_id | |
| transcription = [json.loads(_['transcription']) for _ in transcriptions if json.loads(_['metadata'])['call_id'] == call_id][0] | |
| try: | |
| analytics = [json.loads(_['analytics']) for _ in analytics if call_id == json.loads(_['metadata'])['call_id']][0] | |
| analytics = analytics['analytics'] | |
| analytics = [f for f in analytics if f['name'] in features_to_show] | |
| except: | |
| analytics = None | |
| st.audio(f'data/{call_id}.ogg', format='audio/ogg') | |
| analyze_button = st.button("Get Conversation Analytics") | |
| if not st.session_state.get('analyze_button'): | |
| st.session_state['analyze_button'] = analyze_button | |
| if st.session_state['selectbox'] and st.session_state['analyze_button']: | |
| conversation = process_olivia_data(transcription) | |
| st.text('Conversation (Olivia Speech-to-Text):') | |
| st.markdown(get_div(conversation['text']), unsafe_allow_html=True) | |
| with st.spinner('Loading analytics...'): | |
| st.text('Analytics') | |
| readable_analytics = '' | |
| for i, feature in enumerate(analytics): | |
| if feature['timestamp']: | |
| start_time, end_time = int(feature['timestamp'][0]), int(feature['timestamp'][1]) | |
| start_time, end_time = format_seconds(start_time), format_seconds(end_time) | |
| readable_analytics += f"{i+1}. {feature['name']}: {feature['response']}. Quotation: {feature['quotation']}. Timestamp: {start_time}-{end_time}\n\n\n" | |
| else: | |
| readable_analytics += f"{i+1}. {feature['name']}: {feature['response']}. Quotation: {feature['quotation']}\n\n\n" | |
| st.markdown(get_div(readable_analytics), unsafe_allow_html=True) | |
| if "saved_comments" not in st.session_state: | |
| st.session_state['saved_comments'] = "" | |
| user_comments = st.text_area(f"Comments on {call_id}", key='user_comments', height=350) | |
| def submit(): | |
| st.session_state['saved_comments'] = st.session_state['user_comments'] | |
| st.session_state['user_comments'] = "" | |
| button = st.button("Save comments", on_click=submit) | |
| if button and st.session_state['saved_comments']: | |
| if call_id in comments: | |
| comments[call_id].append( | |
| { | |
| 'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S"), | |
| 'text': st.session_state['saved_comments'] | |
| } | |
| ) | |
| else: | |
| comments[call_id] = [ | |
| { | |
| 'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S"), | |
| 'text': st.session_state['saved_comments'] | |
| } | |
| ] | |
| api.upload_file( | |
| path_or_fileobj=json.dumps(comments).encode('utf-8'), | |
| path_in_repo="comments_report_7.json", | |
| repo_id="trevolution/conversation-analytics-comments", | |
| repo_type="dataset", | |
| token=st.secrets['WRITE_TOKEN'], | |
| commit_message=f"{call_id}_{datetime.now().strftime('%Y-%m-%d')}" | |
| ) | |
| st.success("Saved") | |
| if comments.get(call_id): | |
| value = '' | |
| for comment in comments.get(call_id): | |
| value += f"{comment['timestamp']}: {comment['text']}\n" | |
| st.text_area(label='Comments:', value=value, disabled=True, height=350) | |
| else: | |
| st.text_area(label='Comments:', value="No comments exist at the moment", disabled=True) | |
| if __name__ == "__main__": | |
| main() | |