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 Email(BaseNode): def __init__(self, property_id: str, node_id: str, name: str, config: Dict[str, Any], database_logger=None): """ Initialize Email 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 = "email" self._http_client: Optional[HTTPClient] = None self.base_url = os.environ.get('INFOBIP_BASE_URL', '8v9rp3.api.infobip.com') def validate_config(self) -> bool: return True 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) -> Dict[str, Any]: """ Fetch email content from Firebase. Args: content_id: The identifier for the content to retrieve. Returns: Email content with subject, text, and html """ print(f"Fetching content for content_id: {content_id}") content = fb.db.reference().child(f"account/{self.property_id}/content/email/{content_id}").get() if not content: raise ValueError(f"No Content {content_id} register in Database") if 'subject' not in content: raise ValueError(f"No subject in Content {content_id}!") return { "subject": content.get('subject', ''), "text": content.get('text', ''), "html": content.get('html', '') } def _get_email(self, user_pseudo_id) -> List: """Get and decrypt email addresses for a user""" social_id = fb.db.reference().child(f"account/{self.property_id}/profile/{user_pseudo_id}/email_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_email_addresses(self, grouped_email) -> Dict: """Group email addresses by sender_id""" result = defaultdict(lambda: {"user_pseudo_ids": set(), "email_addresses": set(), "mapping": {}}) if isinstance(grouped_email, dict): grouped_email = [grouped_email] for entry in grouped_email: if not isinstance(entry, dict): continue for user_id, email_list in entry.items(): if not isinstance(email_list, list): continue for email in email_list: result["default"]["user_pseudo_ids"].add(user_id) result["default"]["email_addresses"].add(email) result["default"]["mapping"][email] = user_id return { sender_id: { "user_pseudo_ids": list(data["user_pseudo_ids"]), "email_addresses": list(data["email_addresses"]), "mapping": data["mapping"] } for sender_id, data in result.items() } def _get_sender_config(self, sender_id: str) -> Dict[str, Any]: """Get Email sender configuration from Firebase""" sender_config = fb.db.reference().child(f"property/{self.property_id}/channel/email/{sender_id}").get() if not sender_config: raise ValueError(f"No Email 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 Email 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 = self._get_content(content_id) if not content: 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 email address http_client = self._get_http_client() push_api_url = f"https://{infobip_base_url}/email/3/send" email_addresses = user_pseudo_ids_context.get('email_addresses', []) mapping = user_pseudo_ids_context.get('mapping', {}) # Prepare headers headers = { "Authorization": f"App {infobip_api_key}", "Content-Type": "application/json" } logs = [] for email in email_addresses: try: payload = { "from": sender_config["from_email"], "to": [email], "subject": content["subject"], "text": content.get("text", ""), "html": content.get("html", "") } response = http_client.post(push_api_url, headers=headers, json=payload) print(f"Infobip Email push: {response.status_code} {response.content}") logs.append({ "email": email, "status_code": response.status_code, "content": response.json() if hasattr(response, 'json') else str(response.content) }) except Exception as e: logs.append({ "email": email, "status_code": "EXCEPTION", "content": str(e) }) continue email_success_ids = [log['email'] for log in logs if log['status_code'] == 200] email_fail_ids = [log['email'] for log in logs if log['status_code'] != 200] user_pseudo_id_success = [mapping[j] for j in email_success_ids if j in mapping] user_pseudo_id_fail = [mapping[j] for j in email_fail_ids if j in mapping] return { "channel": channel_id, "status": "success", "user_pseudo_ids": user_pseudo_id_success, "social_uids": email_success_ids, "total_social_uids": len(email_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 Email. {start_time}") try: # Get Email Addresses pack_email_user_pseudo_ids = [] no_email_id = [] user_pseudo_ids_success = [] for uid in context.data['user_pseudo_ids']: email_data = self._get_email(uid) if not email_data: no_email_id.append(uid) user_pseudo_ids_success.append(uid) pack_email_user_pseudo_ids.append(email_data) emails_context = self._manage_email_addresses(pack_email_user_pseudo_ids) if self.config['type'] == 'message': result_pack = [] for channel in self.config['channels']: if channel not in emails_context: print(f"No email addresses in Email channel {channel}") continue print(f"emails_context[channel]: {emails_context[channel]}") result = self._execute_send_message(self.config, channel, emails_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_email_address": no_email_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_email_address": no_email_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"Email sending completed successfully: {result}") return result elif self.config['type'] == 'audience': pass except Exception as e: print(f"Traceback for unexpected error in Email node: {e}") logger.error(f"Error in Email node execution: {e}") return ExecutionResult.error_result( error_message=str(e), node_id=self.node_id )