import os from typing import List, Dict, Any, Optional from dataclasses import dataclass from dotenv import load_dotenv load_dotenv() @dataclass class EmailSummary: summary: str urgency_level: str action_required: str confidence: float needs_response: bool = True class AIAnalyzer: def __init__(self): self.api_key = os.getenv("GROQ_API_KEY") self.model = "llama3-8b-8192" # Check if API key is available if not self.api_key or self.api_key == "your_groq_api_key_here": print("⚠️ GROQ_API_KEY not configured. AI analysis will be disabled.") self.client = None return try: from groq import Groq self.client = Groq(api_key=self.api_key) print("✅ Groq AI client initialized successfully") except Exception as e: print(f"⚠️ Failed to initialize Groq client: {e}") self.client = None def analyze_thread_context(self, thread_messages: List[Dict[str, Any]]) -> EmailSummary: """Analyze email thread context and generate summary""" if not thread_messages: return EmailSummary("No messages", "low", "none", 0.0, False) # If AI client is not available, use basic analysis if not self.client: return self._basic_analysis(thread_messages) # Prepare context for analysis context = self._prepare_thread_context(thread_messages) prompt = f""" Analyze this email and determine if it requires a response. Be selective and only mark as actionable if the email genuinely needs a reply. Consider: 1. Is this a real request/question that needs an answer? 2. Is this from a real person (not automated/marketing/promotional)? 3. Does this require specific action or information? 4. Is this urgent or time-sensitive? 5. Is this a complaint, inquiry, or request for service? 6. Does this require follow-up or acknowledgment? 7. Is this a business-related email that needs attention? 8. Is this from a client, customer, or stakeholder? IMPORTANT: DO NOT mark as actionable if the email is: - Marketing or promotional content - Automated notifications or updates - Newsletter or subscription content - System-generated messages - General announcements that don't require action Thread Context: {context} IMPORTANT: Respond ONLY in this exact format (no extra text, no explanations): SUMMARY: [2-3 sentence summary] URGENCY: [low/medium/high/critical] ACTION: [specific action needed or "no response needed"] CONFIDENCE: [0.0-1.0] NEEDS_RESPONSE: [true/false] """ try: response = self.client.chat.completions.create( model=self.model, messages=[{"role": "user", "content": prompt}], max_tokens=300, temperature=0.3 ) result = response.choices[0].message.content parsed_result = self._parse_ai_response(result) return parsed_result except Exception as e: print(f"AI analysis error: {e}") # Return a default response that indicates no action needed return EmailSummary("AI analysis failed", "low", "Review manually", 0.0, False) def _prepare_thread_context(self, messages: List[Dict[str, Any]]) -> str: """Prepare thread context for AI analysis""" context_parts = [] for i, msg in enumerate(messages[-4:], 1): # Last 4 messages sender = msg.get('from', 'Unknown') subject = msg.get('subject', 'No subject') snippet = msg.get('snippet', '') date = msg.get('date', '') context_parts.append(f"Message {i} ({date}):") context_parts.append(f"From: {sender}") context_parts.append(f"Subject: {subject}") context_parts.append(f"Content: {snippet}") context_parts.append("") return "\n".join(context_parts) def _parse_ai_response(self, response: str) -> EmailSummary: """Parse AI response into structured format""" lines = response.split('\n') summary = "No summary available" urgency = "medium" action = "Review manually" confidence = 0.5 needs_response = True for line in lines: line = line.strip() # Simple parsing for consistent format if line.startswith("SUMMARY:"): summary = line.replace("SUMMARY:", "").strip() elif line.startswith("URGENCY:"): urgency = line.replace("URGENCY:", "").strip().lower() elif line.startswith("ACTION:"): action = line.replace("ACTION:", "").strip() elif line.startswith("CONFIDENCE:"): try: confidence_text = line.replace("CONFIDENCE:", "").strip() confidence = float(confidence_text) except: confidence = 0.5 elif line.startswith("NEEDS_RESPONSE:"): needs_response_text = line.replace("NEEDS_RESPONSE:", "").strip().lower() needs_response = needs_response_text in ["true", "yes", "1"] return EmailSummary(summary, urgency, action, confidence, needs_response) def _basic_analysis(self, thread_messages: List[Dict[str, Any]]) -> EmailSummary: """Basic email analysis when AI is not available""" if not thread_messages: return EmailSummary("No messages", "low", "none", 0.0, False) # Get the latest email latest_email = thread_messages[-1] from_email = latest_email.get('from', '').lower() subject = latest_email.get('subject', '').lower() snippet = latest_email.get('snippet', '').lower() # Basic filtering logic non_actionable_keywords = [ 'newsletter', 'promotion', 'unsubscribe', 'confirm your email', 'password reset', 'no-reply', 'noreply', 'marketing', 'advertisement' ] # Check if email is from own domain if 'projects@manaknightdigital.com' in from_email: return EmailSummary("Own email - no action needed", "low", "none", 0.9, False) # Check for non-actionable keywords text = f"{from_email} {subject} {snippet}".lower() for keyword in non_actionable_keywords: if keyword in text: return EmailSummary(f"Automated email detected ({keyword})", "low", "none", 0.8, False) # Check for actionable indicators actionable_indicators = ['?', 'can you', 'could you', 'please', 'help', 'urgent', 'asap'] action_score = sum(1 for indicator in actionable_indicators if indicator in text) if action_score > 0: return EmailSummary("Email requires response", "medium", "Reply needed", 0.7, True) # Default to no action needed return EmailSummary("Standard email - review if needed", "low", "Review manually", 0.5, False) def generate_alert_message(self, thread_id: str, summary: EmailSummary, alert_level: int, email_data: Dict[str, Any] = None) -> str: """Generate formatted alert message for WhatsApp""" alert_levels = { 1: "🚨 LEVEL 1 ALERT (1-24 Hours)", 2: "🚨🚨 LEVEL 2 ALERT (24-48 Hours - URGENT)", 3: "🚨🚨🚨 LEVEL 3 ALERT (48+ Hours - CRITICAL)" } urgency_icons = { "low": "🟢", "medium": "🟡", "high": "🟠", "critical": "🔴" } # Extract email details if provided if email_data: sender = email_data.get('from', 'Unknown') subject = email_data.get('subject', 'No subject') date = email_data.get('date', 'Unknown time') body = email_data.get('snippet', 'No content') # Format the date nicely try: from datetime import datetime if isinstance(date, str): # Try to parse the date parsed_date = datetime.fromisoformat(date.replace('Z', '+00:00')) formatted_date = parsed_date.strftime('%Y-%m-%d %H:%M') else: formatted_date = str(date) except: formatted_date = str(date) else: sender = "Unknown" subject = "No subject" formatted_date = "Unknown time" body = "No content" message = f""" {alert_levels.get(alert_level, "ALERT")} {urgency_icons.get(summary.urgency_level, "⚪")} Urgency: {summary.urgency_level.upper()} 📧 Thread ID: {thread_id} 📧 Email Details: 👤 From: {sender} 📋 Subject: {subject} ⏰ Sent: {formatted_date} 📄 Body: {body[:200]}{'...' if len(body) > 200 else ''} 📝 AI Summary: {summary.summary} 🎯 Action Required: {summary.action_required} """.strip() return message if __name__ == "__main__": # Test with mock data analyzer = AIAnalyzer() mock_thread = [ { 'from': 'client@example.com', 'subject': 'Login issue follow-up', 'snippet': 'I\'m still having trouble with the login system. When will this be resolved?', 'date': '2024-01-15T10:30:00' }, { 'from': 'support@company.com', 'subject': 'Re: Login issue follow-up', 'snippet': 'We\'re investigating the issue. Will update you soon.', 'date': '2024-01-15T11:00:00' }, { 'from': 'client@example.com', 'subject': 'Re: Login issue follow-up', 'snippet': 'This is urgent - I need access today. Can you please expedite?', 'date': '2024-01-15T14:00:00' } ] summary = analyzer.analyze_thread_context(mock_thread) print("AI Analysis Result:") print(f"Summary: {summary.summary}") print(f"Urgency: {summary.urgency_level}") print(f"Action: {summary.action_required}") print(f"Confidence: {summary.confidence:.1%}")