import logging from connectors.looker.lookerSDK import LookerSDK from datetime import datetime, timedelta from typing import Dict, Any, List from ..models.base_node import BaseNode from ..models.execution_context import ExecutionContext from ..models.execution_result import ExecutionResult # from feature.audience.audience_builder import AudinceBuilder logger = logging.getLogger(__name__) class SwitchCaseNode(BaseNode): """ Switch Case Node for conditional branching in workflows. Evaluates multiple conditions and routes execution to the first matching branch. Supports a default branch for cases where no conditions match. """ def __init__(self, property_id: str, node_id: str, name: str, config: Dict[str, Any], database_logger=None): """ Initialize the Switch Case Node. Args: property_id: Property identifier for this node instance node_id: Unique identifier for the node name: Human-readable name for the node config: Node configuration containing cases and default branch database_logger: Optional database logger for node-level logging """ super().__init__(property_id, node_id, name, config, database_logger) self.node_type = "switchcase" self.looker = LookerSDK(property_id=self.property_id) # self.condition_evaluator = ConditionEvaluator() # Validate configuration during initialization if not self.validate_config(): raise ValueError(f"Invalid configuration for SwitchCaseNode {node_id}") def validate_config(self) -> bool: """ Validate the node's configuration. Returns: True if configuration is valid, False otherwise """ config = getattr(self, "config", None) if not isinstance(config, dict): return False # Validate required top-level keys if not all(k in config for k in ("option", "conditions")): return False if not isinstance(config["option"], str): return False if not isinstance(config["conditions"], list): return False for cond in config["conditions"]: if not isinstance(cond, dict): return False # Validate required fields for key in ("id", "name", "type"): if key not in cond or not isinstance(cond[key], str): return False cond_type = cond["type"] if cond_type == "condition": # Must have "condition" (list), "target" (str) if not isinstance(cond.get("condition"), list): return False if not isinstance(cond.get("target"), str): return False # Optional description must be string if exists desc = cond.get("description") if desc is not None and not isinstance(desc, str): return False # Validate condition rules for rule in cond["condition"]: if not isinstance(rule, dict): return False for key in ("dimension", "condition", "value"): if key not in rule: return False if not isinstance(rule["dimension"], str): return False if not isinstance(rule["condition"], str): return False if not isinstance(rule["value"], list): return False # Special rule for "between" if rule["condition"] == "between": if len(rule["value"]) != 2: return False for v in rule["value"]: try: datetime.strptime(v, "%Y/%m/%d") except ValueError: return False elif cond_type == "others": # No extra validation needed for "others" pass else: # Invalid type return False return True def _fetch_node_data(self, view, filters): """A helper method to fetch data for a single node.""" conditions = self.looker.build_filter_expression([filters]) data = self.looker.build_audience(view, conditions, False) user_list = data[0].get('event.list_of_user', []) return user_list if user_list else [] def execute(self, context: ExecutionContext) -> ExecutionResult: try: 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 ) result.set_execution_time(start_time) return result context_user_pseudo_ids = context.data["user_pseudo_ids"] context_user_pseudo_ids_set = set(context_user_pseudo_ids) option = self.get_config_value('option') conditions = self.get_config_value('conditions', []) view = self.get_config_value('view', 'api_event') conditionsNew = [] if option not in ['priority', 'all']: pass # Execute audience before check for con in conditions: condition = con.get('condition', []) condition_type = con['type'] if condition_type == 'others': user_query = [] else: user_query = self._fetch_node_data(view, condition) con['user_pseudo_ids'] = user_query conditionsNew.append(con) conditions = conditionsNew conditionsNew = [] if option == 'priority': pass_user_pseudo_ids = [] for con in conditions: user_pseudo_ids = con.get('user_pseudo_ids', []) user_pseudo_ids_set = set(user_pseudo_ids) common = context_user_pseudo_ids_set.intersection(user_pseudo_ids_set) pass_user_pseudo_ids_set = set(pass_user_pseudo_ids) _final = common - pass_user_pseudo_ids_set con['user_pseudo_ids'] = list(_final) conditionsNew.append(con) pass_user_pseudo_ids.extend(list(common)) others_context_user_pseudo_ids = context_user_pseudo_ids_set - set(pass_user_pseudo_ids) elif option == 'all': pass_user_pseudo_ids = [] for con in conditions: user_pseudo_ids = con.get('user_pseudo_ids', []) user_pseudo_ids_set = set(user_pseudo_ids) common = context_user_pseudo_ids_set.intersection(user_pseudo_ids_set) con['user_pseudo_ids'] = list(common) conditionsNew.append(con) pass_user_pseudo_ids.extend(list(common)) others_context_user_pseudo_ids = context_user_pseudo_ids_set - set(pass_user_pseudo_ids) print(f"others_context_user_pseudo_ids: {others_context_user_pseudo_ids}") # Update condition else for i in conditionsNew: if i['id'] == 'others': i['user_pseudo_ids'] = list(others_context_user_pseudo_ids) if others_context_user_pseudo_ids else [] #Build result log_result = [] condition_context = {} for con in conditionsNew: target = con.get('target') condition_id = con.get('id', '-') user_pseudo_ids = con.get('user_pseudo_ids', []) # if not user_pseudo_ids: # continue condition_context.update( { target: { 'user_pseudo_ids': user_pseudo_ids } } ) for user in con['user_pseudo_ids']: data = { "status": 'completed', "target": target, "condition_id": condition_id, "message": f'Go to Node {target}', "user_pseudo_id": user, "start_time": start_time.isoformat() } log_result.append(data) print() print('condition_context') print(condition_context) print() print(log_result) # Create success result result = ExecutionResult.success_result( data={ "user_pseudo_ids": context_user_pseudo_ids, f"{self.node_id}": { "user_pseudo_ids": context_user_pseudo_ids, "log_result": log_result, "conditions": condition_context } }, node_id=self.node_id ) result.set_execution_time(start_time) print(f"Switch case completed successfully: {result.data}") return result except Exception as e: logging.error(e)