1class Task(ABC):
2    @abstractmethod
3    def execute(self):
4        pass
5
6class PrintMessageTask(Task):
7    def __init__(self, message: str):
8        self.message = message
9
10    def execute(self):
11        print(f"[{datetime.now().strftime('%H:%M:%S.%f')[:-3]}] Executing PrintMessageTask: {self.message}")
12
13class DataBackupTask(Task):
14    def __init__(self, source: str, destination: str):
15        self.source = source
16        self.destination = destination
17
18    def execute(self):
19        print(f"[{datetime.now().strftime('%H:%M:%S.%f')[:-3]}] Executing DataBackupTask: Backing up from {self.source} to {self.destination}...")
20        print(f"[{datetime.now().strftime('%H:%M:%S.%f')[:-3]}] DataBackupTask: Backup complete.")

1class ScheduledTask:
2    def __init__(self, task: Task, strategy: 'SchedulingStrategy'):
3        self.id = str(uuid.uuid4())
4        self.task = task
5        self.strategy = strategy
6        self.next_execution_time = None
7        self.last_execution_time = None
8        self.update_next_execution_time()
9
10    def update_next_execution_time(self):
11        next_time = self.strategy.get_next_execution_time(self.last_execution_time)
12        self.next_execution_time = next_time
13
14    def update_last_execution_time(self):
15        self.last_execution_time = self.next_execution_time
16
17    def __lt__(self, other):
18        return self.next_execution_time < other.next_execution_time
19
20    def has_more_executions(self):
21        return self.next_execution_time is not None
22
23    # Getters

1class SchedulingStrategy(ABC):
2    @abstractmethod
3    def get_next_execution_time(self, last_execution_time: Optional[datetime]) -> Optional[datetime]:
4        pass
5
6class OneTimeSchedulingStrategy(SchedulingStrategy):
7    def __init__(self, execution_time: datetime):
8        self.execution_time = execution_time
9
10    def get_next_execution_time(self, last_execution_time: Optional[datetime]) -> Optional[datetime]:
11        return self.execution_time if last_execution_time is None else None
12
13class RecurringSchedulingStrategy(SchedulingStrategy):
14    def __init__(self, interval: timedelta):
15        self.interval = interval
16
17    def get_next_execution_time(self, last_execution_time: Optional[datetime]) -> Optional[datetime]:
18        base_time = datetime.now() if last_execution_time is None else last_execution_time
19        return base_time + self.interval

1class TaskExecutionObserver(ABC):
2    @abstractmethod
3    def on_task_started(self, task: ScheduledTask):
4        pass
5
6    @abstractmethod
7    def on_task_completed(self, task: ScheduledTask):
8        pass
9
10    @abstractmethod
11    def on_task_failed(self, task: ScheduledTask, exception: Exception):
12        pass
13        
14class LoggingObserver(TaskExecutionObserver):
15    def on_task_started(self, task: ScheduledTask):
16        print(f"[LOG - {datetime.now().strftime('%H:%M:%S.%f')[:-3]}] [{threading.current_thread().name}] Task {task.get_id()} started.")
17
18    def on_task_completed(self, task: ScheduledTask):
19        print(f"[LOG - {datetime.now().strftime('%H:%M:%S.%f')[:-3]}] [{threading.current_thread().name}] Task {task.get_id()} completed successfully.")
20
21    def on_task_failed(self, task: ScheduledTask, exception: Exception):
22        print(f"[LOG - {datetime.now().strftime('%H:%M:%S.%f')[:-3]}] [{threading.current_thread().name}] Task {task.get_id()} failed: {str(exception)}")
23

1class TaskSchedulerService:
2    _instance = None
3    _lock = threading.Lock()
4
5    def __new__(cls):
6        if cls._instance is None:
7            with cls._lock:
8                if cls._instance is None:
9                    cls._instance = super(TaskSchedulerService, cls).__new__(cls)
10                    cls._instance._initialized = False
11        return cls._instance
12
13    def __init__(self):
14        if not hasattr(self, '_initialized') or not self._initialized:
15            self.task_queue = queue.PriorityQueue()
16            self.observers = []
17            self.workers = []
18            self.running = True
19            self._initialized = True
20
21    @staticmethod
22    def get_instance():
23        return TaskSchedulerService()
24
25    def initialize(self, worker_count: int):
26        if worker_count <= 0:
27            raise ValueError("Worker count must be >= 1")
28        self.workers = []
29        self.start_workers(worker_count)
30
31    def schedule(self, task: Task, strategy: SchedulingStrategy) -> str:
32        scheduled_task = ScheduledTask(task, strategy)
33        self.task_queue.put(scheduled_task)
34        return scheduled_task.get_id()
35
36    def start_workers(self, worker_count: int):
37        for i in range(worker_count):
38            worker = threading.Thread(target=self.run_worker, name=f"WorkerThread-{i}", daemon=True)
39            self.workers.append(worker)
40            worker.start()
41
42    def run_worker(self):
43        while self.running:
44            try:
45                task = self.task_queue.get(timeout=1)
46                now = datetime.now()
47                wait_time = 0
48
49                if task.get_next_execution_time() > now:
50                    wait_time = (task.get_next_execution_time() - now).total_seconds()
51
52                if wait_time > 0:
53                    time.sleep(wait_time)
54
55                # Check if a higher-priority task has arrived while we were sleeping
56                try:
57                    head = self.task_queue.get_nowait()
58                    if head < task:
59                        self.task_queue.put(task)
60                        self.task_queue.put(head)
61                        continue
62                    else:
63                        self.task_queue.put(head)
64                except queue.Empty:
65                    pass
66
67                self.execute(task)
68                self.task_queue.task_done()
69            except queue.Empty:
70                continue
71            except Exception as e:
72                print(f"Worker thread error: {e}")
73                break
74
75        print(f"{threading.current_thread().name} stopped.")
76
77    def execute(self, task: ScheduledTask):
78        for observer in self.observers:
79            observer.on_task_started(task)
80        
81        try:
82            task.get_task().execute()
83            task.update_last_execution_time()
84            for observer in self.observers:
85                observer.on_task_completed(task)
86        except Exception as e:
87            for observer in self.observers:
88                observer.on_task_failed(task, e)
89            print(f"Task {task.get_id()} failed with error: {str(e)}")
90        finally:
91            task.update_next_execution_time()
92            
93            if task.has_more_executions():
94                self.task_queue.put(task)
95            else:
96                print(f"Task {task.get_id()} has no more executions and will not be rescheduled.")
97
98    def shutdown(self):
99        self.running = False
100        for worker in self.workers:
101            worker.join(timeout=1)
102        print("Scheduler shut down.")
103
104    def add_observer(self, observer: TaskExecutionObserver):
105        self.observers.append(observer)

1def main():
2    # 1. Setup the facade and observers
3    scheduler = TaskSchedulerService.get_instance()
4    scheduler.add_observer(LoggingObserver())
5
6    # 2. Initialize the scheduler
7    scheduler.initialize(10)
8
9    # 3. Define tasks and strategies
10    # Scenario 1: One-time task, 1 second from now
11    one_time_task = PrintMessageTask("This is a one-time task.")
12    one_time_strategy = OneTimeSchedulingStrategy(datetime.now() + timedelta(seconds=1))
13
14    # Scenario 2: Recurring task, every 2 seconds
15    recurring_task = PrintMessageTask("This is a recurring task.")
16    recurring_strategy = RecurringSchedulingStrategy(timedelta(seconds=2))
17
18    # Scenario 3: A long-running backup task, scheduled to run in 3 seconds
19    backup_task = DataBackupTask("/data/source", "/data/backup")
20    long_running_recurring_strategy = OneTimeSchedulingStrategy(datetime.now() + timedelta(seconds=3))
21
22    # 4. Schedule the tasks using the facade
23    print("Scheduling tasks...")
24    scheduler.schedule(one_time_task, one_time_strategy)
25    scheduler.schedule(recurring_task, recurring_strategy)
26    scheduler.schedule(backup_task, long_running_recurring_strategy)
27
28    # 5. Let the demo run for a while
29    print("Scheduler is running. Waiting for tasks to execute... (Demo will run for 6 seconds)")
30    time.sleep(6)
31
32    # 6. Shutdown the scheduler
33    scheduler.shutdown()
34
35if __name__ == "__main__":
36    main()