import os import re import json import logging from collections import defaultdict from datetime import datetime from typing import Dict, Any, Optional, Union, List from urllib.parse import urljoin from connectors.firebase.firebase import Firebase import utility.function as func from ..models.base_node import BaseNode from ..models.execution_context import ExecutionContext from ..models.execution_result import ExecutionResult from ..integrations.http_client import HTTPClient, HTTPClientError logger = logging.getLogger(__name__) fb = Firebase(host=os.environ.get("FIREBASE_HOST")) class SMS(BaseNode): def __init__(self, property_id: str, node_id: str, name: str, config: Dict[str, Any], database_logger=None): """ Initialize API Call Node. Args: property_id: Property identifier for this node instance node_id: Unique identifier for this node instance name: Human-readable name for the node config: Node configuration containing HTTP request details database_logger: Optional database logger for node-level logging """ super().__init__(property_id, node_id, name, config, database_logger) self.node_type = "sms" self._http_client: Optional[HTTPClient] = None self.base_url = os.environ.get('INFOBIP_BASE_URL', '8v9rp3.api.infobip.com') def _get_http_client(self) -> HTTPClient: """ Get or create HTTP client instance. Returns: HTTPClient instance """ if self._http_client is None: timeout = self.config.get("timeout", 3600) default_headers = self.config.get("default_headers", {}) self._http_client = HTTPClient(timeout=timeout, default_headers=default_headers) return self._http_client def _get_content(self, content_id: str) -> List[Dict[str, Any]]: """ !!! IMPORTANT !!! This is a placeholder method for you to replace. It simulates fetching message content (e.g., from Firestore, a CMS, or a database) using the content_id from the node's configuration. Args: content_id: The identifier for the content to retrieve. context: The execution context, which may be needed to access databases. Returns: A list of valid SMS message objects. """ print(f"Fetching content for content_id: {content_id}") content = fb.db.reference().child(f"account/{self.property_id}/content/sms/{content_id}").get() if not content: raise ValueError(f"No Content {content_id} register in Database") if 'text' not in content: raise ValueError(f"No JSON object in Content {content_id}!") return content['text'] def _get_tel_number(self, user_pseudo_id) -> List: social_id = fb.db.reference().child(f"account/{self.property_id}/profile/{user_pseudo_id}/phoneNumber_PGP").get() if social_id: grouped = [] private_key = func.Key.load_key_from_env("private_key") for user in social_id.values(): user_id = user['id'] decode_id = func.Key.pgp_decrypt(user_id, private_key) grouped.append(decode_id) return {user_pseudo_id: grouped} return None def _manage_phone_numbers(self, grouped_phone) -> Dict: result = defaultdict(lambda: {"user_pseudo_ids": set(), "phone_numbers": set(), "mapping": {}}) if isinstance(grouped_phone, dict): grouped_phone = [grouped_phone] for entry in grouped_phone: if not isinstance(entry, dict): continue for user_id, phone_list in entry.items(): if not isinstance(phone_list, list): continue for phone in phone_list: result["default"]["user_pseudo_ids"].add(user_id) result["default"]["phone_numbers"].add(phone) result["default"]["mapping"][phone] = user_id return { sender_id: { "user_pseudo_ids": list(data["user_pseudo_ids"]), "phone_numbers": list(data["phone_numbers"]), "mapping": data["mapping"] } for sender_id, data in result.items() } def _get_sender_config(self, sender_id: str) -> Dict[str, Any]: """Get SMS sender configuration from Firebase""" sender_config = fb.db.reference().child(f"property/{self.property_id}/channel/sms/{sender_id}").get() if not sender_config: raise ValueError(f"No SMS sender {sender_id} found for property {self.property_id}") return sender_config def _execute_send_message(self, config: Dict[str, Any], channel_id:str, user_pseudo_ids_context:List) -> ExecutionResult: """ Handles the specific logic for sending SMS messages via Infobip API. """ content_id = config.get("content_id") channels = config.get("channels", []) print(f"user_pseudo_ids_context: {user_pseudo_ids_context}") if not content_id: return ExecutionResult(success=False, data={"error": "'content_id' not provided for message type."}) if not isinstance(channels, list) or not channels: return ExecutionResult(success=False, data={"error": "'channels' must be a non-empty list."}) content_text = self._get_content(content_id) if not content_text: return ExecutionResult(success=False,data={ f"{self.node_id}": { "user_pseudo_ids": [], "log_result": [] } } ) # Get sender configuration sender_id = channels[0] if channels else channel_id try: sender_config = self._get_sender_config(sender_id) except ValueError as e: return ExecutionResult(success=False, data={"error": str(e)}) # Get Infobip credentials infobip_api_key = os.environ.get('INFOBIP_API_KEY') infobip_base_url = os.environ.get('INFOBIP_BASE_URL', self.base_url) if not infobip_api_key: return ExecutionResult(success=False, data={"error": "INFOBIP_API_KEY not configured"}) # Prepare to send to each phone number http_client = self._get_http_client() push_api_url = f"https://{infobip_base_url}/sms/3/messages" phone_numbers = user_pseudo_ids_context.get('phone_numbers', []) mapping = user_pseudo_ids_context.get('mapping', {}) # Prepare headers headers = { "Authorization": f"App {infobip_api_key}", "Content-Type": "application/json" } logs = [] for phone in phone_numbers: try: payload = { "messages": [ { "sender": sender_config["sender_value"], "destinations": [{"to": phone}], "content": { "text": content_text } } ] } response = http_client.post(push_api_url, headers=headers, json=payload) print(f"Infobip SMS push: {response.status_code} {response.content}") logs.append({ "phone": phone, "status_code": response.status_code, "content": response.json() if hasattr(response, 'json') else str(response.content) }) except Exception as e: logs.append({ "phone": phone, "status_code": "EXCEPTION", "content": str(e) }) continue phone_success_ids = [log['phone'] for log in logs if log['status_code'] == 200] phone_fail_ids = [log['phone'] for log in logs if log['status_code'] != 200] user_pseudo_id_success = [mapping[j] for j in phone_success_ids if j in mapping] user_pseudo_id_fail = [mapping[j] for j in phone_fail_ids if j in mapping] return { "channel": channel_id, "status": "success", "user_pseudo_ids": user_pseudo_id_success, "social_uids": phone_success_ids, "total_social_uids": len(phone_success_ids), "total_user_pseudo_ids": len(user_pseudo_id_success), "user_pseudo_ids_fail": user_pseudo_id_fail, "total_user_pseudo_ids_fail": len(user_pseudo_id_fail) } def execute(self, context: ExecutionContext) -> ExecutionResult: start_time = datetime.utcnow() if not context.data["user_pseudo_ids"]: logger.error(f"No audience in node {self.node_id}") result = ExecutionResult.error_result( error_message=f"No audience in node", node_id=self.node_id, data={ f"{self.node_id}": { "user_pseudo_ids": [], "log_result": [] } } ) result.set_execution_time(start_time) return result print(f"Starting execution SMS. {start_time}") try: # Get Phone Numbers pack_phone_user_pseudo_ids = [] no_phone_id = [] user_pseudo_ids_success = [] for uid in context.data['user_pseudo_ids']: phone_data = self._get_tel_number(uid) if not phone_data: no_phone_id.append(uid) user_pseudo_ids_success.append(uid) pack_phone_user_pseudo_ids.append(phone_data) phones_context = self._manage_phone_numbers(pack_phone_user_pseudo_ids) if self.config['type'] == 'message': result_pack = [] for channel in self.config['channels']: if channel not in phones_context: print(f"No phone numbers in SMS channel {channel}") continue print(f"phones_context[channel]: {phones_context[channel]}") result = self._execute_send_message(self.config, channel, phones_context[channel]) if isinstance(result, ExecutionResult) and not result.success: continue success_user_pseudo_ids = result.get('user_pseudo_ids', []) fail_user_pseudo_ids = result.get('user_pseudo_ids_fail', []) status_groups = { "completed": success_user_pseudo_ids, "failed": fail_user_pseudo_ids, "no_phone_number": no_phone_id, } for status, users in status_groups.items(): result_pack.extend([ { "status": status, "channel": result.get("channel", channel), "user_pseudo_id": u, } for u in users ]) # Log result result = ExecutionResult.success_result( data={ f"{self.node_id}": { "user_pseudo_ids_no_phone_number": no_phone_id, "user_pseudo_ids": success_user_pseudo_ids, "user_pseudo_ids_failed": fail_user_pseudo_ids, "log_result": result_pack } }, node_id=self.node_id ) result.set_execution_time(start_time) logger.info(f"SMS sending completed successfully: {result}") return result elif self.config['type'] == 'audience': pass except Exception as e: print(f"Traceback for unexpected error in SMS node: {e}") logger.error(f"Error in SMS node execution: {e}") return ExecutionResult.error_result( error_message=str(e), node_id=self.node_id )