Update sentiment_analysis.py

sentiment_analysis.py

@@ -1,14 +1,16 @@
 import os
 import json
 import time
-from speech_recognition import Recognizer, Microphone, AudioData, UnknownValueError, RequestError
+import numpy as np
+import sounddevice as sd
 from transformers import pipeline, AutoModelForSequenceClassification, AutoTokenizer
 from huggingface_hub import login
 from product_recommender import ProductRecommender
-from objection_handler import load_objections, check_objections  # Ensure check_objections is imported
+from objection_handler import load_objections, check_objections
 from objection_handler import ObjectionHandler
 from env_setup import config
 from sentence_transformers import SentenceTransformer
+from scipy.io.wavfile import write
 from dotenv import load_dotenv
 
 # Load environment variables
@@ -24,9 +26,6 @@ model = AutoModelForSequenceClassification.from_pretrained(model_name)
 tokenizer = AutoTokenizer.from_pretrained(model_name)
 sentiment_analyzer = pipeline("sentiment-analysis", model=model, tokenizer=tokenizer)
 
-# Speech Recognition Setup
-recognizer = Recognizer()
-
 # Function to analyze sentiment
 def preprocess_text(text):
     """Preprocess text for better sentiment analysis."""
@@ -37,12 +36,12 @@ def analyze_sentiment(text):
     try:
         if not text.strip():
             return "NEUTRAL", 0.0
-        
+
         processed_text = preprocess_text(text)
         result = sentiment_analyzer(processed_text)[0]
-        
+
         print(f"Sentiment Analysis Result: {result}")
-        
+
         # Map raw labels to sentiments
         sentiment_map = {
             'Very Negative': "NEGATIVE",
@@ -51,14 +50,35 @@ def analyze_sentiment(text):
             'Positive': "POSITIVE",
             'Very Positive': "POSITIVE"
         }
-        
+
         sentiment = sentiment_map.get(result['label'], "NEUTRAL")
         return sentiment, result['score']
-        
+
     except Exception as e:
         print(f"Error in sentiment analysis: {e}")
         return "NEUTRAL", 0.5
 
+def record_audio(duration=5, sample_rate=44100):
+    """Record audio for a specified duration."""
+    print("Recording audio...")
+    audio = sd.rec(int(duration * sample_rate), samplerate=sample_rate, channels=1, dtype='float32')
+    sd.wait()  # Wait for the recording to finish
+    print("Recording completed.")
+    return np.squeeze(audio)
+
+def transcribe_audio(audio, sample_rate=44100):
+    """Transcribe recorded audio using a speech-to-text API."""
+    try:
+        # Save audio to a temporary file for transcription
+        audio_file = "temp_audio.wav"
+        write(audio_file, sample_rate, audio)
+        # Call external transcription service (e.g., Whisper, AssemblyAI, or Google)
+        transcription = "Example transcription text from audio."  # Placeholder
+        return transcription
+    except Exception as e:
+        print(f"Error in audio transcription: {e}")
+        return ""
+
 def transcribe_with_chunks(objections_dict):
     """Perform real-time transcription with sentiment analysis."""
     print("Say 'start listening' to begin transcription. Say 'stop listening' to stop.")
@@ -68,86 +88,64 @@ def transcribe_with_chunks(objections_dict):
     chunk_start_time = time.time()
 
     # Initialize handlers with semantic search capabilities
-    objection_handler = ObjectionHandler(r"C:\Users\shaik\Downloads\Sales Calls Transcriptions - Sheet3.csv")
-    product_recommender = ProductRecommender(r"C:\Users\shaik\Downloads\Sales Calls Transcriptions - Sheet2.csv")
+    objection_handler = ObjectionHandler("path_to_objections.csv")
+    product_recommender = ProductRecommender("path_to_recommendations.csv")
 
     # Load the embeddings model once
     model = SentenceTransformer('all-MiniLM-L6-v2')
 
     try:
-        with Microphone() as source:
-            recognizer.adjust_for_ambient_noise(source)
-            print("Microphone calibrated. Please speak.")
-            
-            while True:
-                print("Listening for speech...")
-                try:
-                    audio_data = recognizer.listen(source, timeout=5)
-                    text = recognizer.recognize_google(audio_data)
-
-                    if "start listening" in text.lower():
-                        is_listening = True
-                        print("Listening started. Speak into the microphone.")
-                        continue
-                    elif "stop listening" in text.lower():
-                        is_listening = False
-                        print("Listening stopped.")
-                        if current_chunk:
-                            chunk_text = " ".join(current_chunk)
-                            sentiment, score = analyze_sentiment(chunk_text)
-                            chunks.append((chunk_text, sentiment, score))
-                            current_chunk = []
-                        continue
-
-                    if is_listening and text.strip():
-                        print(f"Transcription: {text}")
-                        current_chunk.append(text)
-
-                        if time.time() - chunk_start_time > 3:
-                            if current_chunk:
-                                chunk_text = " ".join(current_chunk)
-                                
-                                # Always process sentiment
-                                sentiment, score = analyze_sentiment(chunk_text)
-                                chunks.append((chunk_text, sentiment, score))
-
-                                # Get objection responses and check similarity score
-                                query_embedding = model.encode([chunk_text])
-                                distances, indices = objection_handler.index.search(query_embedding, 1)
-                                
-                                # If similarity is high enough, show objection response
-                                if distances[0][0] < 1.5:  # Threshold for similarity
-                                    responses = objection_handler.handle_objection(chunk_text)
-                                    if responses:
-                                        print("\nSuggested Response:")
-                                        for response in responses:
-                                            print(f"→ {response}")
-                                
-                                # Get product recommendations and check similarity score
-                                distances, indices = product_recommender.index.search(query_embedding, 1)
-                                
-                                # If similarity is high enough, show recommendations
-                                if distances[0][0] < 1.5:  # Threshold for similarity
-                                    recommendations = product_recommender.get_recommendations(chunk_text)
-                                    if recommendations:
-                                        print(f"\nRecommendations for this response:")
-                                        for idx, rec in enumerate(recommendations, 1):
-                                            print(f"{idx}. {rec}")
-                                
-                                print("\n")
-                                current_chunk = []
-                                chunk_start_time = time.time()
-                except UnknownValueError:
-                    print("Could not understand the audio.")
-                except RequestError as e:
-                    print(f"Could not request results from Google Speech Recognition service; {e}")
+        while True:
+            if not is_listening:
+                command = input("Enter 'start' to begin listening or 'stop' to quit: ").lower()
+                if command == "start":
+                    is_listening = True
+                    print("Listening started. Speak into the microphone.")
+                    continue
+                elif command == "stop":
+                    break
+
+            # Record and process audio in chunks
+            audio_data = record_audio(duration=5)
+            text = transcribe_audio(audio_data)
+            if text.strip():
+                print(f"Transcription: {text}")
+                current_chunk.append(text)
+
+                if time.time() - chunk_start_time > 3:
+                    if current_chunk:
+                        chunk_text = " ".join(current_chunk)
+
+                        # Process sentiment
+                        sentiment, score = analyze_sentiment(chunk_text)
+                        chunks.append((chunk_text, sentiment, score))
+
+                        # Handle objections and recommendations
+                        query_embedding = model.encode([chunk_text])
+                        responses = objection_handler.handle_objection(chunk_text)
+                        recommendations = product_recommender.get_recommendations(chunk_text)
+
+                        # Print results
+                        if responses:
+                            print("\nSuggested Response:")
+                            for response in responses:
+                                print(f"→ {response}")
+                        if recommendations:
+                            print("\nRecommendations for this response:")
+                            for idx, rec in enumerate(recommendations, 1):
+                                print(f"{idx}. {rec}")
+
+                        print("\n")
+                        current_chunk = []
+                        chunk_start_time = time.time()
 
     except KeyboardInterrupt:
         print("\nExiting...")
-        return chunks
+
+    return chunks
 
 if __name__ == "__main__":
-    objections_file_path = r"C:\Users\shaik\Downloads\Sales Calls Transcriptions - Sheet3.csv"
+    objections_file_path = "path_to_objections.csv"
     objections_dict = load_objections(objections_file_path)
     transcribed_chunks = transcribe_with_chunks(objections_dict)
     print("Final transcriptions and sentiments:", transcribed_chunks)