feat: implement scheduling functionality in ScheduledTrigger and enhance trigger deployment logic

Author: liuruibin

apps/trigger/handler/impl/trigger/scheduled_trigger.py

@@ -1,12 +1,200 @@
 # coding=utf-8
-"""
-    @project: MaxKB
-    @Author：虎虎
-    @file： scheduled_trigger.py
-    @date：2026/1/14 18:57
-    @desc:
-"""
+from __future__ import annotations
+
+import random
+
+from django.db.models import QuerySet
+
+from common.utils.lock import RedisLock
+from common.utils.logger import maxkb_logger
+from ops import celery_app
 from trigger.handler.base_trigger import BaseTrigger
+from trigger.models import TriggerTask
+
+
+def _parse_hhmm(value: str) -> tuple[int, int]:
+    hour_str, minute_str = (value or "").split(":")
+    hour = int(hour_str)
+    minute = int(minute_str)
+    if not (0 <= hour <= 23 and 0 <= minute <= 59):
+        raise ValueError("hour/minute out of range")
+    return hour, minute
+
+
+def _weekday_to_cron(d: int | str) -> str:
+    mapping = {1: "mon", 2: "tue", 3: "wed", 4: "thu", 5: "fri", 6: "sat", 7: "sun", 0: "sun"}
+    di = int(d)
+    if di not in mapping:
+        raise ValueError("invalid weekday")
+    return mapping[di]
+
+
+def _get_active_trigger_tasks(trigger_id: str) -> list[dict]:
+    return list(
+        QuerySet(TriggerTask)
+        .filter(trigger_id=trigger_id, is_active=True)
+        .values("id", "source_type", "source_id", "parameter")
+    )
+
+
+def _deploy_daily(trigger: dict, trigger_tasks: list[dict], setting: dict, trigger_id: str, func) -> None:
+    from common.job import scheduler
+
+    times = setting.get("time") or []
+    for t in times:
+        try:
+            hour, minute = _parse_hhmm(t)
+        except Exception:
+            maxkb_logger.warning(f"invalid time={t}, trigger_id={trigger_id}")
+            continue
+
+        for task in trigger_tasks:
+            job_id = f"trigger:{trigger_id}:task:{task['id']}:daily:{hour:02d}{minute:02d}"
+            scheduler.add_job(
+                func,
+                trigger="cron",
+                hour=str(hour),
+                minute=str(minute),
+                id=job_id,
+                kwargs={"trigger": trigger, "trigger_tasks": trigger_tasks},
+                replace_existing=True,
+            )
+
+
+def _deploy_weekly(trigger: dict, trigger_tasks: list[dict], setting: dict, trigger_id: str, func) -> None:
+    from common.job import scheduler
+
+    times = setting.get("time") or []
+    days = setting.get("days") or []
+    if not times or not days:
+        maxkb_logger.warning(f"empty weekly setting, trigger_id={trigger_id}")
+        return
+
+    for d in days:
+        try:
+            dow = _weekday_to_cron(d)
+        except Exception:
+            maxkb_logger.warning(f"invalid weekday={d}, trigger_id={trigger_id}")
+            continue
+
+        for t in times:
+            try:
+                hour, minute = _parse_hhmm(t)
+            except Exception:
+                maxkb_logger.warning(f"invalid time={t}, trigger_id={trigger_id}")
+                continue
+
+            for task in trigger_tasks:
+                job_id = f"trigger:{trigger_id}:task:{task['id']}:weekly:{dow}:{hour:02d}{minute:02d}"
+                scheduler.add_job(
+                    func,
+                    trigger="cron",
+                    day_of_week=dow,
+                    hour=str(hour),
+                    minute=str(minute),
+                    id=job_id,
+                    kwargs={"trigger": trigger, "trigger_tasks": trigger_tasks},
+                    replace_existing=True,
+                )
+
+
+def _deploy_monthly(trigger: dict, trigger_tasks: list[dict], setting: dict, trigger_id: str, func) -> None:
+    from common.job import scheduler
+
+    times = setting.get("time") or []
+    days = setting.get("days") or []
+    if not times or not days:
+        maxkb_logger.warning(f"empty monthly setting, trigger_id={trigger_id}")
+        return
+
+    for d in days:
+        try:
+            dom = int(d)
+            if not (1 <= dom <= 31):
+                raise ValueError("invalid day of month")
+        except Exception:
+            maxkb_logger.warning(f"invalid day={d}, trigger_id={trigger_id}")
+            continue
+
+        for t in times:
+            try:
+                hour, minute = _parse_hhmm(t)
+            except Exception:
+                maxkb_logger.warning(f"invalid time={t}, trigger_id={trigger_id}")
+                continue
+
+            for task in trigger_tasks:
+                job_id = f"trigger:{trigger_id}:task:{task['id']}:monthly:{dom:02d}:{hour:02d}{minute:02d}"
+                scheduler.add_job(
+                    func,
+                    trigger="cron",
+                    day=str(dom),
+                    hour=str(hour),
+                    minute=str(minute),
+                    id=job_id,
+                    kwargs={"trigger": trigger, "trigger_tasks": trigger_tasks},
+                    replace_existing=True,
+                )
+
+
+def _deploy_interval(trigger: dict, trigger_tasks: list[dict], setting: dict, trigger_id: str, func) -> None:
+    from common.job import scheduler
+
+    unit = (setting.get("interval_unit") or "").strip()
+    value = setting.get("interval_value")
+
+    try:
+        value_i = int(value)
+        if value_i <= 0:
+            raise ValueError("interval_value must be positive")
+    except Exception:
+        maxkb_logger.warning(f"invalid interval_value={value}, trigger_id={trigger_id}")
+        return
+
+    if unit not in {"seconds", "minutes", "hours", "days"}:
+        maxkb_logger.warning(f"invalid interval_unit={unit}, trigger_id={trigger_id}")
+        return
+
+    for task in trigger_tasks:
+        job_id = f"trigger:{trigger_id}:task:{task['id']}:interval:{unit}:{value_i}"
+        scheduler.add_job(
+            func,
+            trigger="interval",
+            id=job_id,
+            kwargs={"trigger": trigger, "trigger_tasks": trigger_tasks},
+            replace_existing=True,
+            **{unit: value_i},
+        )
+
+
+def _remove_trigger_jobs(trigger_id: str) -> None:
+    from common.job import scheduler
+
+    prefix = f"trigger:{trigger_id}:"
+    for job in scheduler.get_jobs():
+        if getattr(job, "id", "").startswith(prefix):
+            try:
+                job.remove()
+            except Exception as e:
+                maxkb_logger.warning(f"remove job failed, job_id={job.id}, err={e}")
+
+
+@celery_app.task(name='celery:deploy_scheduled_trigger')
+def deploy_scheduled_trigger(trigger: dict, trigger_tasks: list[dict], setting: dict, schedule_type: str, func) -> None:
+    _remove_trigger_jobs(trigger['id'])
+
+    deployers = {
+        "daily": _deploy_daily,
+        "weekly": _deploy_weekly,
+        "monthly": _deploy_monthly,
+        "interval": _deploy_interval,
+    }
+    fn = deployers.get(schedule_type)
+    if not fn:
+        maxkb_logger.warning(f"unsupported schedule_type={schedule_type}, trigger_id={trigger['id']}")
+        return
+
+    fn(trigger, trigger_tasks, setting, trigger['id'], func)
 
 
 class ScheduledTrigger(BaseTrigger):
@@ -16,13 +204,49 @@ class ScheduledTrigger(BaseTrigger):
 
     @staticmethod
     def execute(trigger, **kwargs):
-        pass
+        n = random.randint(1, 1_000_000_000)
+
+        maxkb_logger.info(f"scheduled trigger execute, trigger={n}")
 
     def support(self, trigger, **kwargs):
-        return trigger.get('trigger_type') == 'SCHEDULED'
+        return trigger.get("trigger_type") == "SCHEDULED"
 
     def deploy(self, trigger, **kwargs):
-        pass
+        trigger_id = str(trigger["id"])
+        setting = trigger.get("trigger_setting") or {}
+        schedule_type = setting.get("schedule_type")
+
+        if not trigger.get("is_active", True):
+            self.undeploy(trigger, **kwargs)
+            return
+
+        trigger_tasks = _get_active_trigger_tasks(trigger["id"])
+        if not trigger_tasks:
+            maxkb_logger.warning(f"no active trigger_tasks, trigger_id={trigger_id}")
+            self.undeploy(trigger, **kwargs)
+            return
+
+        rlock = RedisLock()
+        lock_key = f"scheduled_trigger_deploy:{trigger_id}"
+        if not rlock.try_lock(lock_key, 30):
+            return
+
+        try:
+            maxkb_logger.debug(f"get lock {lock_key}")
+            deploy_scheduled_trigger.delay(trigger, trigger_tasks, setting, schedule_type, self.execute)
+
+        finally:
+            rlock.un_lock(lock_key)
 
     def undeploy(self, trigger, **kwargs):
-        pass
+        trigger_id = str(trigger["id"])
+
+        rlock = RedisLock()
+        lock_key = f"scheduled_trigger_deploy:{trigger_id}"
+        if not rlock.try_lock(lock_key, 30):
+            return
+
+        try:
+            _remove_trigger_jobs(trigger_id)
+        finally:
+            rlock.un_lock(lock_key)

apps/trigger/serializers/trigger.py

@@ -25,6 +25,7 @@
 from knowledge.serializers.common import BatchSerializer
 from maxkb.conf import PROJECT_DIR
 from tools.models import Tool
+from trigger.handler.simple_tools import deploy, undeploy
 from trigger.models import TriggerTypeChoices, Trigger, TriggerTaskTypeChoices, TriggerTask, TaskRecord
 
 
@@ -401,12 +402,17 @@ def batch_delete(self, instance: Dict, with_valid=True):
                 self.is_valid(raise_exception=True)
             workspace_id = self.data.get("workspace_id")
             trigger_id_list = instance.get("id_list")
+            for trigger_id in trigger_id_list:
+                trigger = QuerySet(Trigger).filter(id=trigger_id).first()
+                undeploy(TriggerModelSerializer(trigger).data, **{})
+
             TaskRecord.objects.filter(trigger_id__in=trigger_id_list).delete()
             TriggerTask.objects.filter(trigger_id__in=trigger_id_list).delete()
             Trigger.objects.filter(workspace_id=workspace_id, id__in=trigger_id_list).delete()
 
             return True
 
+        @transaction.atomic
         def batch_switch(self, instance: Dict, with_valid=True):
             if with_valid:
                 BatchActiveSerializer(data=instance).is_valid(model=Trigger, raise_exception=True)
@@ -416,6 +422,15 @@ def batch_switch(self, instance: Dict, with_valid=True):
             is_active = instance.get("is_active")
             Trigger.objects.filter(workspace_id=workspace_id, id__in=trigger_id_list, is_active=not is_active).update(
                 is_active=is_active)
+            if is_active:
+                for trigger_id in trigger_id_list:
+                    trigger = QuerySet(Trigger).filter(id=trigger_id).first()
+                    deploy(TriggerModelSerializer(trigger).data, **{})
+            else:
+                for trigger_id in trigger_id_list:
+                    trigger = QuerySet(Trigger).filter(id=trigger_id).first()
+                    undeploy(TriggerModelSerializer(trigger).data, **{})
+
             return True
 
 
@@ -464,11 +479,20 @@ def edit(self, instance: Dict, with_valid=True):
             ) for task_data in trigger_tasks]
             TriggerTask.objects.bulk_create(trigger_task_model_list)
 
+        # 重新部署触发器任务
+        if trigger.is_active:
+            deploy(TriggerModelSerializer(trigger).data, **{})
+        else:
+            undeploy(TriggerModelSerializer(trigger).data, **{})
+
         return self.one(with_valid=False)
 
     def delete(self):
         self.is_valid(raise_exception=True)
         trigger_id = self.data.get('trigger_id')
+        trigger = QuerySet(Trigger).filter(workspace_id=self.data.get('workspace_id'), id=trigger_id).first()
+        if trigger:
+            undeploy(TriggerModelSerializer(trigger).data, **{})
         TaskRecord.objects.filter(trigger_id=trigger_id).delete()
         TriggerTask.objects.filter(trigger_id=trigger_id).delete()
         Trigger.objects.filter(id=trigger_id).delete()

apps/users/apps.py

@@ -6,4 +6,4 @@ class UsersConfig(AppConfig):
     name = 'users'
 
     def ready(self):
-        from ops.celery import signal_handler
+        from ops.celery import signal_handler  # noqa