import streamlit as st
import pandas as pd
import sqlite3
import json
import os
from datetime import datetime
import random
from deepgram import Deepgram
import asyncio
import logging
import plotly.express as px
import plotly.graph_objects as go
import re
from pydub import AudioSegment
import requests
from ftplib import FTP
import tempfile
from docx import Document
import io
from streamlit_lottie import st_lottie
import requests
# Setup logging
logging.basicConfig(filename="app.log", level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s")
# Use your Deepgram API key
DEEPGRAM_API_KEY = "5858e763ed38a41fae5aaee972f962bc1004da6d"
dg_client = Deepgram(DEEPGRAM_API_KEY)
PERPLEXITY_API_KEY = "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"
PERPLEXITY_API_URL = "https://api.perplexity.ai/chat/completions"
# File to store our call data
DATABASE_FILE = "calls.db"
# Grading criteria
GRADING_CRITERIA = [
"Debt Problem",
"Ability to Pay",
"Profile Completion",
"Cooperativeness",
"Agent Performance",
"Compliance"
]
# Set page config
st.set_page_config(layout="wide", page_title="Space-Themed Dashboard", page_icon="🚀")
# Custom CSS (same as before)
st.markdown("""
""", unsafe_allow_html=True)
# Database functions
def init_db():
conn = sqlite3.connect(DATABASE_FILE)
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS calls
(id INTEGER PRIMARY KEY AUTOINCREMENT,
agent TEXT,
date TEXT,
duration REAL,
transcript TEXT,
grades TEXT,
overall_grade REAL,
direction TEXT,
call_type TEXT,
campaign TEXT,
phone_number TEXT,
number_of_segments INTEGER,
sentiment TEXT,
script_adherence TEXT,
audio_file BLOB,
audio_filename TEXT,
call_status TEXT,
ai_analysis TEXT,
agent_speaking_time REAL,
speaking_time_percent REAL)''')
c.execute('''CREATE TABLE IF NOT EXISTS campaigns
(id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE)''')
c.execute('''CREATE TABLE IF NOT EXISTS agents
(id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE)''')
c.execute("INSERT OR IGNORE INTO campaigns (name) VALUES ('Default Campaign')")
c.execute("INSERT OR IGNORE INTO agents (name) VALUES ('Default Agent')")
c.execute('''CREATE TABLE IF NOT EXISTS knowledge_base(id INTEGER PRIMARY KEY AUTOINCREMENT,content TEXT)''')
conn.commit()
conn.close()
def load_call_data_no_cache():
conn = sqlite3.connect(DATABASE_FILE)
c = conn.cursor()
c.execute("SELECT * FROM calls")
columns = [column[0] for column in c.description]
calls = []
for row in c.fetchall():
call = dict(zip(columns, row))
call['grades'] = json.loads(call['grades'])
if call['audio_file']:
logging.info(f"Retrieved audio file for call {call['id']}, size: {len(call['audio_file'])} bytes")
else:
logging.warning(f"No audio file found for call {call['id']}")
calls.append(call)
conn.close()
return calls
def get_campaigns():
conn = sqlite3.connect(DATABASE_FILE)
c = conn.cursor()
c.execute("SELECT name FROM campaigns")
campaigns = [row[0] for row in c.fetchall()]
conn.close()
return campaigns if campaigns else ["Default Campaign"]
def get_agents():
conn = sqlite3.connect(DATABASE_FILE)
c = conn.cursor()
c.execute("SELECT name FROM agents")
agents = [row[0] for row in c.fetchall()]
conn.close()
return agents if agents else ["Default Agent"]
def add_campaign(name):
conn = sqlite3.connect(DATABASE_FILE)
c = conn.cursor()
c.execute("INSERT OR IGNORE INTO campaigns (name) VALUES (?)", (name,))
conn.commit()
conn.close()
def add_agent(name):
conn = sqlite3.connect(DATABASE_FILE)
c = conn.cursor()
c.execute("INSERT OR IGNORE INTO agents (name) VALUES (?)", (name,))
conn.commit()
conn.close()
def clear_all_data():
conn = sqlite3.connect(DATABASE_FILE)
c = conn.cursor()
c.execute("DELETE FROM calls")
c.execute("DELETE FROM campaigns")
c.execute("DELETE FROM agents")
conn.commit()
conn.close()
def test_openai_api():
try:
response = openai.ChatCompletion.create(
model="gpt-4-turbo",
messages=[{"role": "user", "content": "Hello, can you hear me?"}],
max_tokens=50,
n=1,
temperature=0.5
)
if response and response.choices and response.choices[0].message:
return True, response.choices[0].message['content'].strip()
else:
return False, "No valid response received"
except Exception as e:
return False, str(e)
async def transcribe_audio_deepgram(audio_file):
mime_type = audio_file.type
audio_bytes = audio_file.read()
logging.info(f"Starting Deepgram transcription for file: {audio_file.name}, MIME type: {mime_type}")
try:
response = await dg_client.transcription.prerecorded(
{'buffer': audio_bytes, 'mimetype': mime_type},
{
'punctuate': True,
'diarize': True,
'language': 'en-US'
}
)
logging.info("Received response from Deepgram")
if response and 'results' in response:
channels = response['results'].get('channels', [])
if channels and 'alternatives' in channels[0]:
alternatives = channels[0]['alternatives']
diarized_segments = alternatives[0].get('words', [])
current_speaker = None
transcript_by_speaker = []
for word_info in diarized_segments:
speaker = word_info.get('speaker', 'Unknown')
word = word_info['punctuated_word']
if speaker != current_speaker:
current_speaker = speaker
transcript_by_speaker.append(f"\nSpeaker {speaker}:")
transcript_by_speaker.append(word)
transcript_text = " ".join(transcript_by_speaker)
transcript_text = re.sub(r'\s([?.!,"])', r'\1', transcript_text)
logging.info(f"Transcription successful. Transcript length: {len(transcript_text)} characters")
return transcript_text.strip()
logging.error("Transcription failed: Unexpected response structure")
logging.debug(f"Deepgram response: {response}")
return ""
except Exception as e:
logging.error(f"Error during Deepgram transcription: {str(e)}", exc_info=True)
return ""
def analyze_call_with_ai(call_id):
try:
conn = sqlite3.connect(DATABASE_FILE)
df = pd.read_sql_query(f"SELECT * FROM calls WHERE id = {call_id}", conn)
conn.close()
if df.empty:
return False, "Call not found."
call = df.iloc[0].to_dict()
transcript = call['transcript']
logging.info(f"Analyzing call ID: {call_id}")
logging.info(f"Transcript length: {len(transcript)} characters")
analysis_prompt = f"""
Analyze the following call transcript and provide a structured analysis in the following format:
Customer Details:
[Extract and list the customer's full name, address, social security number (if mentioned), credit score (if mentioned), type of income, payment status (if behind or paying good on debts), and any other relevant personal information.]
Speaking Time Percentage:
[Estimate the percentage of time the sales agent was speaking versus the customer.]
Questions Asked:
[List all questions asked by the sales agent during the call.]
Number of Questions:
[Provide the total number of questions asked by the sales agent.]
Call Status:
[Based on the transcript, determine the most appropriate status from the following options:
- Closed
- Follow Up
- Can't Afford Payments
- Not Interested
Provide a brief explanation for the chosen status.]
Compliance Questions:
- Asked for next/Same day? (Yes/No):
- Asked for Referrals? (Yes/No):
If Yes, provide name and number of referral. If name or number not given, state the reason or 'None provided'.
- Did agent complete budget analysis? (Yes/No):
- Did the agent take the time to have the client really share their pain? (Yes/No):
- Did the agent ask how these debts add to the client's stress in life? (Yes/No):
- Did the agent ask the client to repeat their understanding of the program? (Yes/No):
- Did the agent mention that the legal group will handle legal summons? (Yes/No):
- Did the agent make the client feel that they are together in this all the way through? (Yes/No):
- Did they understand that they have to provide all third party collection letters? (Yes/No):
- Did the agent mention that our company is going to pay off their debt? (Yes/No):
- Did the Agent clarify that this is not a loan, consolidation, or settlement? (Yes/No):
- Did the agent say that their credit score will not be affected? (Yes/No):
- Did the agent say that they will be able to use credit cards enrolled into the program? (Yes/No):
- Did agent guarantee or promise credit improvement within a certain time frame? (Yes/No):
- Did agent say that doc pro's will contact creditors directly right away? (Yes/No):
- Did the agent make any guarantees or promises such as "I will be able to reduce your debt by 70%"? (Yes/No):
- Did agent ask permission to pull credit? (Yes/No):
- Did the agent sign the doc for the client? (Yes/No):
For each question, answer "Yes" with an explanation if it was asked exactly as stated. If it was similar but not exact, answer "No" with an explanation of what was actually said or how it differed. If neither exact nor similar, simply answer "No" without explanation.
Transcript:
{transcript}
"""
payload = {
"model": "sonar",
"messages": [
{"role": "system", "content": "You are an AI assistant that analyzes call transcripts."},
{"role": "user", "content": analysis_prompt}
]
}
headers = {
"accept": "application/json",
"content-type": "application/json",
"authorization": f"Bearer {PERPLEXITY_API_KEY}"
}
response = requests.post(PERPLEXITY_API_URL, json=payload, headers=headers)
response_json = response.json()
if response.status_code == 200 and 'choices' in response_json:
analysis_result = response_json['choices'][0]['message']['content'].strip()
logging.info(f"Received analysis result. Length: {len(analysis_result)} characters")
conn = sqlite3.connect(DATABASE_FILE)
c = conn.cursor()
c.execute("UPDATE calls SET ai_analysis = ? WHERE id = ?", (analysis_result, call_id))
conn.commit()
conn.close()
return True, analysis_result
else:
logging.error("Invalid or empty response from Perplexity API")
return False, "Invalid or empty response from Perplexity API for analysis"
except Exception as e:
logging.error(f"Error during AI call analysis: {str(e)}", exc_info=True)
return False, f"Error during AI call analysis: {str(e)}"
def process_audio(audio_file, campaign, agent, phone_number, speaking_time_percent):
try:
logging.info(f"Starting audio processing for file: {audio_file.name}")
logging.info(f"Processing audio file: {audio_file.name}, size: {audio_file.size} bytes")
file_size = audio_file.size
logging.info(f"Audio file size: {file_size} bytes")
if file_size == 0:
return False, "Audio file is empty"
audio = AudioSegment.from_file(audio_file)
duration_seconds = len(audio) / 1000
duration_minutes = round(duration_seconds / 60, 2)
audio_file.seek(0)
transcript = asyncio.run(transcribe_audio_deepgram(audio_file))
if transcript:
speaking_time = analyze_speaking_time(transcript)
else:
speaking_time = 0
if not transcript:
logging.error("Transcript is empty, but continuing with processing.")
transcript = "Transcription failed. Please check the audio file."
logging.info("Audio transcription completed")
audio_file.seek(0)
audio_bytes = audio_file.read()
logging.info(f"Audio file size: {len(audio_bytes)} bytes")
new_call = {
"agent": agent,
"date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"duration": duration_minutes,
"transcript": transcript,
"grades": {},
"overall_grade": None,
"direction": "Inbound",
"call_type": "Validation",
"campaign": campaign,
"phone_number": phone_number,
"number_of_segments": 1,
"sentiment": "",
"script_adherence": f"{random.uniform(70, 100):.1f}%",
"audio_file": audio_bytes,
"audio_filename": audio_file.name,
"call_status": "New",
"agent_speaking_time": speaking_time,
"speaking_time_percent": speaking_time_percent,
}
success = save_call_data(new_call)
if success:
logging.info("Call added successfully to the database with the transcript and audio.")
return True, "Audio processed and saved successfully"
else:
logging.error("Failed to save call data to the database.")
return False, "Failed to save call data"
except Exception as e:
logging.error(f"Error processing audio: {str(e)}", exc_info=True)
return False, f"Error processing audio: {str(e)}"
def process_multiple_audio_files(audio_files, campaign, agent, phone_numbers, speaking_time_percents):
results = []
for audio_file, phone_number, speaking_time_percent in zip(audio_files, phone_numbers, speaking_time_percents):
success, message = process_audio(audio_file, campaign, agent, phone_number, float(speaking_time_percent))
results.append((audio_file.name, success, message))
return results
def receive_call_from_vici_api(api_url, call_data):
try:
response = requests.post(api_url, json=call_data)
response.raise_for_status()
audio_url = call_data.get('audio_url')
if audio_url:
audio_response = requests.get(audio_url)
audio_response.raise_for_status()
with tempfile.NamedTemporaryFile(delete=False, suffix='.wav') as temp_audio:
temp_audio.write(audio_response.content)
temp_audio.flush()
with open(temp_audio.name, 'rb') as audio_file:
success, message = process_audio(audio_file, call_data.get('campaign'), call_data.get('agent'))
os.unlink(temp_audio.name)
return success, message
else:
return False, "No audio URL provided in call data"
except requests.RequestException as e:
logging.error(f"Error receiving call from VICI API: {str(e)}")
return False, f"Error receiving call from VICI API: {str(e)}"
# ... [Previous code remains the same] ...
def receive_calls_from_ftp(ftp_host, ftp_user, ftp_pass, ftp_directory):
try:
with FTP(ftp_host) as ftp:
ftp.login(user=ftp_user, passwd=ftp_pass)
ftp.cwd(ftp_directory)
files = ftp.nlst()
results = []
for file in files:
if file.endswith(('.wav', '.mp3', '.m4a')):
with tempfile.NamedTemporaryFile(delete=False, suffix=f'.{file.split(".")[-1]}') as temp_audio:
ftp.retrbinary(f"RETR {file}", temp_audio.write)
temp_audio.flush()
with open(temp_audio.name, 'rb') as audio_file:
success, message = process_audio(audio_file, 'FTP Campaign', 'FTP Agent')
results.append((file, success, message))
os.unlink(temp_audio.name)
# Optional: Delete the file from FTP server after processing
# ftp.delete(file)
return results
except Exception as e:
logging.error(f"Error receiving calls from FTP: {str(e)}")
return [(None, False, f"Error receiving calls from FTP: {str(e)}")]
def validate_call_data(call):
required_fields = ['agent', 'date', 'duration', 'transcript', 'campaign', 'phone_number']
for field in required_fields:
if field not in call or call[field] is None:
logging.error(f"Missing required field: {field}")
return False
return True
def save_call_data_with_retry(call, max_retries=3):
for attempt in range(max_retries):
if save_call_data(call):
return True
logging.warning(f"Failed to save call data. Attempt {attempt + 1} of {max_retries}")
time.sleep(1) # Wait for 1 second before retrying
return False
def save_call_data(call):
try:
with sqlite3.connect(DATABASE_FILE, timeout=20) as conn:
c = conn.cursor()
c.execute('''INSERT INTO calls (agent, date, duration, transcript, grades,
overall_grade, direction, call_type, campaign, phone_number,
number_of_segments, sentiment, script_adherence, audio_file,
audio_filename, call_status, agent_speaking_time, speaking_time_percent)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)''',
(call['agent'], call['date'], call['duration'],
call['transcript'], json.dumps(call['grades']), call['overall_grade'],
call['direction'], call['call_type'], call['campaign'],
call['phone_number'], call['number_of_segments'],
call['sentiment'], call['script_adherence'],
call['audio_file'], call['audio_filename'], call['call_status'],
call['agent_speaking_time'], call['speaking_time_percent']))
conn.commit()
logging.info("Call data saved successfully")
return True
except sqlite3.Error as e:
logging.error(f"SQLite error: {str(e)}", exc_info=True)
return False
except Exception as e:
logging.error(f"Unexpected error saving call data: {str(e)}", exc_info=True)
return False
except sqlite3.Error as e:
conn.rollback()
if "database is locked" in str(e):
logging.error("Database is locked. Try again later.")
else:
logging.error(f"SQLite error: {str(e)}", exc_info=True)
return False
except Exception as e:
conn.rollback()
logging.error(f"Unexpected error saving call data: {str(e)}", exc_info=True)
return False
def update_call_analysis(call_id, analysis):
conn = sqlite3.connect(DATABASE_FILE)
c = conn.cursor()
c.execute("UPDATE calls SET ai_analysis = ? WHERE id = ?", (analysis, call_id))
conn.commit()
conn.close()
def update_call_status(call_id, new_status):
conn = sqlite3.connect(DATABASE_FILE)
c = conn.cursor()
c.execute("UPDATE calls SET call_status = ? WHERE id = ?", (new_status, call_id))
conn.commit()
conn.close()
def ask_ai_about_call(call_id, question, use_knowledge_base=False):
conn = sqlite3.connect(DATABASE_FILE)
df = pd.read_sql_query(f"SELECT * FROM calls WHERE id = {call_id}", conn)
conn.close()
if df.empty:
return "Call not found."
call = df.iloc[0].to_dict()
transcript = call['transcript']
modified_transcript = transcript.replace("Speaker 0:", "Agent:").replace("Speaker 1:", "Client:")
knowledge_base = get_knowledge_base() if use_knowledge_base else ""
prompt = f"""
Based on the following call transcript and knowledge base, please answer the question:
Transcript:
{modified_transcript}
Knowledge Base:
{knowledge_base}
Question: {question}
"""
try:
payload = {
"model": "sonar",
"messages": [
{"role": "user", "content": prompt}
]
}
headers = {
"accept": "application/json",
"content-type": "application/json",
"authorization": f"Bearer {PERPLEXITY_API_KEY}"
}
response = requests.post(PERPLEXITY_API_URL, json=payload, headers=headers)
response_json = response.json()
if response.status_code == 200 and 'choices' in response_json:
return response_json['choices'][0]['message']['content'].strip()
else:
logging.error("Invalid or empty response from Perplexity API.")
return "Error: Invalid or empty response from Perplexity API."
except Exception as e:
logging.error(f"Error in AI question answering: {str(e)}")
return f"Error: {str(e)}"
def generate_word_document(transcript, ai_analysis):
doc = Document()
doc.add_heading('Call Transcript and Analysis', 0)
doc.add_heading('Transcript', level=1)
# Modify the transcript to replace Speaker labels
modified_transcript = transcript.replace("Speaker 0:", "Agent:").replace("Speaker 1:", "Client:")
doc.add_paragraph(modified_transcript)
doc.add_heading('AI Analysis', level=1)
doc.add_paragraph(ai_analysis)
buffer = io.BytesIO()
doc.save(buffer)
buffer.seek(0)
return buffer
def analyze_speaking_time(transcript):
agent_words = 0
client_words = 0
current_speaker = None
for line in transcript.split('\n'):
if line.startswith("Speaker 0:"):
current_speaker = "agent"
words = line.replace("Speaker 0:", "").split()
elif line.startswith("Speaker 1:"):
current_speaker = "client"
words = line.replace("Speaker 1:", "").split()
else:
words = line.split()
if current_speaker == "agent":
agent_words += len(words)
elif current_speaker == "client":
client_words += len(words)
total_words = agent_words + client_words
if total_words > 0:
agent_percentage = (agent_words / total_words) * 100
else:
agent_percentage = 0
return agent_percentage
def update_speaking_time(call_id, speaking_time):
conn = sqlite3.connect(DATABASE_FILE)
c = conn.cursor()
c.execute("UPDATE calls SET agent_speaking_time = ? WHERE id = ?", (speaking_time, call_id))
conn.commit()
conn.close()
def render_animation():
animation_response = requests.get('https://assets1.lottiefiles.com/packages/lf20_vykpwt8b.json')
animation_json = dict()
if animation_response.status_code == 200:
animation_json = animation_response.json()
else:
print("Error in the URL")
return st_lottie(animation_json, height=200, width=300)
def add_to_knowledge_base(content):
conn = sqlite3.connect(DATABASE_FILE)
c = conn.cursor()
c.execute("INSERT INTO knowledge_base (content) VALUES (?)", (content,))
conn.commit()
conn.close()
def get_knowledge_base():
conn = sqlite3.connect(DATABASE_FILE)
c = conn.cursor()
c.execute("SELECT content FROM knowledge_base")
knowledge_base = c.fetchone()
conn.close()
return knowledge_base[0] if knowledge_base else ""
def clear_knowledge_base():
conn = sqlite3.connect(DATABASE_FILE)
c = conn.cursor()
c.execute("DELETE FROM knowledge_base")
conn.commit()
conn.close()
if 'theme' not in st.session_state:
st.session_state.theme = 'light'
def toggle_theme():
st.session_state.theme = 'dark' if st.session_state.theme == 'light' else 'light'
st.button("Toggle Theme", on_click=toggle_theme)
if st.session_state.theme == 'dark':
st.markdown("""
""", unsafe_allow_html=True)
def main():
st.title("Call Analysis Dashboard")
tab1, tab2, tab3, tab4, tab5, tab6, tab7 = st.tabs(["📊 Overview", "📝 Call List", "➕ Add Call", "🔄 Bulk Upload", "📞 VICI Integration", "🎭 Roleplaying", "⚙️ Settings"])
with tab1:
calls = load_call_data_no_cache()
if calls:
df = pd.DataFrame(calls)
total_calls = len(df)
avg_duration = df['duration'].mean()
positive_sentiment = (df['sentiment'] == 'Positive').mean() * 100
col1, col2, col3 = st.columns(3)
with col1:
st.metric("Total Calls", total_calls)
with col2:
st.metric("Average Duration", f"{avg_duration:.1f} min")
with col3:
st.metric("Positive Sentiment", f"{positive_sentiment:.1f}%")
st.subheader("Agent Performance Ranking")
agent_performance = df.groupby('agent')['overall_grade'].mean().sort_values(ascending=False)
for rank, (agent, score) in enumerate(agent_performance.items(), 1):
st.text(f"{rank}. {agent}: {score:.2f}")
st.subheader("Agent Speaking Time")
if 'agent_speaking_time' in df.columns:
agent_speaking_time = df.groupby('agent')['agent_speaking_time'].mean().sort_values(ascending=False)
for agent, time in agent_speaking_time.items():
st.text(f"{agent}: {time:.2f}%")
else:
st.info("No agent speaking time data available.")
else:
st.info("No call data available.")
total_calls = 0
avg_duration = 0
positive_sentiment = 0
col1, col2, col3 = st.columns(3)
with col1:
st.metric("Total Calls", total_calls)
with col2:
st.metric("Average Duration", f"{avg_duration:.1f} min")
with col3:
st.metric("Positive Sentiment", f"{positive_sentiment:.1f}%")
st.subheader("Agent Performance Ranking")
st.info("No data available for agent performance ranking.")
st.subheader("Agent Speaking Time")
st.info("No agent speaking time data available.")
with tab2:
calls = load_call_data_no_cache()
col1, col2, col3 = st.columns(3)
with col1:
agent_filter = st.selectbox("Filter by Agent", ["All"] + get_agents())
with col2:
date_filter = st.date_input("Filter by Date", value=None)
with col3:
search_query = st.text_input("Search (Agent, Client, Phone)")
if not calls:
st.info("No calls available yet.")
else:
filtered_calls = calls
if agent_filter != "All":
filtered_calls = [call for call in filtered_calls if call['agent'] == agent_filter]
if date_filter:
filtered_calls = [call for call in filtered_calls if datetime.strptime(call['date'], "%Y-%m-%d %H:%M:%S").date() == date_filter]
if search_query:
filtered_calls = [call for call in filtered_calls if search_query.lower() in call['agent'].lower() or search_query in call['phone_number']]
for i, call in enumerate(filtered_calls):
col1, col2, col3, col4, col5 = st.columns([2, 2, 1, 1, 1])
with col1:
st.write(f"**Date:** {call['date']}")
with col2:
st.write(f"**Agent:** {call['agent']}")
with col3:
st.write(f"**Duration:** {call['duration']:.2f} min")
with col4:
st.write(f"**Phone:** {call['phone_number']}")
with col5:
st.write(f"**Speaking Time %:** {call.get('speaking_time_percent', 'N/A')}%")
with st.expander("View Call Details"):
col1, col2 = st.columns([6, 1])
with col1:
call_tabs = st.tabs(["Call Info", "AI Analysis", "Ask AI"])
with col2:
export_button = st.button("📤", key=f"export_{call['id']}", help="Export as Word document")
if export_button:
with st.spinner("Generating document..."):
render_animation() # Custom spinner animation
transcript = call.get("transcript", "No transcript available.")
ai_analysis = call.get("ai_analysis", "No AI analysis available.")
doc_buffer = generate_word_document(transcript, ai_analysis)
st.success("Document generated successfully!")
st.download_button(
label="Download Word Document",
data=doc_buffer,
file_name=f"call_{call['id']}_export.docx",
mime="application/vnd.openxmlformats-officedocument.wordprocessingml.document")
with call_tabs[0]:
if 'audio_file' in call and call['audio_file']:
try:
st.audio(call['audio_file'], format="audio/wav")
except Exception as e:
st.error(f"Error playing audio: {str(e)}")
else:
st.warning("No audio file available for this call.")
st.markdown("### Transcript")
transcript_text = call.get("transcript", "No transcript available.")
# Modify the transcript to replace Speaker labels and add color
modified_transcript = transcript_text.replace("Speaker 0:", "Agent:").replace("Speaker 1:", "Client:")
# Split the transcript into lines
lines = modified_transcript.split('\n')
# Process each line and add color
colored_lines = []
for line in lines:
if line.startswith("Agent:"):
colored_lines.append(f'{line}')
elif line.startswith("Client:"):
colored_lines.append(f'{line}')
else:
colored_lines.append(line)
# Join the lines back together
colored_transcript = '
'.join(colored_lines)
# Display the modified and colored transcript
st.markdown(colored_transcript, unsafe_allow_html=True)
current_status = call.get('call_status', 'New')
new_status = st.selectbox(
"Call Status",
["Closed", "Follow Up", "Can't Afford Payments", "Not Interested", "New"],
index=["Closed", "Follow Up", "Can't Afford Payments", "Not Interested", "New"].index(current_status),
key=f"status_{call['id']}"
)
if new_status != current_status:
update_call_status(call['id'], new_status)
with call_tabs[1]:
if call.get('ai_analysis'):
st.markdown("### AI Analysis Results")
st.markdown(call['ai_analysis'])
else:
if st.button("Analyze Call", key=f"analyze_{call['id']}"):
with st.spinner("Analyzing call..."):
success, result = analyze_call_with_ai(call['id'])
if success:
st.success("Call analysis completed successfully.")
st.markdown("### AI Analysis Results")
st.markdown(result)
update_call_analysis(call['id'], result)
else:
st.error(result)
with call_tabs[2]:
st.markdown("### Ask AI about this call")
use_kb = st.checkbox("Use Knowledge Base", key=f"use_kb_{call['id']}")
if f"conversation_{call['id']}" not in st.session_state:
st.session_state[f"conversation_{call['id']}"] = []
for q, a in st.session_state[f"conversation_{call['id']}"]:
st.markdown(f"You:{q}",unsafe_allow_html=True)
st.markdown(f"AI:{a}",unsafe_allow_html=True)
user_question = st.text_input("Enter your question:", key=f"question_{call['id']}")
if st.button("Ask", key=f"ask_{call['id']}"):
if user_question:
with st.spinner("AI is thinking..."):
ai_response = ask_ai_about_call(call['id'], user_question)
st.session_state[f"conversation_{call['id']}"].append((user_question, ai_response))
st.markdown(f"You:{user_question}",unsafe_allow_html=True)
st.markdown(f"AI:{ai_response}",unsafe_allow_html=True)
st.experimental_rerun()
else:
st.warning("Please enter a question.")
with tab3:
st.header("Add New Call")
add_method = st.radio("Choose method", ["Upload Recording", "Manually Add Details"])
if add_method == "Upload Recording":
uploaded_file = st.file_uploader("Upload Recording", type=['wav','mp3','m4a'])
agent = st.selectbox("Select Agent", get_agents())
campaign = st.selectbox("Select Campaign", get_campaigns())
phone_number = st.text_input("Enter Phone Number", value="")
speaking_time_percent = st.number_input("Speaking Time Percent", min_value=0.0, max_value=100.0, value=0.0, step=0.1)
col1, col2 = st.columns(2)
with col1:
if st.button("Process Audio"):
if uploaded_file is not None and phone_number:
with st.spinner("Processing audio..."):
success, message = process_audio(uploaded_file, campaign, agent, phone_number, speaking_time_percent)
if success:
st.success(message)
else:
st.error(message)
else:
st.warning("Please upload an audio file and enter a phone number.")
# ... (rest of the code remains the same)
with col2:
if st.button("Analyze Speaking Time"):
if uploaded_file is not None:
with st.spinner("Analyzing speaking time..."):
audio_file = uploaded_file
audio_file.seek(0)
transcript = asyncio.run(transcribe_audio_deepgram(audio_file))
if transcript:
speaking_time = analyze_speaking_time(transcript)
st.success(f"Agent speaking time: {speaking_time:.2f}%")
else:
st.error("Failed to transcribe audio for analysis.")
else:
st.warning("Please upload an audio file first.")
else: # Manually Add Details
agent = st.selectbox("Select Agent", get_agents())
campaign = st.selectbox("Select Campaign", get_campaigns())
phone_number = st.text_input("Enter Phone Number", value="")
call_duration = st.number_input("Call Duration (minutes)", min_value=0.0, step=0.1)
call_transcript = st.text_area("Call Transcript", height=200)
call_sentiment = st.selectbox("Call Sentiment", ["Positive", "Neutral", "Negative"])
if st.button("Add Call"):
if phone_number and call_transcript:
new_call = {
"agent": agent,
"date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"duration": call_duration,
"transcript": call_transcript,
"grades": {},
"overall_grade": None,
"direction": "Inbound",
"call_type": "Validation",
"campaign": campaign,
"phone_number": phone_number,
"number_of_segments": 1,
"sentiment": call_sentiment,
"script_adherence": f"{random.uniform(70, 100):.1f}%",
"audio_file": None,
"audio_filename": None,
"call_status": "New"
}
success = save_call_data(new_call)
if success:
st.success("Call added successfully")
else:
st.error("Failed to add call")
else:
st.warning("Please enter a phone number and call transcript.")
with tab4:
st.header("Bulk Upload")
add_method = st.radio("Choose method", ["Upload Recordings", "Manually Add Details"])
if add_method == "Upload Recordings":
agent = st.selectbox("Select Agent for Bulk Upload", get_agents())
campaign = st.selectbox("Select Campaign for Bulk Upload", get_campaigns())
phone_numbers = st.text_area("Enter Phone Numbers (one per line)", height=100)
speaking_time_percents = st.text_area("Enter Speaking Time Percents (one per line)", height=100)
uploaded_files = st.file_uploader("Upload Recordings", type=['wav','mp3','m4a'], accept_multiple_files=True)
col1, col2 = st.columns(2)
with col1:
if st.button("Process All Files"):
if uploaded_files and phone_numbers and speaking_time_percents:
with st.spinner("Processing audio files..."):
phone_number_list = phone_numbers.split('\n')
speaking_time_percent_list = speaking_time_percents.split('\n')
if len(phone_number_list) != len(uploaded_files) or len(speaking_time_percent_list) != len(uploaded_files):
st.error("The number of phone numbers and speaking time percents must match the number of uploaded files.")
else:
results = process_multiple_audio_files(uploaded_files, campaign, agent, phone_number_list, speaking_time_percent_list)
for file_name, success, message in results:
if success:
st.success(f"{file_name}: {message}")
else:
st.error(f"{file_name}: {message}")
else:
st.warning("Please upload audio files, enter phone numbers, and speaking time percents.")
# ... (rest of the code remains the same)
with col2:
if st.button("Analyze Speaking Time (All Files)"):
if uploaded_files:
with st.spinner("Analyzing speaking time for all files..."):
results = []
for audio_file in uploaded_files:
audio_file.seek(0)
transcript = asyncio.run(transcribe_audio_deepgram(audio_file))
if transcript:
speaking_time = analyze_speaking_time(transcript)
results.append((audio_file.name, speaking_time))
else:
results.append((audio_file.name, None))
for file_name, speaking_time in results:
if speaking_time is not None:
st.success(f"{file_name}: Agent speaking time: {speaking_time:.2f}%")
else:
st.error(f"{file_name}: Failed to analyze speaking time.")
else:
st.warning("Please upload audio files first.")
else: # Manually Add Details
st.warning("Bulk manual entry is not supported. Please use the 'Add Call' tab for manual entries.")
with tab5:
api_url = st.text_input("VICI API URL")
if st.button("Receive Call from VICI API"):
if api_url:
with st.spinner("Receiving call from VICI API..."):
success, message = receive_call_from_vici_api(api_url, {}) # You may need to modify this to include necessary call data
if success:
st.success(message)
else:
st.error(message)
else:
st.warning("Please enter the VICI API URL.")
st.subheader("FTP Integration")
ftp_host = st.text_input("FTP Host")
ftp_user = st.text_input("FTP Username")
ftp_pass = st.text_input("FTP Password", type="password")
ftp_directory = st.text_input("FTP Directory")
if st.button("Receive Calls from FTP"):
if ftp_host and ftp_user and ftp_pass and ftp_directory:
with st.spinner("Receiving calls from FTP..."):
results = receive_calls_from_ftp(ftp_host, ftp_user, ftp_pass, ftp_directory)
for file_name, success, message in results:
if success:
st.success(f"{file_name}: {message}")
else:
st.error(f"{file_name}: {message}")
else:
st.warning("Please fill in all FTP details.")
with tab6:
st.write("Use the embedded roleplaying demo below to practice your sales techniques.")
# Embed the VAPI Assistant demo link as an iframe
vapi_demo_url = "https://vapi.ai?demo=true&shareKey=d3d9a3bf-4be8-4675-9e1f-24cf0c716de8&assistantId=7c3372f2-32a1-47ba-8c17-4351d970fb9f"
st.components.v1.iframe(vapi_demo_url, height=700)
with tab7:
col1, col2 = st.columns(2)
with col1:
st.subheader("Add New Campaign")
new_campaign = st.text_input("New Campaign Name")
if st.button("Add Campaign"):
add_campaign(new_campaign)
st.success(f"Campaign '{new_campaign}' added successfully!")
with col2:
st.subheader("Add New Agent")
new_agent = st.text_input("New Agent Name")
if st.button("Add Agent"):
add_agent(new_agent)
st.success(f"Agent '{new_agent}' added successfully!")
st.subheader("Manage Existing Data")
if st.button("Clear All Data"):
if st.checkbox("I understand this will delete all call data permanently"):
clear_all_data()
st.success("All data has been cleared.")
st.subheader("Test OpenAI API")
if st.button("Test OpenAI Connection"):
success, result = test_openai_api()
if success:
st.success(f"OpenAI API is working. Response: {result}")
else:
st.error(f"OpenAI API test failed. Error: {result}")
st.subheader("Knowledge Base Management")
uploaded_file = st.file_uploader("Upload Knowledge Base (TXT file)", type="txt")
if uploaded_file is not None:
try:
# Try UTF-8 first
content = uploaded_file.read().decode("utf-8")
except UnicodeDecodeError:
try:
# If UTF-8 fails, try ISO-8859-1
uploaded_file.seek(0)
content = uploaded_file.read().decode("iso-8859-1")
except UnicodeDecodeError:
# If ISO-8859-1 also fails, try with errors='ignore'
uploaded_file.seek(0)
content = uploaded_file.read().decode("utf-8", errors="ignore")
if st.button("Import Knowledge Base"):
clear_knowledge_base()
add_to_knowledge_base(content)
st.success("Knowledge base imported successfully!")
st.subheader("Current Knowledge Base")
current_kb = get_knowledge_base()
st.text_area("Current Knowledge Base Content", value=current_kb, height=200, disabled=True)
if st.button("Clear Knowledge Base"):
clear_knowledge_base()
st.success("Knowledge base cleared successfully!")
if __name__ == "__main__":
init_db()
main()