Custom Plugins Directory

此目录用于存放热加载的自定义插件。

热加载机制

插件文件放在此目录下
修改插件文件后会自动重新加载
无需重启 Docker

创建插件

# ///
# my_plugin.py
# 描述：我的自定义插件
# ///

from plugins.base import PluginBase, PluginType

class MyPlugin(PluginBase):
    plugin_name = "my_plugin"
    plugin_version = "1.0.0"
    plugin_type = PluginType.CUSTOM
    plugin_description = "我的自定义插件"
    plugin_author = "开发者"

    def initialize(self, config: dict) -> bool:
        self.config = config
        return True

    def execute(self, params: dict) -> dict:
        return {"success": True, "result": "done"}

定时任务插件示例

# ///
# my_scheduled_plugin.py
# 描述：定时任务插件示例
# ///

from plugins.base import ScheduledTaskPlugin
from plugins.plugin_config_schema import (
    PluginConfigSchema, ConfigSection, ConfigField, FieldType
)

def get_config_schema() -> PluginConfigSchema:
    schema = PluginConfigSchema("my_scheduled_plugin")
    schema.add_section(ConfigSection(
        name="basic",
        label="基本设置",
        fields=[
            ConfigField(name="cron", label="执行周期", field_type=FieldType.CRON, default="*/5 * * * *"),
            ConfigField(name="enabled", label="启用", field_type=FieldType.BOOLEAN, default=False)
        ]
    ))
    return schema

class MyScheduledPlugin(ScheduledTaskPlugin):
    plugin_name = "my_scheduled_plugin"
    plugin_version = "1.0.0"
    plugin_description = "定时任务插件"
    plugin_author = "开发者"
    config_schema = staticmethod(get_config_schema)

    def get_cron_expression(self) -> str:
        return self.cron_expression

    def should_run_and_get_next(self) -> tuple:
        return True, None

    def run_task(self) -> dict:
        self.notify_success("任务完成", "执行成功")
        return {"success": True}

通知功能

所有继承 ScheduledTaskPlugin 的插件可以使用通知：

self.notify("标题", "消息", "info")
self.notify_error("错误", "错误详情")
self.notify_warning("警告", "警告内容")
self.notify_success("成功", "成功详情")

表单化配置

from plugins.plugin_config_schema import (
    PluginConfigSchema, ConfigSection, ConfigField, FieldType
)

def get_config_schema() -> PluginConfigSchema:
    schema = PluginConfigSchema("my_plugin")

    schema.add_section(ConfigSection(
        name="basic",
        label="基本设置",
        fields=[
            ConfigField(name="key", label="键", field_type=FieldType.STRING, required=True),
            ConfigField(name="timeout", label="超时", field_type=FieldType.NUMBER, default=15),
            ConfigField(name="enabled", label="启用", field_type=FieldType.BOOLEAN, default=False)
        ]
    ))

    return schema

字段类型

类型	说明
`STRING`	文本输入
`PASSWORD`	密码输入
`NUMBER`	数字输入
`BOOLEAN`	开关
`SELECT`	下拉选择
`TEXTAREA`	多行文本
`CRON`	Cron表达式

更多信息请参阅插件开发指南