""" TriggerNode implementation for workflow initiation with simple interval scheduling. """ from datetime import datetime, timedelta, timezone 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 ..utils.interval_scheduler import IntervalScheduler import pytz # timezone = pytz.timezone('Asia/Bangkok') class TriggerNode(BaseNode): """ Trigger node that initiates workflow execution based on simple interval schedules. Supports interval types: minutes, hours, days, weeks, months with numeric values. Evaluates simple interval schedule and triggers workflow start. """ def __init__(self, property_id: str, node_id: str, name: str, config: list, database_logger=None): """ Initialize trigger 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 interval settings database_logger: Optional database logger for node-level logging """ super().__init__(property_id, node_id, name, config, database_logger) self.node_type = "trigger" self.scheduler = IntervalScheduler() def execute(self, context: ExecutionContext) -> ExecutionResult: """ Execute the trigger node with interval-based scheduling logic. Args: context: Current execution context containing workflow state and data Returns: ExecutionResult with trigger metadata and scheduling information """ start_time = datetime.utcnow() try: # Get interval configuration interval_type = self.get_config_value("interval_type") interval_value = self.get_config_value("interval_value") # Validate configuration if not self.validate_config(): return ExecutionResult.error_result( error_message="Invalid trigger configuration", node_id=self.node_id ) # Get last execution time from context or use current time last_execution = context.get_variable("last_trigger_execution") if last_execution and isinstance(last_execution, str): try: last_execution = datetime.fromisoformat(last_execution) except ValueError: last_execution = None # Calculate next execution time next_execution = self.scheduler.calculate_next_execution( interval_type, interval_value, last_execution ) # Check if execution is due (for scheduling purposes) current_time = datetime.utcnow() is_due = self.scheduler.is_execution_due( interval_type, interval_value, last_execution or current_time, current_time ) # Create trigger metadata trigger_data = { "trigger_type": "interval", "interval_type": interval_type, "interval_value": interval_value, "current_time": current_time.isoformat(), "last_execution": last_execution.isoformat() if last_execution else None, "next_execution": next_execution.isoformat(), "is_due": is_due, "timezone": self.get_config_value("timezone", "UTC"), "enabled": self.get_config_value("enabled", True) } # Add trigger metadata to context context.set_variable("trigger_metadata", trigger_data) context.set_variable("last_trigger_execution", current_time.isoformat()) # Create successful result result = ExecutionResult.success_result( data=trigger_data, node_id=self.node_id ) result.set_execution_time(start_time) return result except Exception as e: error_result = ExecutionResult.error_result( error_message=f"Trigger execution failed: {str(e)}", node_id=self.node_id ) error_result.set_execution_time(start_time) return error_result def validate_config(self) -> bool: """ Validate the trigger node's configuration. Returns: True if configuration is valid, False otherwise """ return True def get_required_config_keys(self) -> List[str]: """ Get list of required configuration keys for trigger node. Returns: List of required configuration keys """ return ["interval_type", "interval_value"] def is_enabled(self) -> bool: """ Check if the trigger is enabled. Returns: True if trigger is enabled, False otherwise """ return self.get_config_value("enabled", True) def get_interval_type(self) -> str: """ Get the interval type for this trigger. Returns: Interval type string """ return self.get_config_value("interval_type") def get_interval_value(self) -> int: """ Get the interval value for this trigger. Returns: Interval value as integer """ return self.get_config_value("interval_value") def get_timezone(self) -> str: """ Get the timezone for this trigger. Returns: Timezone string (defaults to UTC) """ return self.get_config_value("timezone", "UTC") def calculate_next_execution(self, last_execution: datetime = None) -> datetime: """ Calculate the next execution time for this trigger. Args: last_execution: Last execution time (defaults to now if None) Returns: Next execution datetime """ return self.scheduler.calculate_next_execution( self.get_interval_type(), self.get_interval_value(), last_execution ) def is_execution_due(self, last_execution: datetime, current_time: datetime = None) -> bool: """ Check if execution is due based on the trigger configuration. Args: last_execution: Last execution time current_time: Current time (defaults to now if None) Returns: True if execution is due, False otherwise """ return self.scheduler.is_execution_due( self.get_interval_type(), self.get_interval_value(), last_execution, current_time ) def to_utc(self, dt: datetime, tz_offset_hours: int): """ Convert a datetime from a known timezone offset (in hours) to UTC. Works with both naive and timezone-aware datetimes. """ if dt.tzinfo is None: # Attach known timezone if naive local_tz = timezone(timedelta(hours=int(tz_offset_hours))) dt = dt.replace(tzinfo=local_tz) return dt.astimezone(timezone.utc) # def generate_schedule(self, datebegin:str, dateend:str, timezone_setting:int=0): # """Generate list of times this node should run""" # # if self.alwaywhen == "on": # # return [] # always trigger, no need for specific schedule # print(f'TIMEZOINE FROM SETTING {timezone_setting}') # self.datebegin = datetime.strptime(datebegin, "%Y-%m-%d %H:%M:%S") # self.dateend = datetime.strptime(dateend, "%Y-%m-%d %H:%M:%S") # print(f"This os begin {self.datebegin}") # schedule = [] # for interval in self.config: # current = self.datebegin # while current <= self.dateend: # field = interval.get("field") # trigger_hour = interval.get("triggerAtHour", 0) # trigger_minute = interval.get("triggerAtMinute", 0) # if field == "minutes": # minutes_interval = interval.get("minutesInterval", None) # if minutes_interval is not None: # current = self.datebegin.replace(second=0, microsecond=0) # while current <= self.dateend: # schedule.append(self.to_utc(current, timezone_setting)) # current += timedelta(minutes=minutes_interval) # elif field == "hours": # hours_interval = interval.get("hoursInterval", None) # if hours_interval is not None: # current = self.datebegin.replace(minute=trigger_minute, second=0, microsecond=0) # while current <= self.dateend: # schedule.append(self.to_utc(current, timezone_setting)) # current += timedelta(hours=hours_interval) # elif field == "days": # days_interval = interval.get("daysInterval", None) # if days_interval is not None: # current = self.datebegin.replace(hour=trigger_hour, minute=trigger_minute, second=0, microsecond=0) # if current < self.datebegin: # current = self.datebegin.replace(hour=trigger_hour, minute=trigger_minute, second=0, microsecond=0) # while current <= self.dateend: # schedule.append(self.to_utc(current, timezone_setting)) # current += timedelta(days=days_interval) # if not schedule and self.datebegin.date() == self.dateend.date(): # schedule.append(self.to_utc(self.datebegin.replace( # hour=trigger_hour, minute=trigger_minute, second=0, microsecond=0 # ), timezone_setting)) # elif field == "weeks": # weeks_interval = interval.get("weeksInterval", None) # trigger_days = interval.get("triggerAtDay", []) # if weeks_interval is not None: # current = self.datebegin.replace(hour=trigger_hour, minute=trigger_minute, second=0, microsecond=0) # while current <= self.dateend: # if current.weekday() in trigger_days: # schedule.append(self.to_utc(current, timezone_setting)) # current += timedelta(days=1) # elif field == "months": # months_interval = interval.get("monthsInterval", None) # trigger_day_of_month = interval.get("triggerAtDayOfMonth", [1]) # if months_interval is not None: # reference_month = self.last_triggered_datetime.month # months_since = (current.year - self.last_triggered_datetime.year) * 12 + (current.month - reference_month) # if months_since % months_interval == 0 and current.day in trigger_day_of_month and current.hour == trigger_hour and current.minute == trigger_minute: # schedule.append(self.to_utc(current, timezone_setting)) # elif field == "specific": # for dt_str in interval.get("specific", []): # try: # trigger_time = datetime.fromisoformat(dt_str) # print() # print(f"Thsi is TIME: {trigger_time}") # print() # if trigger_time.date() >= self.datebegin.date() and trigger_time.date() <= self.dateend.date(): # schedule.append(self.to_utc(trigger_time.replace(second=0, microsecond=0), timezone_setting)) # except ValueError as e: # print(e) # continue # current += timedelta(minutes=1) # iterate minute by minute # # print(current) # return sorted(set([dt.strftime("%Y-%m-%d %H:%M:%S") for dt in schedule])), sorted(set([dt.replace(hour=0, minute=trigger_minute, second=0, microsecond=0, tzinfo=self.datebegin.tzinfo) for dt in schedule])) def generate_schedule(self, datebegin: str, dateend: str, timezone_setting: int = 0): self.datebegin = datetime.strptime(datebegin, "%Y-%m-%d %H:%M:%S") self.dateend = datetime.strptime(dateend, "%Y-%m-%d %H:%M:%S") schedule = [] for interval in self.config: field = interval.get("field") trigger_hour = interval.get("triggerAtHour", 0) trigger_minute = interval.get("triggerAtMinute", 0) # --------------------------------------- # MINUTES # --------------------------------------- if field == "minutes": minutes_interval = interval.get("minutesInterval") if minutes_interval: current = self.datebegin.replace(second=0, microsecond=0) while current <= self.dateend: schedule.append(self.to_utc(current, timezone_setting)) current += timedelta(minutes=minutes_interval) # --------------------------------------- # HOURS # --------------------------------------- elif field == "hours": hours_interval = interval.get("hoursInterval") if hours_interval: current = self.datebegin.replace(minute=trigger_minute, second=0, microsecond=0) while current <= self.dateend: schedule.append(self.to_utc(current, timezone_setting)) current += timedelta(hours=hours_interval) # --------------------------------------- # DAYS # --------------------------------------- elif field == "days": days_interval = interval.get("daysInterval") if days_interval: current = self.datebegin.replace(hour=trigger_hour, minute=trigger_minute, second=0, microsecond=0) if current < self.datebegin: current += timedelta(days=1) while current <= self.dateend: schedule.append(self.to_utc(current, timezone_setting)) current += timedelta(days=days_interval) # --------------------------------------- # WEEKS # --------------------------------------- elif field == "weeks": weeks_interval = interval.get("weeksInterval") trigger_days = interval.get("triggerAtDay", []) if weeks_interval: current = self.datebegin.replace(hour=trigger_hour, minute=trigger_minute, second=0, microsecond=0) while current <= self.dateend: if current.weekday() in trigger_days: schedule.append(self.to_utc(current, timezone_setting)) current += timedelta(days=1) # --------------------------------------- # SPECIFIC TIMES # --------------------------------------- elif field == "specific": print('''field == "specific"''') for dt_str in interval.get("specific", []): try: trigger_time = datetime.fromisoformat(dt_str) print() print(f"GG: {trigger_time}") if self.datebegin <= trigger_time <= self.dateend: print(f"xx: {trigger_time}") schedule.append(self.to_utc(trigger_time, timezone_setting)) except ValueError: continue # Return distinct sorted list schedule = sorted(set(schedule)) return ( [dt.strftime("%Y-%m-%d %H:%M:%S") for dt in schedule], sorted({dt.replace(hour=0, minute=0, second=0, microsecond=0) for dt in schedule}) )