otm

Ассистент переноса просроченных задач (Old Task Mover).

Запускается по расписанию APScheduler и переносит все невыполненные задачи с прошедших дат на текущий день.

 1"""
 2Ассистент переноса просроченных задач (Old Task Mover).
 3
 4Запускается по расписанию APScheduler и переносит все невыполненные
 5задачи с прошедших дат на текущий день.
 6"""
 7
 8from langfuse import observe
 9from loguru import logger
10
11from Assistants.assistant import Assistant
12from taskman import TaskManager
13
14
15class OTMAssistant(Assistant):
16    """
17    Ассистент для автоматического переноса просроченных задач.
18
19    При запуске перемещает все невыполненные задачи, дата которых
20    раньше сегодняшнего дня, на текущую дату.
21    """
22
23    def __init__(self, name, task_manager: TaskManager):
24        """
25        Инициализирует ассистента переноса задач.
26
27        Args:
28            name: Название ассистента
29            task_manager: Экземпляр TaskManager для работы с задачами
30        """
31        super().__init__(self)
32        self.name = name
33        self.task_manager = task_manager
34
35    @observe(name="Daily task migration")
36    def run(self):
37        """
38        Переносит все просроченные невыполненные задачи на сегодня.
39
40        Запускается ежедневно по расписанию APScheduler.
41        В случае ошибки логирует её и завершает выполнение.
42        """
43        try:
44            logger.info(
45                f'Запуск цикла переноса прошедших невыполненных задач на сегодня для "{self.name}"...'
46            )
47            self.task_manager.move_incomplete_tasks_to_today()
48            logger.info(
49                f'Цикл переноса прошедших невыполненных задач на сегодня завершен для "{self.name}".'
50            )
51        except Exception as e:
52            logger.error(f'Ошибка в OTM Assistant "{self.name}": {e}')
class OTMAssistant(Assistants.assistant.Assistant): 
16class OTMAssistant(Assistant):
17    """
18    Ассистент для автоматического переноса просроченных задач.
19
20    При запуске перемещает все невыполненные задачи, дата которых
21    раньше сегодняшнего дня, на текущую дату.
22    """
23
24    def __init__(self, name, task_manager: TaskManager):
25        """
26        Инициализирует ассистента переноса задач.
27
28        Args:
29            name: Название ассистента
30            task_manager: Экземпляр TaskManager для работы с задачами
31        """
32        super().__init__(self)
33        self.name = name
34        self.task_manager = task_manager
35
36    @observe(name="Daily task migration")
37    def run(self):
38        """
39        Переносит все просроченные невыполненные задачи на сегодня.
40
41        Запускается ежедневно по расписанию APScheduler.
42        В случае ошибки логирует её и завершает выполнение.
43        """
44        try:
45            logger.info(
46                f'Запуск цикла переноса прошедших невыполненных задач на сегодня для "{self.name}"...'
47            )
48            self.task_manager.move_incomplete_tasks_to_today()
49            logger.info(
50                f'Цикл переноса прошедших невыполненных задач на сегодня завершен для "{self.name}".'
51            )
52        except Exception as e:
53            logger.error(f'Ошибка в OTM Assistant "{self.name}": {e}')

Ассистент для автоматического переноса просроченных задач.

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

OTMAssistant(name, task_manager: taskman.TaskManager) 
24    def __init__(self, name, task_manager: TaskManager):
25        """
26        Инициализирует ассистента переноса задач.
27
28        Args:
29            name: Название ассистента
30            task_manager: Экземпляр TaskManager для работы с задачами
31        """
32        super().__init__(self)
33        self.name = name
34        self.task_manager = task_manager

Инициализирует ассистента переноса задач.

Arguments:
  • name: Название ассистента
  • task_manager: Экземпляр TaskManager для работы с задачами
name
task_manager
@observe(name='Daily task migration')
def run(self): 
36    @observe(name="Daily task migration")
37    def run(self):
38        """
39        Переносит все просроченные невыполненные задачи на сегодня.
40
41        Запускается ежедневно по расписанию APScheduler.
42        В случае ошибки логирует её и завершает выполнение.
43        """
44        try:
45            logger.info(
46                f'Запуск цикла переноса прошедших невыполненных задач на сегодня для "{self.name}"...'
47            )
48            self.task_manager.move_incomplete_tasks_to_today()
49            logger.info(
50                f'Цикл переноса прошедших невыполненных задач на сегодня завершен для "{self.name}".'
51            )
52        except Exception as e:
53            logger.error(f'Ошибка в OTM Assistant "{self.name}": {e}')

Переносит все просроченные невыполненные задачи на сегодня.

Запускается ежедневно по расписанию APScheduler. В случае ошибки логирует её и завершает выполнение.