Assistants.scheduled

Обёртка для периодического запуска ассистентов.

Запускает переданного ассистента с заданным интервалом в отдельном потоке с помощью threading.Timer.

View Source

 1"""
 2Обёртка для периодического запуска ассистентов.
 3
 4Запускает переданного ассистента с заданным интервалом в отдельном
 5потоке с помощью threading.Timer.
 6"""
 7
 8import threading
 9
10from loguru import logger
11
12from Assistants import Assistant
13from Utilities.time_utils import format_time
14
15
16class ScheduledAssistant(Assistant):
17    """Запускает ассистента периодически с заданным интервалом.
18
19    Использует threading.Timer для планирования повторных запусков.
20    Поддерживает корректный останов через метод stop().
21
22    Attributes:
23        assistant: Оборачиваемый ассистент.
24        interval: Интервал между запусками в секундах.
25        timer: Таймер для следующего запуска.
26        is_running: Флаг состояния планировщика.
27    """
28
29    def __init__(self, assistant: Assistant, interval: int):
30        """Инициализирует планировщик для ассистента.
31
32        Args:
33            assistant: Ассистент, который будет запускаться периодически.
34            interval: Интервал между запусками в секундах.
35        """
36        super().__init__(assistant.name)
37
38        self.assistant = assistant
39        self.interval = interval
40        self.timer: threading.Timer | None = None
41        self.is_running: bool = False
42        self._lock: threading.Lock = threading.Lock()
43
44    def run(self):
45        """Выполняет один цикл работы ассистента и планирует следующий.
46
47        При возникновении исключения логирует ошибку, но продолжает
48        планировать следующие запуски, пока планировщик активен.
49        """
50        try:
51            self.assistant.run()
52        except Exception as e:
53            logger.error(f"Ошибка в контроллере: {e}")
54        finally:
55            # Планируем следующий запуск
56            if self.is_running:
57                self.timer = threading.Timer(self.interval, self.run)
58                self.timer.start()
59
60    def start(self) -> None:
61        """Запускает планировщик.
62
63        Если планировщик уже запущен, повторный запуск игнорируется.
64        """
65        with self._lock:
66            if self.is_running:
67                logger.warning(f'Планировщик для "{self.assistant.name}" уже запущен')
68                return
69            self.is_running = True
70        logger.info(
71            f'Запуск планировщика для "{self.assistant.name}" с интервалом {format_time(self.interval)}'
72        )
73        self.run()
74
75    def stop(self) -> None:
76        """Останавливает планировщик.
77
78        Отменяет ожидающий таймер и устанавливает флаг остановки.
79        """
80        with self._lock:
81            self.is_running = False
82            if self.timer:
83                self.timer.cancel()
84                self.timer = None
85        logger.info(f'Планировщик остановлен для "{self.assistant.name}"')

class ScheduledAssistant(Assistants.assistant.Assistant): View Source

17class ScheduledAssistant(Assistant):
18    """Запускает ассистента периодически с заданным интервалом.
19
20    Использует threading.Timer для планирования повторных запусков.
21    Поддерживает корректный останов через метод stop().
22
23    Attributes:
24        assistant: Оборачиваемый ассистент.
25        interval: Интервал между запусками в секундах.
26        timer: Таймер для следующего запуска.
27        is_running: Флаг состояния планировщика.
28    """
29
30    def __init__(self, assistant: Assistant, interval: int):
31        """Инициализирует планировщик для ассистента.
32
33        Args:
34            assistant: Ассистент, который будет запускаться периодически.
35            interval: Интервал между запусками в секундах.
36        """
37        super().__init__(assistant.name)
38
39        self.assistant = assistant
40        self.interval = interval
41        self.timer: threading.Timer | None = None
42        self.is_running: bool = False
43        self._lock: threading.Lock = threading.Lock()
44
45    def run(self):
46        """Выполняет один цикл работы ассистента и планирует следующий.
47
48        При возникновении исключения логирует ошибку, но продолжает
49        планировать следующие запуски, пока планировщик активен.
50        """
51        try:
52            self.assistant.run()
53        except Exception as e:
54            logger.error(f"Ошибка в контроллере: {e}")
55        finally:
56            # Планируем следующий запуск
57            if self.is_running:
58                self.timer = threading.Timer(self.interval, self.run)
59                self.timer.start()
60
61    def start(self) -> None:
62        """Запускает планировщик.
63
64        Если планировщик уже запущен, повторный запуск игнорируется.
65        """
66        with self._lock:
67            if self.is_running:
68                logger.warning(f'Планировщик для "{self.assistant.name}" уже запущен')
69                return
70            self.is_running = True
71        logger.info(
72            f'Запуск планировщика для "{self.assistant.name}" с интервалом {format_time(self.interval)}'
73        )
74        self.run()
75
76    def stop(self) -> None:
77        """Останавливает планировщик.
78
79        Отменяет ожидающий таймер и устанавливает флаг остановки.
80        """
81        with self._lock:
82            self.is_running = False
83            if self.timer:
84                self.timer.cancel()
85                self.timer = None
86        logger.info(f'Планировщик остановлен для "{self.assistant.name}"')

Запускает ассистента периодически с заданным интервалом.

Использует threading.Timer для планирования повторных запусков. Поддерживает корректный останов через метод stop().

Attributes: assistant: Оборачиваемый ассистент. interval: Интервал между запусками в секундах. timer: Таймер для следующего запуска. is_running: Флаг состояния планировщика.

ScheduledAssistant(assistant: Assistants.assistant.Assistant, interval: int) View Source

30    def __init__(self, assistant: Assistant, interval: int):
31        """Инициализирует планировщик для ассистента.
32
33        Args:
34            assistant: Ассистент, который будет запускаться периодически.
35            interval: Интервал между запусками в секундах.
36        """
37        super().__init__(assistant.name)
38
39        self.assistant = assistant
40        self.interval = interval
41        self.timer: threading.Timer | None = None
42        self.is_running: bool = False
43        self._lock: threading.Lock = threading.Lock()

Инициализирует планировщик для ассистента.

Args: assistant: Ассистент, который будет запускаться периодически. interval: Интервал между запусками в секундах.

assistant

interval

timer: threading.Timer | None

is_running: bool

def run(self): View Source

45    def run(self):
46        """Выполняет один цикл работы ассистента и планирует следующий.
47
48        При возникновении исключения логирует ошибку, но продолжает
49        планировать следующие запуски, пока планировщик активен.
50        """
51        try:
52            self.assistant.run()
53        except Exception as e:
54            logger.error(f"Ошибка в контроллере: {e}")
55        finally:
56            # Планируем следующий запуск
57            if self.is_running:
58                self.timer = threading.Timer(self.interval, self.run)
59                self.timer.start()

Выполняет один цикл работы ассистента и планирует следующий.

При возникновении исключения логирует ошибку, но продолжает планировать следующие запуски, пока планировщик активен.

def start(self) -> None: View Source

61    def start(self) -> None:
62        """Запускает планировщик.
63
64        Если планировщик уже запущен, повторный запуск игнорируется.
65        """
66        with self._lock:
67            if self.is_running:
68                logger.warning(f'Планировщик для "{self.assistant.name}" уже запущен')
69                return
70            self.is_running = True
71        logger.info(
72            f'Запуск планировщика для "{self.assistant.name}" с интервалом {format_time(self.interval)}'
73        )
74        self.run()

Запускает планировщик.

Если планировщик уже запущен, повторный запуск игнорируется.

def stop(self) -> None: View Source

76    def stop(self) -> None:
77        """Останавливает планировщик.
78
79        Отменяет ожидающий таймер и устанавливает флаг остановки.
80        """
81        with self._lock:
82            self.is_running = False
83            if self.timer:
84                self.timer.cancel()
85                self.timer = None
86        logger.info(f'Планировщик остановлен для "{self.assistant.name}"')

Останавливает планировщик.

Отменяет ожидающий таймер и устанавливает флаг остановки.

Inherited Members

Assistants.assistant.Assistant: name