feat: Trigger handler init

Author: shaohuzhang1

apps/trigger/handler/base_task.py

@@ -0,0 +1,23 @@
+# coding=utf-8
+"""
+    @project: MaxKB
+    @Author：虎虎
+    @file： base_task.py
+    @date：2026/1/14 19:03
+    @desc:
+"""
+from abc import ABC, abstractmethod
+
+
+class BaseTriggerTask(ABC):
+    """
+    任务执行器抽象
+    """
+
+    @abstractmethod
+    def support(self, trigger_task, **kwargs):
+        pass
+
+    @abstractmethod
+    def execute(self, trigger_task, **kwargs):
+        pass

apps/trigger/handler/base_trigger.py

@@ -0,0 +1,28 @@
+# coding=utf-8
+"""
+    @project: MaxKB
+    @Author：虎虎
+    @file： Trigger.py
+    @date：2026/1/14 18:45
+    @desc:
+"""
+
+from abc import ABC, abstractmethod
+
+
+class BaseTrigger(ABC):
+    """
+    触发器抽象
+    """
+
+    @abstractmethod
+    def support(self, trigger, **kwargs):
+        pass
+
+    @abstractmethod
+    def deploy(self, trigger, **kwargs):
+        pass
+
+    @abstractmethod
+    def undeploy(self, trigger, **kwargs):
+        pass

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

@@ -0,0 +1,17 @@
+# coding=utf-8
+"""
+    @project: MaxKB
+    @Author：虎虎
+    @file： application_task.py
+    @date：2026/1/14 19:14
+    @desc:
+"""
+from trigger.handler.base_task import BaseTriggerTask
+
+
+class ApplicationTask(BaseTriggerTask):
+    def support(self, trigger_task, **kwargs):
+        return trigger_task.get('source_type') == 'APPLICATION'
+
+    def execute(self, trigger_task, **kwargs):
+        pass

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

@@ -0,0 +1,21 @@
+# coding=utf-8
+"""
+    @project: MaxKB
+    @Author：虎虎
+    @file： scheduled_trigger.py
+    @date：2026/1/14 18:57
+    @desc:
+"""
+from trigger.handler.base_trigger import BaseTrigger
+
+
+class ScheduledTrigger(BaseTrigger):
+
+    def support(self, trigger, **kwargs):
+        return trigger.get('trigger_type') == 'SCHEDULED'
+
+    def deploy(self, trigger, **kwargs):
+        pass
+
+    def undeploy(self, trigger, **kwargs):
+        pass

apps/trigger/handler/simple_tools.py

@@ -0,0 +1,53 @@
+# coding=utf-8
+"""
+    @project: MaxKB
+    @Author：虎虎
+    @file： simple_task.py
+    @date：2026/1/14 19:18
+    @desc:
+"""
+from trigger.handler.impl.task.application_task import ApplicationTask
+from trigger.handler.impl.trigger.scheduled_trigger import ScheduledTrigger
+
+simple_task_handlers = [ApplicationTask]
+
+simple_trigger_handlers = [ScheduledTrigger]
+
+
+def execute(trigger_task, **kwargs):
+    """
+    执行触发器任务
+    @param trigger_task:  触发器任务数据
+    @param kwargs:        额外数据
+    @return:
+    """
+    for simple_task_handler in simple_task_handlers:
+        if simple_task_handler.support(trigger_task, **kwargs):
+            return simple_task_handler.execute(trigger_task, **kwargs)
+    raise Exception("不支持的处理器类型")
+
+
+def deploy(trigger, **kwargs):
+    """
+    部署触发器
+    @param trigger: 触发器字典数据
+    @param kwargs:  额外数据
+    @return:
+    """
+    for simple_trigger_handler in simple_trigger_handlers:
+        if simple_trigger_handler.support(trigger, **kwargs):
+            return simple_trigger_handler.deploy(trigger, **kwargs)
+    raise Exception("不支持的触发器类型")
+
+
+def undeploy(trigger, **kwargs):
+    """
+    取消部署触发器
+    @param trigger: 触发器字典数据
+    @param kwargs:  额外数据
+    @return:
+    """
+    for simple_trigger_handler in simple_trigger_handlers:
+        if simple_trigger_handler.support(trigger, **kwargs):
+            return simple_trigger_handler.undeploy(trigger, **kwargs)
+    raise Exception("不支持的触发器类型")