import os from typing import List, Dict, Any from twilio.rest import Client from twilio.base.exceptions import TwilioException from dotenv import load_dotenv load_dotenv() class WhatsAppSender: def __init__(self): self.account_sid = os.getenv("TWILIO_ACCOUNT_SID") self.auth_token = os.getenv("TWILIO_AUTH_TOKEN") self.from_number = os.getenv("TWILIO_WHATSAPP_NUMBER") self.to_number = os.getenv("WHATSAPP_TO_NUMBER") # Individual phone number if self.account_sid and self.auth_token: try: self.client = Client(self.account_sid, self.auth_token) self.use_mock = False except Exception as e: print(f"Warning: Twilio client failed to initialize: {e}") self.use_mock = True else: self.use_mock = True print("Note: Using mock WhatsApp sender (add Twilio credentials to .env)") # Use real WhatsApp mode self.use_mock = False print("📱 Using WhatsApp mode") def send_alert(self, alert_message: str, thread_id: str = None) -> Dict[str, Any]: """Send alert message to WhatsApp""" if self.use_mock: return self._mock_send(alert_message, thread_id) try: # Format message for WhatsApp formatted_message = self._format_message(alert_message) # Send to WhatsApp message = self.client.messages.create( from_=f"whatsapp:{self.from_number}", body=formatted_message, to=f"whatsapp:{self.to_number}" ) return { 'status': 'success', 'message_sid': message.sid, 'thread_id': thread_id, 'sent_at': message.date_created } except TwilioException as e: print(f"WhatsApp send error: {e}") return { 'status': 'error', 'error': str(e), 'thread_id': thread_id } def _mock_send(self, alert_message: str, thread_id: str = None) -> Dict[str, Any]: """Mock WhatsApp sending for testing""" print(f"📱 [MOCK] WhatsApp Alert Sent:") print(f" To: {self.to_number or 'your_number'}") print(f" Thread ID: {thread_id}") print(f" Message: {alert_message[:100]}...") return { 'status': 'success', 'message_sid': 'mock_sid_123', 'thread_id': thread_id, 'sent_at': '2024-01-15T10:00:00Z' } def _format_message(self, alert_message: str) -> str: """Format alert message for WhatsApp""" # WhatsApp has character limits, so we might need to truncate max_length = 1000 if len(alert_message) > max_length: alert_message = alert_message[:max_length-3] + "..." return alert_message def send_bulk_alerts(self, alerts: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Send multiple alerts to WhatsApp""" results = [] for alert in alerts: message = alert.get('message', '') thread_id = alert.get('thread_id', 'unknown') result = self.send_alert(message, thread_id) results.append(result) # Add small delay between messages to avoid rate limits import time time.sleep(1) return results if __name__ == "__main__": # Test WhatsApp sender sender = WhatsAppSender() test_message = """ 🚨 LEVEL 1 ALERT (24 Hours) 🟢 Urgency: LOW 📧 Thread ID: test_thread_123 📝 Summary: Client inquiry about project status. Requires follow-up. 🎯 Action Required: Respond to client question ⏰ Confidence: 70.0% """.strip() result = sender.send_alert(test_message, "test_thread_123") print(f"Send result: {result}")