Simplify task crontab expression validation (#733)

backend/app/task/service/scheduler_service.py

@@ -65,14 +65,7 @@ class TaskSchedulerService:
             if task_scheduler:
                 raise errors.ConflictError(msg='任务调度已存在')
             if obj.type == TaskSchedulerType.CRONTAB:
-                crontab_split = obj.crontab.split(' ')
-                if len(crontab_split) != 5:
-                    raise errors.RequestError(msg='Crontab 表达式非法')
-                crontab_verify('m', crontab_split[0])
-                crontab_verify('h', crontab_split[1])
-                crontab_verify('dow', crontab_split[2])
-                crontab_verify('dom', crontab_split[3])
-                crontab_verify('moy', crontab_split[4])
+                crontab_verify(obj.crontab)
             await task_scheduler_dao.create(db, obj)
 
     @staticmethod
@@ -92,14 +85,7 @@ class TaskSchedulerService:
                 if await task_scheduler_dao.get_by_name(db, obj.name):
                     raise errors.ConflictError(msg='任务调度已存在')
             if task_scheduler.type == TaskSchedulerType.CRONTAB:
-                crontab_split = obj.crontab.split(' ')
-                if len(crontab_split) != 5:
-                    raise errors.RequestError(msg='Crontab 表达式非法')
-                crontab_verify('m', crontab_split[0])
-                crontab_verify('h', crontab_split[1])
-                crontab_verify('dow', crontab_split[2])
-                crontab_verify('dom', crontab_split[3])
-                crontab_verify('moy', crontab_split[4])
+                crontab_verify(obj.crontab)
             count = await task_scheduler_dao.update(db, pk, obj)
             return count
 

backend/app/task/utils/schedulers.py

@@ -188,21 +188,12 @@ class ModelEntry(ScheduleEntry):
                 if not obj:
                     obj = TaskScheduler(**CreateTaskSchedulerParam(task=task, **spec).model_dump())
             elif isinstance(schedule, schedules.crontab):
-                crontab_minute = schedule._orig_minute if crontab_verify('m', schedule._orig_minute, False) else '*'
-                crontab_hour = schedule._orig_hour if crontab_verify('h', schedule._orig_hour, False) else '*'
-                crontab_day_of_week = (
-                    schedule._orig_day_of_week if crontab_verify('dom', schedule._orig_day_of_week, False) else '*'
-                )
-                crontab_day_of_month = (
-                    schedule._orig_day_of_month if crontab_verify('dom', schedule._orig_day_of_month, False) else '*'
-                )
-                crontab_month_of_year = (
-                    schedule._orig_month_of_year if crontab_verify('moy', schedule._orig_month_of_year, False) else '*'
-                )
+                crontab = f'{schedule._orig_minute} {schedule._orig_hour} {schedule._orig_day_of_week} {schedule._orig_day_of_month} {schedule._orig_month_of_year}'  # noqa: E501
+                crontab_verify(crontab)
                 spec = {
                     'name': name,
                     'type': TaskSchedulerType.CRONTAB.value,
-                    'crontab': f'{crontab_minute} {crontab_hour} {crontab_day_of_week} {crontab_day_of_month} {crontab_month_of_year}',  # noqa: E501
+                    'crontab': crontab,
                 }
                 stmt = select(TaskScheduler).filter_by(**spec)
                 query = await db.execute(stmt)

backend/app/task/utils/tzcrontab.py

@@ -1,7 +1,6 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 from datetime import datetime
-from typing import Literal
 
 from celery import schedules
 from celery.schedules import ParseException, crontab_parser
@@ -53,34 +52,22 @@ class TzAwareCrontab(schedules.crontab):
         )
 
 
-def crontab_verify(filed: Literal['m', 'h', 'dow', 'dom', 'moy'], value: str, raise_exc: bool = True) -> bool:
+def crontab_verify(crontab: str) -> None:
     """
     验证 Celery crontab 表达式
 
-    :param filed: 验证的字段
-    :param value: 验证的值
-    :param raise_exc: 是否抛出异常
+    :param crontab: 计划表达式
     :return:
     """
-    valid = True
+    crontab_split = crontab.split(' ')
+    if len(crontab_split) != 5:
+        raise errors.RequestError(msg='Crontab 表达式非法')
 
     try:
-        match filed:
-            case 'm':
-                crontab_parser(60, 0).parse(value)
-            case 'h':
-                crontab_parser(24, 0).parse(value)
-            case 'dow':
-                crontab_parser(7, 0).parse(value)
-            case 'dom':
-                crontab_parser(31, 1).parse(value)
-            case 'moy':
-                crontab_parser(12, 1).parse(value)
-            case _:
-                raise errors.ServerError(msg=f'无效字段：{filed}')
+        crontab_parser(60, 0).parse(crontab_split[0])  # minute
+        crontab_parser(24, 0).parse(crontab_split[1])  # hour
+        crontab_parser(7, 0).parse(crontab_split[2])  # day_of_week
+        crontab_parser(31, 1).parse(crontab_split[3])  # day_of_month
+        crontab_parser(12, 1).parse(crontab_split[4])  # month_of_year
     except ParseException:
-        valid = False
-        if raise_exc:
-            raise errors.RequestError(msg=f'crontab 值 {value} 非法')
-
-    return valid
+        raise errors.RequestError(msg='Crontab 表达式非法')