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" if not self.api_key: raise ValueError("GROQ_API_KEY is required. Please add it to your .env file") try: from groq import Groq self.client = Groq(api_key=self.api_key) print("✅ Groq AI client initialized successfully") except Exception as e: raise RuntimeError(f"Failed to initialize Groq client: {e}") 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) # 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 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%}")