Update pipeline.py

pipeline.py

@@ -1109,143 +1109,133 @@ class UltraRobustCallAnalytics:
             torch.cuda.empty_cache()
 
     def _map_emotion_to_sentiment(self, emotion):
-    """Map emotion labels to sentiment with polarity score"""
-    emotion_lower = emotion.lower()
-    
-    positive_emotions = {
-        'happy': 0.8, 'joy': 0.9, 'excited': 0.85, 
-        'pleased': 0.7, 'satisfied': 0.75, 'content': 0.6
-    }
-    
-    negative_emotions = {
-        'sad': -0.6, 'angry': -0.9, 'frustrated': -0.8,
-        'annoyed': -0.7, 'disappointed': -0.65, 'upset': -0.75
-    }
-    
-    if emotion_lower in positive_emotions:
-        return {
-            "sentiment": "positive",
-            "polarity_score": positive_emotions[emotion_lower],
-            "confidence": "high"
+        """Map emotion labels to sentiment with polarity score"""
+        emotion_lower = emotion.lower()
+        
+        positive_emotions = {
+            'happy': 0.8, 'joy': 0.9, 'excited': 0.85, 
+            'pleased': 0.7, 'satisfied': 0.75, 'content': 0.6
         }
-    
-    if emotion_lower in negative_emotions:
+        negative_emotions = {
+            'sad': -0.6, 'angry': -0.9, 'frustrated': -0.8,
+            'annoyed': -0.7, 'disappointed': -0.65, 'upset': -0.75
+        }
+        if emotion_lower in positive_emotions:
+            return {
+                "sentiment": "positive",
+                "polarity_score": positive_emotions[emotion_lower],
+                "confidence": "high"
+            }
+            
+        if emotion_lower in negative_emotions:
+            return {
+                "sentiment": "negative",
+                "polarity_score": negative_emotions[emotion_lower],
+                "confidence": "high"
+            }
         return {
-            "sentiment": "negative",
-            "polarity_score": negative_emotions[emotion_lower],
-            "confidence": "high"
+            "sentiment": "neutral",
+            "polarity_score": 0.0,
+            "confidence": "medium"
         }
-    
-    return {
-        "sentiment": "neutral",
-        "polarity_score": 0.0,
-        "confidence": "medium"
-    }
 
     def _calculate_speech_rate(self, text, duration_seconds):
-    """Calculate words per minute (WPM) and classify pace"""
-    if duration_seconds < 0.1:
-        return {"wpm": 0, "word_count": 0, "speech_pace": "unknown"}
-    
-    words = text.split()
-    word_count = len(words)
-    wpm = (word_count / (duration_seconds / 60.0)) if duration_seconds > 0 else 0
-    
-    if wpm < 100: pace = "slow"
-    elif wpm < 140: pace = "normal"
-    elif wpm < 180: pace = "fast"
-    else: pace = "very_fast"
-    
-    return {
-        "wpm": round(wpm, 1),
-        "word_count": word_count,
-        "speech_pace": pace
-    }
+        """Calculate words per minute (WPM) and classify pace"""
+        if duration_seconds < 0.1:
+            return {"wpm": 0, "word_count": 0, "speech_pace": "unknown"}
+        words = text.split()
+        word_count = len(words)
+        wpm = (word_count / (duration_seconds / 60.0)) if duration_seconds > 0 else 0
+        if wpm < 100: pace = "slow"
+        elif wpm < 140: pace = "normal"
+        elif wpm < 180: pace = "fast"
+        else: pace = "very_fast"
+        
+        return {
+            "wpm": round(wpm, 1),
+            "word_count": word_count,
+            "speech_pace": pace
+        }
 
     def _extract_keywords(self, text, top_n=5):
-    """Extract keywords/keyphrases using KeyBERT"""
-    if self.keyword_model is None or len(text.split()) < 3:
-        return []
-    
-    try:
-        keywords = self.keyword_model.extract_keywords(
-            text,
-            keyphrase_ngram_range=(1, 2),
-            stop_words='english',
-            top_n=top_n,
-            use_maxsum=True,
-            nr_candidates=20
-        )
+        """Extract keywords/keyphrases using KeyBERT"""
+        if self.keyword_model is None or len(text.split()) < 3:
+            return []
         
-        return [
-            {"keyword": kw[0], "relevance": round(float(kw[1]), 3)}
-            for kw in keywords
-        ]
-    except:
-        return []
+        try:
+            keywords = self.keyword_model.extract_keywords(
+                text,
+                keyphrase_ngram_range=(1, 2),
+                stop_words='english',
+                top_n=top_n,
+                use_maxsum=True,
+                nr_candidates=20
+            )
+            return [
+                {"keyword": kw[0], "relevance": round(float(kw[1]), 3)}
+                for kw in keywords
+            ]
+        except:
+            return []
 
 
     def _classify_topic(self, text):
-    """Classify text into call center topics"""
-    if self.topic_classifier is None or len(text.split()) < 5:
-        return {"topic": "unknown", "confidence": 0.0}
-    
-    try:
-        result = self.topic_classifier(text, self.topic_labels, multi_label=False)
-        
-        return {
-            "topic": result['labels'][0],
-            "confidence": round(float(result['scores'][0]), 3),
-            "top_3_topics": [
-                {"topic": label, "score": round(float(score), 3)}
-                for label, score in zip(result['labels'][:3], result['scores'][:3])
-            ]
-        }
-    except:
-        return {"topic": "unknown", "confidence": 0.0}
+        """Classify text into call center topics"""
+        if self.topic_classifier is None or len(text.split()) < 5:
+            return {"topic": "unknown", "confidence": 0.0}
+        try:
+            result = self.topic_classifier(text, self.topic_labels, multi_label=False)
+            return {
+                "topic": result['labels'][0],
+                "confidence": round(float(result['scores'][0]), 3),
+                "top_3_topics": [
+                    {"topic": label, "score": round(float(score), 3)}
+                    for label, score in zip(result['labels'][:3], result['scores'][:3])
+                ]
+            }
+        except:
+            return {"topic": "unknown", "confidence": 0.0}
 
 
     def _aggregate_call_insights(self, results):
-    """Aggregate keywords and topics at call level"""
-    if not results:
-        return {"top_keywords": [], "primary_topic": {"topic": "unknown"}}
+        """Aggregate keywords and topics at call level"""
+        if not results:
+            return {"top_keywords": [], "primary_topic": {"topic": "unknown"}}
+        all_keywords = {}
+        for seg in results:
+            if 'keywords' in seg:
+                for kw in seg['keywords']:
+                    keyword = kw['keyword']
+                    score = kw['relevance']
+                    all_keywords[keyword] = max(all_keywords.get(keyword, 0), score)
     
-    # Aggregate keywords
-    all_keywords = {}
-    for seg in results:
-        if 'keywords' in seg:
-            for kw in seg['keywords']:
-                keyword = kw['keyword']
-                score = kw['relevance']
-                all_keywords[keyword] = max(all_keywords.get(keyword, 0), score)
-    
-    top_keywords = [
-        {"keyword": k, "relevance": round(v, 3)}
-        for k, v in sorted(all_keywords.items(), key=lambda x: x[1], reverse=True)[:10]
-    ]
+        top_keywords = [
+            {"keyword": k, "relevance": round(v, 3)}
+            for k, v in sorted(all_keywords.items(), key=lambda x: x[1], reverse=True)[:10]
+        ]
     
     # Aggregate topics
-    topic_votes = defaultdict(float)
-    for seg in results:
-        if 'topic' in seg and seg['topic']['confidence'] > 0.5:
-            topic_votes[seg['topic']['topic']] += seg['topic']['confidence']
+        topic_votes = defaultdict(float)
+        for seg in results:
+            if 'topic' in seg and seg['topic']['confidence'] > 0.5:
+                topic_votes[seg['topic']['topic']] += seg['topic']['confidence']
     
-    primary_topic = {
-        "topic": max(topic_votes, key=topic_votes.get) if topic_votes else "unknown",
-        "confidence": round(topic_votes[max(topic_votes, key=topic_votes.get)] / len(results), 3) if topic_votes else 0.0
-    }
+        primary_topic = {
+            "topic": max(topic_votes, key=topic_votes.get) if topic_votes else "unknown",
+            "confidence": round(topic_votes[max(topic_votes, key=topic_votes.get)] / len(results), 3) if topic_votes else 0.0
+        }
     
     # Calculate stats
-    total_words = sum(seg.get('speech_rate', {}).get('word_count', 0) for seg in results)
-    wpm_values = [seg.get('speech_rate', {}).get('wpm', 0) for seg in results if seg.get('speech_rate', {}).get('wpm', 0) > 0]
-    average_wpm = round(np.mean(wpm_values), 1) if wpm_values else 0
+        total_words = sum(seg.get('speech_rate', {}).get('word_count', 0) for seg in results)
+        wpm_values = [seg.get('speech_rate', {}).get('wpm', 0) for seg in results if seg.get('speech_rate', {}).get('wpm', 0) > 0]
+        average_wpm = round(np.mean(wpm_values), 1) if wpm_values else 0
     
-    return {
-        "top_keywords": top_keywords,
-        "primary_topic": primary_topic,
-        "total_words": total_words,
-        "average_wpm": average_wpm
-    }
+        return {
+            "top_keywords": top_keywords,
+            "primary_topic": primary_topic,
+            "total_words": total_words,
+            "average_wpm": average_wpm
+        }
 
 
 if __name__ == "__main__":