wxauto_api/doc/wxauto_center/PLUGIN.md

///

PLUGIN.md

描述：插件系统开发指南

作者：User

创建日期：2026-04-05

更新日期：2026-04-07

///

WXAuto Center 插件系统开发指南

目录

1. 概述
2. 插件类型
3. 创建插件
4. 表单化配置系统
5. 插件通知系统
6. 热加载插件
7. 内置插件
8. 高级多节点推送插件
9. 插件管理API
10. 前端使用
11. 最佳实践
12. 目录结构
13. 生命周期

---

概述

WXAuto Center 采用模块化插件架构，支持热插拔功能扩展。开发者可以创建自定义插件来处理消息、数据源、定时任务等。

核心组件

组件	说明
PluginBase	所有插件的基类
ScheduledTaskPlugin	定时任务插件基类（含通知功能）
PluginRegistry	插件注册表，管理所有插件
PluginType	插件类型枚举
PluginConfigSchema	表单化配置Schema系统
HotReloadPluginLoader	热加载服务，监控插件目录变化
plugin_notification_service	插件通知服务

---

插件类型

类型	值	说明
MESSAGE_HANDLER	message_handler	消息处理器，用于自动回复、消息过滤
DATA_SOURCE	data_source	数据源，用于获取天气、新闻、股票等
ACTION_TRIGGER	action_trigger	动作触发器，用于Webhook回调、定时任务
AI_AGENT	ai_agent	AI智能体，用于LLM对话、智能回复
SCHEDULED_TASK	scheduled_task	定时任务，执行周期性操作
HTTP_SCHEDULED_SENDER	http_scheduled_sender	定时HTTP推送，定时请求API并发送结果
CUSTOM	custom	自定义类型

---

创建插件

基础结构

# ///
# 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("任务完成", "定时任务执行成功")
        return {"success": True}

---

表单化配置系统

字段类型

类型	值	说明	对应UI
STRING	string	文本输入	input[type=text]
PASSWORD	password	密码输入	input[type=password]
NUMBER	number	数字输入	input[type=number]
BOOLEAN	boolean	开关	checkbox
SELECT	select	下拉选择	select
TEXTAREA	textarea	多行文本	textarea
CRON	cron	Cron表达式	input[type=text]

配置示例

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="cron",
                label="执行周期",
                field_type=FieldType.CRON,
                required=True,
                default="*/5 * * * *",
                description="Cron表达式，如 */5 * * * *"
            ),
            ConfigField(
                name="node_id",
                label="节点ID",
                field_type=FieldType.STRING,
                required=True,
                default="wx1"
            ),
            ConfigField(
                name="receiver",
                label="接收人",
                field_type=FieldType.STRING,
                required=True
            ),
            ConfigField(
                name="timeout",
                label="超时时间",
                field_type=FieldType.NUMBER,
                default=15,
                min_value=5,
                max_value=60
            ),
            ConfigField(
                name="enabled",
                label="启用插件",
                field_type=FieldType.BOOLEAN,
                default=False
            )
        ]
    ))

    return schema

API 调用

# 获取插件表单Schema
curl http://localhost:8080/api/v1/plugins/{plugin_name}/schema \
  -H "X-API-Key: your-external-api-key"

# 响应示例
{
  "success": true,
  "schema": {
    "plugin_name": "my_plugin",
    "sections": [
      {
        "name": "basic",
        "label": "基本设置",
        "fields": [
          {"name": "cron", "label": "执行周期", "type": "cron", ...}
        ]
      }
    ]
  },
  "has_form": true
}

---

插件通知系统

所有继承 ScheduledTaskPlugin 的插件都可以使用通知功能。

使用方法

# 普通通知
self.notify(title, message, level)

# 快捷方法
self.notify_error(title, message)      # 错误通知
self.notify_warning(title, message)     # 警告通知
self.notify_success(title, message)     # 成功通知

配置接收

在告警配置中添加 Webhook，勾选 "插件通知" 事件类型：

curl -X POST "http://localhost:8080/api/v1/webhook/" \
  -H "X-API-Key: your-api-key" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "插件通知",
    "url": "https://api.day.app/your-bark-key",
    "format": "bark",
    "event_types": ["plugin_notification"]
  }'

事件类型

事件类型	说明
api_error	API错误
wechat_offline	微信离线
wechat_online	微信上线
node_offline	节点离线
node_online	节点上线
plugin_notification	插件通知

---

热加载插件

工作原理

┌─────────────────────────────────────────────────────────────┐
│              HotReloadPluginLoader (独立线程)                │
│                                                              │
│  1. watchdog 监控 /app/plugins/custom 目录                  │
│  2. 检测到 .py 文件变化 → 自动重新加载                      │
│  3. importlib 动态导入模块                                   │
│  4. 自动注册到 PluginRegistry                               │
└─────────────────────────────────────────────────────────────┘

插件目录

• Docker 内路径: /app/plugins/custom
• 宿主机路径: ./plugins/custom

挂载配置

# docker-compose.yml
volumes:
  - ./plugins/custom:/app/plugins/custom  # 双向同步

---

高级多节点推送插件

简介

advanced_sender 是高级多节点数据推送插件，支持：

• 多数据源独立配置
• 每数据源独立节点和接收人设置
• 外部 MySQL 数据库去重
• 班次时段路由
• Cookie 失效检测和通知
• 熔断保护机制

表单配置

默认设置（可被数据源覆盖）
├── 默认节点ID: wx1
└── 默认接收人: asq

数据源1
├── Cookie: [文本框]
├── 节点ID: [留空使用默认]
└── 接收人: [留空使用默认]

数据源2
├── Cookie: [文本框]
├── 节点ID: [留空使用默认]
└── 接收人: [留空使用默认]

外部数据库
├── 数据库地址: 192.168.2.27
├── 端口: 3306
├── 用户名: root2
├── 密码: ****
├── 数据库名: addb
└── 表名: user_data

抓取设置
├── 请求超时(秒): 15
└── 数据源间隔(秒): 20

配置参数

参数	说明	默认值
cron	执行周期	*/2 * * * *
enable_cron	启用定时任务	false
node_id	默认节点ID	wx1
receiver	默认接收人	asq
cookie_1	数据源1 Cookie	-
cookie_2	数据源2 Cookie	-
node_id_1	数据源1节点ID（留空用默认）	-
receiver_1	数据源1接收人（留空用默认）	-
node_id_2	数据源2节点ID（留空用默认）	-
receiver_2	数据源2接收人（留空用默认）	-
db_host	数据库地址	192.168.2.27
db_port	端口	3306
db_user	用户名	root2
db_password	密码	root@root
db_name	数据库名	addb
db_table	表名	user_data
fetch_timeout	请求超时(秒)	15
send_interval	数据源间隔(秒)	20

班次路由

时段	路由规则
0-7点	跳过
8-16点	发送
17-21点	发送
22-23点	跳过

操作

action	说明
send	执行完整流程
test	测试配置
test_send	测试发送（可指定node_id和receiver）
query_pending	查询待发送数据

测试发送

# 使用默认节点和接收人
curl -X POST "http://localhost:8080/api/v1/plugins/advanced_sender/execute" \
  -H "X-API-Key: your-api-key" \
  -d '{"action": "test_send", "msg": "测试消息"}'

# 指定节点和接收人
curl -X POST "http://localhost:8080/api/v1/plugins/advanced_sender/execute" \
  -H "X-API-Key: your-api-key" \
  -d '{"action": "test_send", "node_id": "wx1", "receiver": "asq", "msg": "测试"}'

获取日志

curl "http://localhost:8080/api/v1/plugins/advanced_sender/logs?lines=100" \
  -H "X-API-Key: your-api-key"

Cookie 失效检测

• 当 HTTP 返回 403/401 时，判定为 Cookie 失效
• 当数据源返回空数据时，判定为 Cookie 可能失效
• 首次检测到问题发送告警通知
• 恢复成功后发送成功通知
• 后续执行不再重复报警

---

插件管理API

接口	方法	说明
GET /api/v1/plugins/	GET	获取所有插件
GET /api/v1/plugins/hotloaded	GET	获取热加载插件列表
POST /api/v1/plugins/hotreload	POST	手动触发热重载
GET /api/v1/plugins/{name}	GET	获取插件详情
GET /api/v1/plugins/{name}/schema	GET	获取插件表单Schema
POST /api/v1/plugins/{name}/enable	POST	启用插件
POST /api/v1/plugins/{name}/disable	POST	禁用插件
PUT /api/v1/plugins/{name}/config	PUT	更新插件配置
POST /api/v1/plugins/{name}/execute	POST	执行插件
GET /api/v1/plugins/{name}/logs	GET	获取插件日志
POST /api/v1/plugins/{name}/logs/clear	POST	清空插件日志

配置插件示例

curl -X PUT "http://localhost:8080/api/v1/plugins/advanced_sender/config" \
  -H "X-API-Key: your-api-key" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "advanced_sender",
    "config": {
      "cron": "*/2 * * * *",
      "enable_cron": true,
      "node_id": "wx1",
      "receiver": "asq",
      "cookie_1": "PHPSESSID=xxx; think_var=zh-cn",
      "cookie_2": "PHPSESSID=yyy; think_var=zh-cn",
      "db_host": "192.168.2.27",
      "db_port": 3306,
      "db_user": "root2",
      "db_password": "root@root",
      "db_name": "addb",
      "db_table": "user_data"
    }
  }'

执行插件示例

# 执行发送任务
curl -X POST "http://localhost:8080/api/v1/plugins/advanced_sender/execute" \
  -H "X-API-Key: your-api-key" \
  -H "Content-Type: application/json" \
  -d '{"action": "send"}'

# 测试配置
curl -X POST "http://localhost:8080/api/v1/plugins/advanced_sender/execute" \
  -H "X-API-Key: your-api-key" \
  -H "Content-Type: application/json" \
  -d '{"action": "test"}'

---

前端使用

访问 插件管理 页面（侧边栏菜单）：

1. 查看插件列表 - 显示所有已注册插件
2. 启用/禁用 - 点击按钮控制插件状态
3. 配置 - 点击配置按钮打开表单化配置界面
4. 日志 - 点击日志按钮查看插件独立日志
5. 执行 - 点击执行按钮测试插件

配置界面

• 表单化配置，无需手动编辑 JSON
• Section 可折叠，方便管理
• 支持所有字段类型（文本、数字、密码、开关、下拉等）
• 自动验证必填项

日志界面

• 实时显示插件日志
• 支持刷新
• 日志文件保存在 /app/data/logs/{plugin_name}.log

---

最佳实践

1. 使用表单化配置

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)
        ]
    ))
    return schema

class MyPlugin(ScheduledTaskPlugin):
    config_schema = staticmethod(get_config_schema)

2. 使用通知功能

class MyPlugin(ScheduledTaskPlugin):
    def run_task(self):
        try:
            # 业务逻辑
            self.notify_success("任务完成", "发送了 10 条消息")
        except Exception as e:
            self.notify_error("任务失败", str(e))

3. 错误处理

def execute(self, params: Dict[str, Any]) -> Dict[str, Any]:
    try:
        return {"success": True, "data": result}
    except Exception as e:
        return {"success": False, "error": str(e)}

4. 定时任务防重复

def run_task(self) -> Dict[str, Any]:
    if self._running:
        logger.warning("Previous task still running, skipping")
        return {"success": False, "error": "上一次执行尚未完成"}

---

目录结构

wxauto_api/
├── plugins/
│   ├── __init__.py
│   ├── base.py                  # 插件基类和注册表
│   ├── builtin.py               # 内置插件实现
│   ├── plugin_loader.py         # 热加载插件加载器
│   ├── plugin_config_schema.py  # 表单化配置系统
│   ├── http_scheduled_sender.py  # 定时HTTP推送插件
│   ├── advanced_sender_plugin.py # 高级多节点推送插件
│   └── custom/                  # 自定义插件目录（热加载）
│       ├── __init__.py
│       └── my_plugin.py
├── services/
│   └── plugin_notification_service.py  # 插件通知服务
├── static/
│   ├── js/
│   │   └── app.js              # 前端插件管理
│   └── css/
│       └── style.css
└── doc/
    └── wxauto_center/
        └── PLUGIN.md           # 本文档

---

生命周期

1. 注册 - PluginRegistry.register(plugin)
2. 初始化 - plugin.initialize(config)
3. 启用 - plugin.enable()
4. 执行 - plugin.execute(params) 或 plugin.run_task()
5. 禁用 - plugin.disable()
6. 卸载 - PluginRegistry.unregister(name)

热加载流程

1. 启动 - HotReloadPluginLoader.start()
2. 扫描 - 扫描 /app/plugins/custom 目录
3. 监听 - watchdog 监控文件变化
4. 加载 - importlib 动态导入模块
5. 注册 - 自动注册到 PluginRegistry
6. 停止 - HotReloadPluginLoader.stop()