# ---------------------------------------------------------------------------- # Copyright (c) 2025, Gnaneshwar. All rights reserved. # Purpose: Post Call Analysis # Author: Gnaneshwar # Description: This script performs post-call analysis, analyzing various metrics # and insights from the call data. This can be used to evaluate agent # performance and improve call center efficiency. # ---------------------------------------------------------------------------- # # openai_helper.py # import os # import json # import logging # import requests # from db_config import get_db_connection # from dotenv import load_dotenv # load_dotenv() # OPENAI_API_KEY = os.getenv('OPENAI_API_KEY') # def send_openai_analysis(call_id): # conn = get_db_connection() # cursor = conn.cursor() # cursor.execute("SELECT * FROM 6840_sarvamresponse WHERE callid = %s AND status = 1", (call_id,)) # calls = cursor.fetchall() # for call in calls: # transcript = call[2] # Assuming 'transcript' is at index 2 # prompt = f""" # Analyze the following customer call transcript and return the following in JSON format: # 1. Top 5 Keywords # 2. Overall Sentiment (Positive, Negative, Neutral) # 3. Detected Emotions (e.g., Happy, Angry, Frustrated) # 4. Summary of the Call # 5. Sales Intent (High, Medium, Low) # Transcript: # \"{transcript}\" # """ # headers = { # "Authorization": f"Bearer {OPENAI_API_KEY}", # "Content-Type": "application/json" # } # body = { # "model": "gpt-4", # "messages": [ # {"role": "system", "content": "You are a helpful assistant that analyzes call center transcripts."}, # {"role": "user", "content": prompt} # ], # "temperature": 0.7 # } # try: # response = requests.post("https://api.openai.com/v1/chat/completions", headers=headers, json=body, verify=False) # data = response.json() # content = data.get("choices", [{}])[0].get("message", {}).get("content", "") # parsed = json.loads(content) # # Extracted fields # keywords = json.dumps(parsed.get("Top 5 Keywords", [])) # sentiment = parsed.get("Overall Sentiment", "") # emotions = json.dumps(parsed.get("Detected Emotions", [])) # summary = parsed.get("Summary of the Call", "") # cursor.execute(""" # UPDATE 6840_sarvamresponse SET # status = 2, # keywords = %s, # sentiment = %s, # emotions = %s, # summary = %s, # openai_raw = %s # WHERE id = %s # """, (keywords, sentiment, emotions, summary, content, call[0])) # call[0] = id # conn.commit() # logging.info(f"✅ OpenAI processed callid {call_id}") # except Exception as e: # logging.error(f"❌ OpenAI error for callid {call_id}: {str(e)}") # cursor.close() # conn.close() # orginal code import os import json import logging import boto3 from botocore.exceptions import BotoCoreError, ClientError from db_config import get_db_connection from dotenv import load_dotenv from message_helper import send_message logging.basicConfig(level=logging.INFO) load_dotenv() # AWS Credentials AWS_ACCESS_KEY_ID = os.getenv('AWS_ACCESS_KEY_ID', 'AKIAUWPMKLMSH5E4QMNY') AWS_SECRET_ACCESS_KEY = os.getenv('AWS_SECRET_ACCESS_KEY', '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') AWS_REGION = os.getenv('AWS_REGION', 'eu-north-1') bedrock = boto3.client( service_name='bedrock-runtime', region_name=AWS_REGION, aws_access_key_id=AWS_ACCESS_KEY_ID, aws_secret_access_key=AWS_SECRET_ACCESS_KEY ) MODEL_ID = "amazon.nova-lite-v1:0" def send_openai_analysis(call_id, bid=None): conn = get_db_connection() cursor = conn.cursor() # Use dynamic table name if bid is provided, otherwise use default if bid: sarvam_response_table = f"{bid}_sarvamresponse" calls_table = f"{bid}_calls" else: sarvam_response_table = "7987_sarvamresponse" # Default fallback calls_table = "7987_calls" cursor.execute(f"SELECT * FROM {sarvam_response_table} WHERE callid = %s AND status = 1", (call_id,)) calls = cursor.fetchall() for call in calls: transcript = call[2] # transcript is at index 1, not 2 # Debug logging logging.info(f"🔍 Retrieved transcript for callid {call_id}: {transcript[:100] if transcript else 'None'}...") if not transcript or transcript.strip() == "": logging.warning(f"⚠️ Empty or null transcript for callid {call_id}") continue prompt = f""" Analyze the following customer call transcript and return the following in JSON format: 1. Top 5 Keywords 2. Overall Sentiment with all (Positive, Negative, Neutral) with percentage 3. Detected Emotions (e.g., Happy, Angry, Frustrated) 4. Summary of the Call 5. Sales Intent (High, Medium, Low) 6. Customer Details (name, phone, email if available) 7. Call Purpose / Issue Type 8. Call Outcome (Resolved, Partially Resolved, Escalated, Follow-up Needed, Unresolved) 9. Action Items for Agent and Customer (clear and actionable) 10. Customer Pain Points (specific problems expressed) 11. Quality of the Call: - Communication clarity (agent & customer) - Background noise or call quality issues - Long pauses or hold time - Interruptions / speaking over each other - Agent professionalism, politeness, and empathy - Overall pacing and structure - Provide a Call Quality Score (1–100) - Include a brief explanation for the score Transcript: \"{transcript}\" """ logging.info(f"Sending prompt for callid {call_id}: {prompt}") body = { "inferenceConfig": { "max_new_tokens": 1000, "temperature": 0.7, "top_p": 0.9 }, "messages": [ { "role": "user", "content":[{"text": prompt}] } ] } try: logging.info(f"Request body: {json.dumps(body)}") response = bedrock.invoke_model( modelId=MODEL_ID, contentType="application/json", accept="application/json", body=json.dumps(body) ) raw_response = response['body'].read().decode('utf-8') logging.info(f"Raw response body for callid {call_id}: {raw_response}") content = '' if raw_response: response_body = json.loads(raw_response) content_list = response_body.get('output', {}).get('message', {}).get('content', []) if content_list and isinstance(content_list, list): content = content_list[0].get('text', '').strip() if not content: logging.error(f"Empty content received for callid {call_id}") raise ValueError(f"Empty content received from Bedrock response for callid {call_id}") if content.startswith("```json"): content = content.replace("```json", "").replace("```", "").strip() elif content.startswith("```"): content = content.replace("```", "").strip() content = content.strip() logging.info(f"Cleaned response content for callid {call_id}: {content}") try: parsed = json.loads(content) except json.JSONDecodeError as e: logging.error(f"JSON decode error for callid {call_id}: {str(e)}") logging.error(f"Invalid content: {content}") raise keywords = json.dumps(parsed.get("Top 5 Keywords", [])) sentiment = json.dumps(parsed.get("Overall Sentiment", [])) emotions = json.dumps(parsed.get("Detected Emotions", [])) summary = parsed.get("Summary of the Call", "") salesintent = parsed.get("Sales Intent", "") customer_details = json.dumps(parsed.get("Customer Details", {})) call_purpose = parsed.get("Call Purpose / Issue Type", "") call_outcome = parsed.get("Call Outcome", "") # Support both possible keys for action items from the model action_items_data = ( parsed.get("Action Items for Agent and Customer") or parsed.get("Action Items") or {} ) action_items = json.dumps(action_items_data) pain_points = json.dumps(parsed.get("Customer Pain Points", [])) call_quality = json.dumps(parsed.get("Quality of the Call", {})) call_quality_score = parsed.get("Quality of the Call", {}).get("Call Quality Score", "") logging.info(f"Parsed data for callid {call_id}: Keywords: {keywords}, Sentiment: {sentiment}, Emotions: {emotions}, Summary: {summary}, Sales: {salesintent}, Customer: {customer_details},Call Purpose: {call_purpose},Call Outcome: {call_outcome},Action Items: {action_items},Pain Points: {pain_points},Call Quality: {call_quality},Call Quality Score: {call_quality_score}") cursor.execute(f""" UPDATE {calls_table} SET status = 2, keywords = %s, sentiments = %s, emotions = %s, summary = %s, sand_logic_response = %s, sales_intent = %s, transcripts = %s, customer_details = %s, call_purpose = %s, call_outcome = %s, action_items = %s, pain_points = %s, call_quality = %s, call_quality_score = %s WHERE callid = %s """, (keywords, sentiment, emotions, summary, content, salesintent, transcript, customer_details, call_purpose, call_outcome, action_items, pain_points,call_quality, call_quality_score,call_id)) conn.commit() send_message(call_id, bid) logging.info(f"✅ Bedrock processed callid {call_id}") except (BotoCoreError, ClientError, json.JSONDecodeError, ValueError) as e: logging.error(f"❌ Bedrock error for callid {call_id}: {str(e)}") logging.error(f"⚠️ Failed content: {content if 'content' in locals() else 'No content available'}") cursor.close() conn.close()