Iyeoluwa Akinrinola
121 lines
3.9 KiB
Python
121 lines
3.9 KiB
Python
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}") |