import os import requests import base64 from typing import Dict, Any, Optional from connectors.firebase.firebase import Firebase import logging logger = logging.getLogger(__name__) fb = Firebase(host=os.environ.get("FIREBASE_HOST")) def send_sms_via_thaibulksms( channel_config: Dict[str, Any], to_phone: str, message: str ) -> Dict[str, Any]: """ Send SMS via ThaiBulkSMS API Args: channel_config: SMS channel configuration from Firebase to_phone: Recipient phone number message: SMS message content Returns: Dict with success status and message """ try: api_key = channel_config.get("api_key") api_secret = channel_config.get("api_secret") sender_value = channel_config.get("sender_value", "") if not api_key or not api_secret: return {"success": False, "error": "ThaiBulkSMS API key or secret not configured"} # ThaiBulkSMS SMS API endpoint # Documentation: https://developer.thaibulksms.com/reference/post_sms api_url = "https://api-v2.thaibulksms.com/sms" # Prepare request with Basic Authentication # ThaiBulkSMS uses Basic Auth with api-key as username and api-secret as password credentials = f"{api_key}:{api_secret}" encoded_credentials = base64.b64encode(credentials.encode()).decode() headers = { "accept": "application/json", "content-type": "application/x-www-form-urlencoded", "authorization": f"Basic {encoded_credentials}" } # ThaiBulkSMS API payload format (form-urlencoded) # Parameters: msisdn (phone number), message, sender data = { "msisdn": to_phone, "message": message, "sender": sender_value } # Make API request (use data instead of json for form-urlencoded) response = requests.post(api_url, headers=headers, data=data, timeout=30) if response.status_code == 200: logger.info(f"SMS sent successfully via ThaiBulkSMS to {to_phone}") return {"success": True, "message": "SMS sent successfully", "response": response.json()} else: # Parse error response for better error message try: error_data = response.json() error_code = error_data.get("error", {}).get("code") error_name = error_data.get("error", {}).get("name", "") error_desc = error_data.get("error", {}).get("description", "") # Handle specific error codes if error_code == 111 or error_name == "ERROR_SENDER_NOT_FOUND": error_msg = ( f"ThaiBulkSMS API error: Sender '{sender_value}' not found. " f"Please ensure the sender ID is registered in your ThaiBulkSMS account. " f"Check your ThaiBulkSMS dashboard to verify registered senders." ) else: error_msg = f"ThaiBulkSMS API error: {response.status_code} - {error_desc or response.text}" except: error_msg = f"ThaiBulkSMS API error: {response.status_code} - {response.text}" logger.error(f"ThaiBulkSMS API error: {response.status_code} - {response.text}") return {"success": False, "error": error_msg} except requests.exceptions.RequestException as e: logger.error(f"ThaiBulkSMS API request error: {e}") return {"success": False, "error": f"API request failed: {str(e)}"} except Exception as e: logger.error(f"Error sending SMS via ThaiBulkSMS: {e}") return {"success": False, "error": f"Failed to send SMS: {str(e)}"} def send_sms_via_infobip( channel_config: Dict[str, Any], to_phone: str, message: str ) -> Dict[str, Any]: """ Send SMS via Infobip API Args: channel_config: SMS channel configuration from Firebase to_phone: Recipient phone number message: SMS message content Returns: Dict with success status and message """ try: infobip_api_key = os.environ.get('INFOBIP_API_KEY') infobip_base_url = os.environ.get('INFOBIP_BASE_URL', '8v9rp3.api.infobip.com') if not infobip_api_key: return {"success": False, "error": "INFOBIP_API_KEY not configured"} sender_value = channel_config.get("sender_value", "") api_url = f"https://{infobip_base_url}/sms/3/messages" headers = { "Authorization": f"App {infobip_api_key}", "Content-Type": "application/json" } payload = { "messages": [ { "sender": sender_value, "destinations": [{"to": to_phone}], "content": { "text": message } } ] } response = requests.post(api_url, json=payload, headers=headers, timeout=30) if response.status_code == 200: logger.info(f"SMS sent successfully via Infobip to {to_phone}") return {"success": True, "message": "SMS sent successfully", "response": response.json()} else: error_msg = f"Infobip API error: {response.status_code} - {response.text}" logger.error(error_msg) return {"success": False, "error": error_msg} except requests.exceptions.RequestException as e: logger.error(f"Infobip API request error: {e}") return {"success": False, "error": f"API request failed: {str(e)}"} except Exception as e: logger.error(f"Error sending SMS via Infobip: {e}") return {"success": False, "error": f"Failed to send SMS: {str(e)}"} def send_test_sms( property_id: str, content_id: str, channel_id: str, to_phone: str ) -> Dict[str, Any]: """ Send test SMS using specified template and channel Args: property_id: Property ID content_id: SMS template content ID channel_id: SMS channel ID to_phone: Recipient phone number Returns: Dict with success status and message/error """ try: # Get SMS template content content_ref = fb.db.reference().child(f"account/{property_id}/content/sms/{content_id}") template = content_ref.get() if not template: return {"success": False, "error": f"SMS template {content_id} not found"} message = template.get("message", "") if not message: return {"success": False, "error": "SMS template missing message"} # Get channel configuration channel_ref = fb.db.reference().child(f"property/{property_id}/channel/sms/{channel_id}") channel_config = channel_ref.get() if not channel_config: return {"success": False, "error": f"SMS channel {channel_id} not found"} if channel_config.get("status") != "active": return {"success": False, "error": f"SMS channel {channel_id} is not active"} provider = channel_config.get("provider", "infobip") # Route to appropriate sender if provider == "thaibulksms": result = send_sms_via_thaibulksms(channel_config, to_phone, message) elif provider == "infobip": result = send_sms_via_infobip(channel_config, to_phone, message) else: return {"success": False, "error": f"Unsupported SMS provider: {provider}"} return result except Exception as e: logger.error(f"Error in send_test_sms: {e}") return {"success": False, "error": f"Failed to send test SMS: {str(e)}"}