src/pattern_detector.py

"""
Pattern Detection Module
Detects risk signals and warning patterns in conversations.
"""

from typing import List, Dict, Any
import re


class PatternDetector:
    """Detects risk signals in message patterns."""
    
    def __init__(self):
        """Initialize pattern detector."""
        self.comparison_keywords = [
            'competitor', 'alternative', 'better', 'cheaper', 'faster',
            'other', 'someone else', 'another', 'different', 'switch',
            'change', 'similar', 'compare', 'versus', 'instead'
        ]
        
        self.frustration_keywords = [
            'slow', 'late', 'delayed', 'wait', 'frustrated', 'annoyed',
            'angry', 'upset', 'disappointed', 'problem', 'issue', 'bug',
            'broken', 'not working', 'fail', 'error', 'impossible'
        ]
        
        self.disengagement_keywords = [
            'cancel', 'stop', 'end', 'quit', 'leave', 'exit', 'goodbye',
            'farewell', 'thanks anyway', 'no thanks', 'decline', 'refuse',
            'not interested', 'moving on', 'consider', 'think about',
            'evaluate', 'looking at'
        ]
        
        self.price_keywords = [
            'expensive', 'cost', 'price', 'cheap', 'expensive', 'fee',
            'charge', 'budget', 'discount', 'negotiate', 'lower'
        ]
    
    def detect_signals(self, messages: List[Dict[str, Any]], 
                      context: str = "general") -> Dict[str, Any]:
        """
        Detect risk signals in conversations.
        
        Args:
            messages: List of messages
            context: Type of relationship (customer/employee/investor/general)
        
        Returns:
            Dictionary with detected signals and risk assessment
        """
        if not messages:
            return self._empty_signals()
        
        signals = []
        risk_scores = []
        breaking_point = None
        key_phrases = []
        
        for i, msg in enumerate(messages):
            # Handle both strings and dicts
            if isinstance(msg, dict):
                text = msg.get('text', '').lower()
                timestamp = msg.get('timestamp', f'Message {i+1}')
            elif isinstance(msg, str):
                text = msg.lower()
                timestamp = f'Message {i+1}'
            else:
                text = str(msg).lower()
                timestamp = f'Message {i+1}'
            
            # Check for various signal types
            comparison_signal = self._check_comparisons(text)
            frustration_signal = self._check_frustration(text)
            disengagement_signal = self._check_disengagement(text)
            price_signal = self._check_price(text)
            
            msg_signals = []
            msg_risk = 0
            
            if comparison_signal['found']:
                msg_signals.append(comparison_signal)
                msg_risk += 25
                key_phrases.append(comparison_signal['text'])
                if breaking_point is None:
                    breaking_point = i + 1
            
            if frustration_signal['found']:
                msg_signals.append(frustration_signal)
                msg_risk += 30
                key_phrases.append(frustration_signal['text'])
                if breaking_point is None and msg_risk > 30:
                    breaking_point = i + 1
            
            if disengagement_signal['found']:
                msg_signals.append(disengagement_signal)
                msg_risk += 35
                key_phrases.append(disengagement_signal['text'])
                if breaking_point is None:
                    breaking_point = i + 1
            
            if price_signal['found']:
                msg_signals.append(price_signal)
                msg_risk += 20
                key_phrases.append(price_signal['text'])
            
            if msg_signals:
                signals.append({
                    'message_index': i + 1,
                    'timestamp': timestamp,
                    'signals': msg_signals,
                    'risk_score': min(100, msg_risk)
                })
                risk_scores.append(msg_risk)
        
        # Calculate overall risk
        overall_risk = max(risk_scores) if risk_scores else 0
        risk_level = self._assess_risk_level(overall_risk)
        
        # Generate recommendations
        recommendations = self._generate_recommendations(
            context, risk_level, signals, breaking_point
        )
        
        return {
            'signals': signals,
            'risk_level': risk_level,
            'confidence': min(100, len(signals) * 15),
            'breaking_point': breaking_point,
            'key_phrases': list(set(key_phrases))[:5],  # Top 5 unique phrases
            'recommendations': recommendations,
            'total_risk_score': min(100, overall_risk)
        }
    
    def _check_comparisons(self, text: str) -> Dict[str, Any]:
        """Check for competitor/alternative mentions."""
        for keyword in self.comparison_keywords:
            if keyword in text:
                return {
                    'found': True,
                    'type': 'COMPETITOR_COMPARISON',
                    'text': keyword,
                    'description': 'Comparing with alternatives or competitors'
                }
        return {'found': False}
    
    def _check_frustration(self, text: str) -> Dict[str, Any]:
        """Check for frustration indicators."""
        for keyword in self.frustration_keywords:
            if keyword in text:
                return {
                    'found': True,
                    'type': 'FRUSTRATION',
                    'text': keyword,
                    'description': 'Expressing dissatisfaction or frustration'
                }
        return {'found': False}
    
    def _check_disengagement(self, text: str) -> Dict[str, Any]:
        """Check for disengagement signals."""
        for keyword in self.disengagement_keywords:
            if keyword in text:
                return {
                    'found': True,
                    'type': 'DISENGAGEMENT',
                    'text': keyword,
                    'description': 'Showing intent to leave or end relationship'
                }
        return {'found': False}
    
    def _check_price(self, text: str) -> Dict[str, Any]:
        """Check for price-related concerns."""
        for keyword in self.price_keywords:
            if keyword in text:
                return {
                    'found': True,
                    'type': 'PRICE_CONCERN',
                    'text': keyword,
                    'description': 'Mentioning cost or pricing concerns'
                }
        return {'found': False}
    
    def _assess_risk_level(self, risk_score: float) -> str:
        """Assess overall risk level."""
        if risk_score >= 70:
            return "CRITICAL"
        elif risk_score >= 50:
            return "HIGH"
        elif risk_score >= 30:
            return "MEDIUM"
        else:
            return "LOW"
    
    def _generate_recommendations(self, context: str, risk_level: str,
                                 signals: List[Dict], breaking_point: int) -> List[str]:
        """Generate actionable recommendations."""
        recommendations = []
        
        if risk_level == "CRITICAL":
            recommendations.append("⚠️ URGENT: Immediate intervention required")
            if breaking_point:
                recommendations.append(f"Breaking point detected at message {breaking_point}")
        
        if context == "customer":
            if risk_level in ["CRITICAL", "HIGH"]:
                recommendations.append("Contact customer immediately to address concerns")
                recommendations.append("Prepare retention offer (discount/upgrade)")
                recommendations.append("Escalate to account manager")
        
        elif context == "employee":
            if risk_level in ["CRITICAL", "HIGH"]:
                recommendations.append("Schedule 1-on-1 with HR or manager")
                recommendations.append("Identify root cause of dissatisfaction")
                recommendations.append("Prepare retention plan")
        
        elif context == "investor":
            if risk_level in ["CRITICAL", "HIGH"]:
                recommendations.append("Prepare detailed response addressing concerns")
                recommendations.append("Schedule follow-up meeting")
        
        # Add general recommendations
        if any(s.get('type') == 'COMPETITOR_COMPARISON' for s in signals):
            recommendations.append("Counter competitive threats with unique value proposition")
        
        if any(s.get('type') == 'PRICE_CONCERN' for s in signals):
            recommendations.append("Review pricing strategy and alternative plans")
        
        return recommendations[:5]  # Top 5 recommendations
    
    def _empty_signals(self) -> Dict[str, Any]:
        """Return empty signals structure."""
        return {
            'signals': [],
            'risk_level': 'UNKNOWN',
            'confidence': 0,
            'breaking_point': None,
            'key_phrases': [],
            'recommendations': []
        }