feat(#12): 支持在创建提醒时用 crontab 参数来指定 crontab 风格的循环模式。

nest/app/entity/plan.py

@@ -4,15 +4,30 @@
 from enum import Enum
 from typing import List, Optional, Set, Tuple, Union
 
+from croniter import croniter
+
 
 class IRepeater(ABC):
     @abstractmethod
     def compute_next_trigger_time(self) -> datetime:
         pass
 
 
+class CrontabRepeater(IRepeater):
+    """按照 crontab 风格来重复提醒的类。"""
+    def __init__(self, *, crontab: str = '', last_trigger_time: datetime, repeat_interval):
+        # TODO: 入参做成一整个 Plan 对象可能更合适。
+        self._crontab = crontab
+        self.last_trigger_time = last_trigger_time
+        self.repeat_interval = repeat_interval
+
+    def compute_next_trigger_time(self) -> timedelta:
+        _iter = croniter(self._crontab, self.last_trigger_time)
+        return _iter.get_next(datetime)
+
+
 class FixedIntervalRepeaterMixin(ABC):
-    def __init__(self, *, last_trigger_time: datetime, repeat_interval):
+    def __init__(self, *, last_trigger_time: datetime, repeat_interval, **_):
         self.last_trigger_time = last_trigger_time
         self.repeat_interval = repeat_interval
 
@@ -30,7 +45,7 @@ def get_interval(self) -> timedelta:
 
 
 class EndOfMonthRepeater(IRepeater):
-    def __init__(self, *, last_trigger_time: datetime, repeat_interval):
+    def __init__(self, *, last_trigger_time: datetime, repeat_interval, **_):
         self.last_trigger_time = last_trigger_time
         self.repeat_interval = repeat_interval
 
@@ -119,7 +134,11 @@ def get_interval(self) -> timedelta:
         return timedelta(days=7)
 
 
+REPEAT_TYPE_CRONTAB = 'crontab'
+
+
 _TYPE_TO_REPEATER_CLASS = {
+    REPEAT_TYPE_CRONTAB: CrontabRepeater,
     'daily': DailyRepeater,
     'end_of_month': EndOfMonthRepeater,
     'hourly': HourRepeater,
@@ -131,9 +150,10 @@ def get_interval(self) -> timedelta:
 
 class RepeaterFactory:
     @classmethod
-    def get_repeater(cls, *, last_trigger_time, repeat_interval, repeat_type):
+    def get_repeater(cls, *, crontab, last_trigger_time, repeat_interval, repeat_type):
         repeater_class = _TYPE_TO_REPEATER_CLASS[repeat_type]
         return repeater_class(
+            crontab=crontab,
             last_trigger_time=last_trigger_time,
             repeat_interval=repeat_interval,
         )
@@ -162,8 +182,12 @@ class PlanStatus(Enum):
 
 
 class Plan:
+    """
+    :ivar crontab: crontab 风格的循环模式。
+    """
     def __init__(self):
         self._duration: Optional[int] = None
+        self.crontab: str = ''
         self.id = None
         self.location_id = None
         self.repeat_interval: Union[None, timedelta] = None
@@ -184,11 +208,27 @@ def duration(self, value):
             raise InvalidDurationError()
         self._duration = value
 
+    def get_next_trigger_time(self):
+        """下一次触发提醒的时刻。"""
+        next_trigger_time: datetime = self.trigger_time
+        now = datetime.now()
+        while next_trigger_time.timestamp() < now.timestamp():
+            repeater = RepeaterFactory.get_repeater(
+                crontab=self.crontab,
+                last_trigger_time=next_trigger_time,
+                repeat_interval=self.repeat_interval,
+                repeat_type=self.repeat_type,
+            )
+            next_trigger_time = repeater.compute_next_trigger_time()
+
+        return next_trigger_time
+
     def get_repeating_description(self) -> str:
         """生成可读的、重复模式的描述。"""
         if self.repeat_type is None:
             return '不重复'
         simple_repeat_types = {
+            REPEAT_TYPE_CRONTAB: self.crontab,
             'daily': '每天',
             'end_of_month': '每月末',
             'hourly': '每小时',
@@ -224,6 +264,7 @@ def get_visible_wdays_description(self) -> str:
 
     @classmethod
     def new(cls, task_id, trigger_time, *,
+            crontab: str = '',
             duration: Union[None, int] = None,
             location_id: Union[None, int] = None,
             repeat_interval: Union[None, timedelta] = None,
@@ -234,6 +275,7 @@ def new(cls, task_id, trigger_time, *,
             raise RepeatIntervalMissingError()
 
         instance = Plan()
+        instance.crontab = crontab
         instance.duration = duration
         instance.location_id = location_id
         instance.repeat_interval = repeat_interval
@@ -274,24 +316,15 @@ def rebirth(self):
         """
         生成下一个触发时间的计划。
         """
-        next_trigger_time: datetime = self.trigger_time
-        now = datetime.now()
-        while next_trigger_time.timestamp() < now.timestamp():
-            repeater = RepeaterFactory.get_repeater(
-                last_trigger_time=next_trigger_time,
-                repeat_interval=self.repeat_interval,
-                repeat_type=self.repeat_type,
-            )
-            next_trigger_time = repeater.compute_next_trigger_time()
-
         instance = Plan()
+        instance.crontab = self.crontab
         instance.duration = self.duration
         instance.location_id = self.location_id
         instance.repeat_interval = self.repeat_interval
         instance.repeat_type = self.repeat_type
         instance.status = self.status
         instance.task_id = self.task_id
-        instance.trigger_time = next_trigger_time
+        instance.trigger_time = self.get_next_trigger_time()
         instance.visible_hours = self.visible_hours
         instance.visible_wdays = self.visible_wdays
         return instance

nest/app/use_case/create_plan.py

@@ -1,4 +1,5 @@
 # -*- coding: utf8 -*-
+import typing
 from abc import ABC, abstractmethod
 from datetime import datetime, timedelta
 from typing import Set, Union
@@ -7,12 +8,17 @@
 from nest.app.entity.plan import (
     InvalidRepeatTypeError,
     IPlanRepository,
-    Plan,
+    Plan, REPEAT_TYPE_CRONTAB,
 )
 from nest.app.entity.task import ITaskRepository
 
 
 class IParams(ABC):
+    @abstractmethod
+    def get_crontab(self) -> typing.Optional[str]:
+        """获取 crontab 风格的、提醒触发的控制字符串。"""
+        pass
+
     @abstractmethod
     def get_duration(self) -> Union[None, int]:
         pass
@@ -60,6 +66,7 @@ def __init__(self, *, location_repository, params, plan_repository,
 
     def run(self):
         params = self.params
+        crontab = params.get_crontab()
         duration = params.get_duration()
         location_id = params.get_location_id()
         if location_id is None:
@@ -71,6 +78,9 @@ def run(self):
             location_id = default_location.id
 
         repeat_type = params.get_repeat_type()
+        if crontab:
+            repeat_type = REPEAT_TYPE_CRONTAB
+
         if repeat_type and not Plan.is_valid_repeat_type(repeat_type):
             raise InvalidRepeatTypeError(repeat_type)
 
@@ -82,6 +92,7 @@ def run(self):
         plan = Plan.new(
             task_id,
             trigger_time,
+            crontab=crontab,
             duration=duration,
             location_id=location_id,
             repeat_interval=params.get_repeat_interval(),

nest/repository/DDL/20240707.sql

@@ -0,0 +1 @@
+ALTER TABLE `t_plan` ADD COLUMN `crontab` CHAR(16) DEFAULT '' COMMENT 'crontab 风格的执行周期' AFTER `id`;

nest/repository/plan.py

@@ -26,6 +26,7 @@ def add(self, plan: Plan):
                 repeat_interval = int(repeat_interval.total_seconds())
 
             insert_id = self.insert_to_db({
+                'crontab': plan.crontab,
                 'duration': plan.duration,
                 'location_id': plan.location_id,
                 'repeat_interval': repeat_interval,
@@ -48,6 +49,7 @@ def add(self, plan: Plan):
 
             query = Query\
                 .update(plan_table)\
+                .set(plan_table.crontab, plan.crontab)\
                 .set(plan_table.duration, plan.duration)\
                 .set(plan_table.location_id, plan.location_id)\
                 .set(plan_table.repeat_interval, repeat_interval)\
@@ -165,6 +167,7 @@ def remove(self, id_: int):
 
     def _row2entity(self, row: dict):
         plan = Plan()
+        plan.crontab = row['crontab']
         plan.duration = row['duration']
         plan.id = row['id']
         plan.location_id = row['location_id']

nest/web/controller/create_plan.py

@@ -5,10 +5,8 @@
 from flask import request
 from webargs import fields
 
-from nest.app.use_case.authenticate import AuthenticateUseCase
 from nest.app.use_case.create_plan import CreatePlanUseCase, InvalidRepeatTypeError, IParams
 from nest.web.authenticate import authenticate
-from nest.web.cookies_params import CookiesParams
 from nest.web.handle_response import wrap_response
 from nest.web.parser import parser
 from nest.web.presenter.plan import PlanPresenter
@@ -17,6 +15,7 @@
 class HTTPParams(IParams):
     def __init__(self):
         args = {
+            'crontab': fields.Str(default=''),
             'duration': fields.Int(allow_none=True),
             'location_id': fields.Int(allow_none=True),
             'repeat_interval': fields.TimeDelta(allow_none=True),
@@ -27,6 +26,7 @@ def __init__(self):
             'visible_wdays': fields.List(fields.Int, allow_none=True),
         }
         parsed_args = parser.parse(args, request)
+        self._crontab = parsed_args.get('crontab')
         self.duration = parsed_args.get('duration')
         self.location_id = parsed_args.get('location_id')
         self.repeat_interval = parsed_args.get('repeat_interval')
@@ -36,6 +36,9 @@ def __init__(self):
         self.visible_hours = set(parsed_args.get('visible_hours') or [])
         self.visible_wdays = set(parsed_args.get('visible_wdays') or [])
 
+    def get_crontab(self):
+        return self._crontab
+
     def get_duration(self) -> Union[None, int]:
         return self.duration
 

requirements.txt

@@ -2,7 +2,9 @@ attrs==20.3.0
 autopep8==1.5.5
 build==0.3.0
 click==7.1.2
+croniter==2.0.5
 DBUtils==2.0.1
+exceptiongroup==1.2.1
 Flask==1.1.2
 gittyleaks==0.0.31
 honcho==1.0.1
@@ -20,11 +22,15 @@ pymysql-pooling==1.0.6
 pyparsing==2.4.7
 PyPika==0.47.7
 pytest==7.4.3
+python-dateutil==2.9.0.post0
+pytz==2024.1
 redis==3.5.3
 rope==0.18.0
 ruff==0.0.271
 scandir==1.10.0
 sh==1.14.1
+six==1.16.0
 toml==0.10.2
+tomli==2.0.1
 webargs==7.0.1
 Werkzeug==1.0.1

tests/use_case/plan/test_create.py

@@ -1,4 +1,5 @@
 # -*- coding: utf8 -*-
+import typing
 from datetime import timedelta
 from typing import List, Set, Union
 import unittest
@@ -31,6 +32,9 @@ class MockParams(IParams):
     def __init__(self, *, duration: int):
         self.duration = duration
 
+    def get_crontab(self) -> typing.Optional[str]:
+        pass
+
     def get_duration(self) -> Union[None, int]:
         return self.duration
 

tests/web/controller/test_change_plan.py

@@ -1,4 +1,5 @@
 # -*- coding: utf8 -*-
+import typing
 from datetime import datetime, timedelta
 import unittest
 from typing import List, Union, Set
@@ -15,6 +16,9 @@ class CreatePlanParams(create_plan.IParams):
     def __init__(self, task_id):
         self._task_id = task_id
 
+    def get_crontab(self) -> typing.Optional[str]:
+        pass
+
     def get_duration(self) -> Union[None, int]:
         return None