""" Audience Node implementation for Firebase integration. """ from typing import Dict, Any, List import logging from datetime import datetime from ..models.base_node import BaseNode from ..models.execution_context import ExecutionContext from ..models.execution_result import ExecutionResult from ..integrations.firebase_client import FirebaseClient class AudienceNode(BaseNode): """ Node that retrieves audience data from Firebase Realtime Database. Queries Firebase using audience_id to retrieve user_pseudo_id lists and passes them to subsequent nodes in the workflow context. """ def __init__(self, property_id: str, node_id: str, name: str, config: Dict[str, Any], database_logger=None): """ Initialize Audience 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 audience_id and Firebase settings database_logger: Optional database logger for node-level logging """ super().__init__(property_id, node_id, name, config, database_logger) self.node_type = "audience" self._firebase_client = None def get_required_config_keys(self) -> List[str]: """ Get list of required configuration keys for this node type. Returns: List of required configuration keys """ return ["audience_id"] def validate_config(self) -> bool: """ Validate the node's configuration. Returns: True if configuration is valid, False otherwise """ # Check required keys if not self.validate_required_config(): return False # Validate audience_id audience_id = self.config.get("audience_id") if not isinstance(audience_id, str) or not audience_id.strip(): return False # Validate firebase_path if provided firebase_path = self.config.get("firebase_path") if firebase_path is not None and not isinstance(firebase_path, str): return False # Validate database_url if provided database_url = self.config.get("database_url") if database_url is not None and not isinstance(database_url, str): return False # Validate credentials_path if provided credentials_path = self.config.get("credentials_path") if credentials_path is not None and not isinstance(credentials_path, str): return False return True def _get_firebase_client(self) -> FirebaseClient: """ Get or create Firebase client instance. Returns: FirebaseClient instance Raises: ValueError: If Firebase configuration is invalid """ if self._firebase_client is None: self._firebase_client = FirebaseClient() return self._firebase_client def execute(self, context: ExecutionContext) -> ExecutionResult: """ Execute the audience node to retrieve user data from Firebase. Args: context: Current execution context containing workflow state and data Returns: ExecutionResult with user_pseudo_ids list or error information """ start_time = datetime.utcnow() try: # Validate configuration before execution if not self.validate_config(): return ExecutionResult.error_result( error_message="Invalid node configuration", node_id=self.node_id ) audience_id = self.config["audience_id"] firebase_path = self.config.get("firebase_path") logging.info(f"Retrieving audience data for audience_id: {audience_id}") # Get Firebase client firebase_client = self._get_firebase_client() # Retrieve user_pseudo_ids from Firebase user_pseudo_ids = firebase_client.get_audience_users(self.property_id, audience_id) #Database logger # for user in user_pseudo_ids: # # data = { # # "execution_id": context.execution_id, # # "node_id": self.node_id, # # "workflow_id": context.workflow_id, # # "user_pseudo_id": user, # # "status": "completed" # # } # try: # self.database_logger.log_user_pseudo_id( # execution_id=context.execution_id, # node_id=self.node_id, # workflow_id=context.workflow_id, # user_pseudo_id=user, # status="completed" # ) # except Exception as e: # print(f"Failed to log_user_pseudo_id: {str(e)}") # Create successful result with user data result = ExecutionResult.success_result( data={ "user_pseudo_ids": user_pseudo_ids, f"{self.node_id}": { "user_pseudo_ids": user_pseudo_ids, "audience_id": audience_id, "user_count": len(user_pseudo_ids), "log_result": [ { "status": "completed", "user_pseudo_id": u, "audience_id": audience_id } for u in user_pseudo_ids ] } }, node_id=self.node_id ) # Update execution context with audience data context.set_variable("user_pseudo_ids", user_pseudo_ids) context.set_variable("audience_id", audience_id) context.set_variable("user_count", len(user_pseudo_ids)) # # Log audience retrieval if database logger is available # if self.database_logger: # try: # for user_pseudo_id in user_pseudo_ids: # self.database_logger.log_user_pseudo_id( # workflow_id=context.data.get('workflow_id'), # execution_id=context.data.get('execution_id'), # node_id=self.node_id, # user_pseudo_id=user_pseudo_id, # status="audience_retrieved" # ) # except Exception as e: # logging.warning(f"Failed to log audience users: {e}") # logging.info(f"Successfully retrieved {len(user_pseudo_ids)} users for audience: {audience_id}") except ValueError as e: logging.error(f"Configuration error in audience node {self.node_id}: {e}") result = ExecutionResult.error_result( error_message=f"Configuration error: {str(e)}", node_id=self.node_id ) except ConnectionError as e: logging.error(f"Firebase connection error in audience node {self.node_id}: {e}") result = ExecutionResult.error_result( error_message=f"Firebase connection failed: {str(e)}", node_id=self.node_id ) except Exception as e: logging.error(f"Unexpected error in audience node {self.node_id}: {e}") result = ExecutionResult.error_result( error_message=f"Unexpected error: {str(e)}", node_id=self.node_id ) # Set execution time result.set_execution_time(start_time) return result def test_firebase_connection(self) -> bool: """ Test Firebase connection without executing the full node. Returns: True if connection is successful, False otherwise """ try: firebase_client = self._get_firebase_client() return firebase_client.test_connection() except Exception as e: logging.error(f"Firebase connection test failed: {e}") return False def get_audience_preview(self, limit: int = 10) -> List[str]: """ Get a preview of audience users (limited number). Args: limit: Maximum number of users to return Returns: List of user_pseudo_ids (limited) Raises: ValueError: If configuration is invalid ConnectionError: If Firebase connection fails """ if not self.validate_config(): raise ValueError("Invalid node configuration") audience_id = self.config["audience_id"] firebase_path = self.config.get("firebase_path") firebase_client = self._get_firebase_client() user_pseudo_ids = firebase_client.get_audience_users(self.property_id, audience_id) return user_pseudo_ids[:limit] def close_firebase_connection(self): """Close Firebase connection if it exists.""" if self._firebase_client: self._firebase_client.close() self._firebase_client = None