feat: add ToolTask implementation and update scheduled trigger functions

Author: liuruibin

apps/trigger/handler/impl/task/tool_task.py

@@ -0,0 +1,109 @@
+# coding=utf-8
+"""
+    @project: MaxKB
+    @Author：虎虎
+    @file： application_task.py
+    @date：2026/1/14 19:14
+    @desc:
+"""
+
+import uuid_utils.compat as uuid
+from django.db.models import QuerySet
+
+from common.utils.tool_code import ToolExecutor
+from knowledge.models.knowledge_action import State
+from tools.models import Tool
+from trigger.handler.base_task import BaseTriggerTask
+from trigger.models import TaskRecord
+
+
+def get_reference(fields, obj):
+    for field in fields:
+        value = obj.get(field)
+        if value is None:
+            return None
+        else:
+            obj = value
+    return obj
+
+
+def get_field_value(value, kwargs):
+    source = value.get('source')
+    if source == 'custom':
+        return value.get('value')
+    else:
+        return get_reference(value.get('value'), kwargs)
+
+
+def get_application_execute_parameters(parameter_setting, kwargs):
+    parameters = {'form_data': {}}
+    question_setting = parameter_setting.get('question')
+    if question_setting:
+        parameters['message'] = get_field_value(question_setting, kwargs)
+    filed_list = ['image_list', 'document_list', 'audio_list', 'video_list', 'other_list']
+    for field in filed_list:
+        field_setting = parameter_setting.get(field)
+        if field_setting:
+            parameters[field] = get_field_value(field_setting, kwargs)
+    api_input_field_list = parameter_setting.get('api_input_field_list')
+    if api_input_field_list:
+        for key, value in api_input_field_list.items():
+            parameters['form_data'][key] = get_field_value(value, kwargs)
+    user_input_field_list = parameter_setting.get('user_input_field_list')
+    if user_input_field_list:
+        for key, value in user_input_field_list.items():
+            parameters['form_data'][key] = get_field_value(value, kwargs)
+    return parameters
+
+
+def get_loop_workflow_node(node_list):
+    result = []
+    for item in node_list:
+        if item.get('type') == 'loop-node':
+            for loop_item in item.get('loop_node_data') or []:
+                for inner_item in loop_item.values():
+                    result.append(inner_item)
+    return result
+
+
+def get_workflow_state(details):
+    node_list = details.values()
+    all_node = [*node_list, *get_loop_workflow_node(node_list)]
+    err = any([True for value in all_node if value.get('status') == 500 and not value.get('enableException')])
+    if err:
+        return State.FAILURE
+    return State.SUCCESS
+
+
+class ToolTask(BaseTriggerTask):
+    def support(self, trigger_task, **kwargs):
+        return trigger_task.get('source_type') == 'TOOL'
+
+    def execute(self, trigger_task, **kwargs):
+        parameter_setting = trigger_task.get('parameter')
+        parameters = get_application_execute_parameters(parameter_setting, kwargs)
+        tool_id = trigger_task.get('source_id')
+        task_record_id = uuid.uuid7()
+
+        TaskRecord(
+            id=task_record_id,
+            trigger_id=trigger_task.get('trigger_id'),
+            trigger_task_id=trigger_task.get('id'),
+            source_type="TOOL",
+            source_id=tool_id,
+            task_record_id=task_record_id,
+            meta={},
+            state=State.STARTED
+        ).save()
+
+        try:
+            tool = QuerySet(Tool).filter(id=tool_id).first()
+            executor = ToolExecutor()
+            # executor.exec_code(tool.code, parameters)
+            print(tool)
+            print(parameters)
+
+            QuerySet(TaskRecord).filter(id=task_record_id).update(state=State.SUCCESS, run_time=0)
+        except Exception as e:
+            state = State.FAILURE
+            QuerySet(TaskRecord).filter(id=task_record_id).update(state=state, run_time=0)

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

@@ -1,5 +1,4 @@
 # coding=utf-8
-from __future__ import annotations
 
 import random
 
@@ -37,7 +36,7 @@ def _get_active_trigger_tasks(trigger_id: str) -> list[dict]:
     )
 
 
-def _deploy_daily(trigger: dict, trigger_tasks: list[dict], setting: dict, trigger_id: str, func) -> None:
+def _deploy_daily(trigger: dict, trigger_tasks: list[dict], setting: dict, trigger_id: str) -> None:
     from common.job import scheduler
 
     times = setting.get("time") or []
@@ -51,17 +50,17 @@ def _deploy_daily(trigger: dict, trigger_tasks: list[dict], setting: dict, trigg
         for task in trigger_tasks:
             job_id = f"trigger:{trigger_id}:task:{task['id']}:daily:{hour:02d}{minute:02d}"
             scheduler.add_job(
-                func,
+                ScheduledTrigger.execute,
                 trigger="cron",
                 hour=str(hour),
                 minute=str(minute),
                 id=job_id,
-                kwargs={"trigger": trigger, "trigger_tasks": trigger_tasks},
+                kwargs={"trigger": trigger, "trigger_task": task},
                 replace_existing=True,
             )
 
 
-def _deploy_weekly(trigger: dict, trigger_tasks: list[dict], setting: dict, trigger_id: str, func) -> None:
+def _deploy_weekly(trigger: dict, trigger_tasks: list[dict], setting: dict, trigger_id: str) -> None:
     from common.job import scheduler
 
     times = setting.get("time") or []
@@ -87,18 +86,18 @@ def _deploy_weekly(trigger: dict, trigger_tasks: list[dict], setting: dict, trig
             for task in trigger_tasks:
                 job_id = f"trigger:{trigger_id}:task:{task['id']}:weekly:{dow}:{hour:02d}{minute:02d}"
                 scheduler.add_job(
-                    func,
+                    ScheduledTrigger.execute,
                     trigger="cron",
                     day_of_week=dow,
                     hour=str(hour),
                     minute=str(minute),
                     id=job_id,
-                    kwargs={"trigger": trigger, "trigger_tasks": trigger_tasks},
+                    kwargs={"trigger": trigger, "trigger_task": task},
                     replace_existing=True,
                 )
 
 
-def _deploy_monthly(trigger: dict, trigger_tasks: list[dict], setting: dict, trigger_id: str, func) -> None:
+def _deploy_monthly(trigger: dict, trigger_tasks: list[dict], setting: dict, trigger_id: str) -> None:
     from common.job import scheduler
 
     times = setting.get("time") or []
@@ -126,18 +125,18 @@ def _deploy_monthly(trigger: dict, trigger_tasks: list[dict], setting: dict, tri
             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,
+                    ScheduledTrigger.execute,
                     trigger="cron",
                     day=str(dom),
                     hour=str(hour),
                     minute=str(minute),
                     id=job_id,
-                    kwargs={"trigger": trigger, "trigger_tasks": trigger_tasks},
+                    kwargs={"trigger": trigger, "trigger_task": task},
                     replace_existing=True,
                 )
 
 
-def _deploy_interval(trigger: dict, trigger_tasks: list[dict], setting: dict, trigger_id: str, func) -> None:
+def _deploy_interval(trigger: dict, trigger_tasks: list[dict], setting: dict, trigger_id: str) -> None:
     from common.job import scheduler
 
     unit = (setting.get("interval_unit") or "").strip()
@@ -158,10 +157,10 @@ def _deploy_interval(trigger: dict, trigger_tasks: list[dict], setting: dict, tr
     for task in trigger_tasks:
         job_id = f"trigger:{trigger_id}:task:{task['id']}:interval:{unit}:{value_i}"
         scheduler.add_job(
-            func,
+            ScheduledTrigger.execute,
             trigger="interval",
             id=job_id,
-            kwargs={"trigger": trigger, "trigger_tasks": trigger_tasks},
+            kwargs={"trigger": trigger, "trigger_task": task},
             replace_existing=True,
             **{unit: value_i},
         )
@@ -179,9 +178,9 @@ def _remove_trigger_jobs(trigger_id: str) -> None:
                 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'])
+@celery_app.task(name="celery:deploy_scheduled_trigger")
+def deploy_scheduled_trigger(trigger: dict, trigger_tasks: list[dict], setting: dict, schedule_type: str) -> None:
+    _remove_trigger_jobs(trigger["id"])
 
     deployers = {
         "daily": _deploy_daily,
@@ -194,7 +193,7 @@ def deploy_scheduled_trigger(trigger: dict, trigger_tasks: list[dict], setting:
         maxkb_logger.warning(f"unsupported schedule_type={schedule_type}, trigger_id={trigger['id']}")
         return
 
-    fn(trigger, trigger_tasks, setting, trigger['id'], func)
+    fn(trigger, trigger_tasks, setting, trigger["id"])
 
 
 class ScheduledTrigger(BaseTrigger):
@@ -205,6 +204,23 @@ class ScheduledTrigger(BaseTrigger):
     @staticmethod
     def execute(trigger, **kwargs):
         n = random.randint(1, 1_000_000_000)
+        trigger_task = kwargs.get("trigger_task")
+        if not trigger_task:
+            maxkb_logger.warning(f"unsupported task={trigger_task}")
+            return
+        source_type = trigger_task["source_type"]
+
+        if source_type == "APPLICATION":
+            from trigger.handler.impl.task.application_task import ApplicationTask
+
+            ApplicationTask.execute(trigger_task, **kwargs)
+        elif source_type == "TOOL":
+            from trigger.handler.impl.task.tool_task import ToolTask
+
+            ToolTask.execute(trigger_task, **kwargs)
+        else:
+            maxkb_logger.warning(f"unsupported source_type={source_type}, task_id={trigger_task['id']}")
+            return
 
         maxkb_logger.info(f"scheduled trigger execute, trigger={n}")
 
@@ -233,8 +249,7 @@ def deploy(self, trigger, **kwargs):
 
         try:
             maxkb_logger.debug(f"get lock {lock_key}")
-            deploy_scheduled_trigger.delay(trigger, trigger_tasks, setting, schedule_type, self.execute)
-
+            deploy_scheduled_trigger.delay(trigger, trigger_tasks, setting, schedule_type)
         finally:
             rlock.un_lock(lock_key)
 

apps/trigger/tasks.py

@@ -0,0 +1,6 @@
+# apps/trigger/tasks.py
+# coding=utf-8
+from __future__ import annotations
+
+# 作为 Celery autodiscover 的入口，确保任务模块被导入从而完成注册
+from trigger.handler.impl.trigger.scheduled_trigger import deploy_scheduled_trigger  # noqa: F401