000-platform/docs/scheduled-tasks.md

定时任务系统文档

功能概述

平台定时任务系统，支持 Python 脚本或 Webhook 定时执行，执行结果可自动推送到钉钉/企微机器人。

核心能力：

• 脚本执行：安全沙箱运行 Python 脚本，内置 AI、HTTP、数据库等 SDK
• 调度方式：指定时间点（多选）或 CRON 表达式
• 消息推送：支持钉钉/企微机器人所有消息格式（markdown、actionCard、feedCard 等）
• 失败处理：支持重试和告警通知

---

数据库表

platform_scheduled_tasks（定时任务表）

CREATE TABLE platform_scheduled_tasks (
    id INT AUTO_INCREMENT PRIMARY KEY,
    tenant_id VARCHAR(50) COMMENT '租户ID，空为全局任务',
    task_name VARCHAR(100) NOT NULL COMMENT '任务名称',
    task_desc VARCHAR(500) COMMENT '任务描述',
    schedule_type ENUM('simple', 'cron') NOT NULL DEFAULT 'simple',
    time_points JSON COMMENT '时间点列表 ["08:00", "12:00"]',
    cron_expression VARCHAR(100) COMMENT 'CRON表达式',
    timezone VARCHAR(50) DEFAULT 'Asia/Shanghai',
    execution_type ENUM('webhook', 'script') NOT NULL DEFAULT 'script',
    webhook_url VARCHAR(500),
    script_content TEXT COMMENT 'Python脚本内容',
    script_deps TEXT COMMENT '脚本依赖',
    input_params JSON COMMENT '输入参数',
    retry_count INT DEFAULT 0,
    retry_interval INT DEFAULT 60,
    alert_on_failure TINYINT(1) DEFAULT 0,
    alert_webhook VARCHAR(500),
    notify_channels JSON COMMENT '通知渠道ID列表',
    notify_wecom_app_id INT COMMENT '企微应用ID',
    is_enabled TINYINT(1) DEFAULT 1,
    last_run_at DATETIME,
    last_run_status ENUM('success', 'failed', 'running'),
    last_run_message TEXT,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

platform_task_notify_channels（通知渠道表）

CREATE TABLE platform_task_notify_channels (
    id INT AUTO_INCREMENT PRIMARY KEY,
    tenant_id VARCHAR(50) NOT NULL COMMENT '租户ID',
    channel_name VARCHAR(100) NOT NULL COMMENT '渠道名称',
    channel_type ENUM('dingtalk_bot', 'wecom_bot') NOT NULL,
    webhook_url VARCHAR(500) NOT NULL,
    sign_secret VARCHAR(200) COMMENT '钉钉加签密钥',
    description VARCHAR(255),
    is_enabled TINYINT(1) DEFAULT 1,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

platform_task_logs（执行日志表）

CREATE TABLE platform_task_logs (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    task_id INT NOT NULL,
    tenant_id VARCHAR(50),
    trace_id VARCHAR(100),
    status ENUM('running', 'success', 'failed'),
    started_at DATETIME,
    finished_at DATETIME,
    duration_ms INT,
    output TEXT,
    error TEXT,
    retry_count INT DEFAULT 0,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);

---

后端文件结构

backend/app/
├── models/
│   ├── scheduled_task.py      # ScheduledTask, TaskLog, ScriptVar, Secret 模型
│   └── notification_channel.py # TaskNotifyChannel 模型
├── routers/
│   ├── tasks.py               # 定时任务 API (/api/scheduled-tasks)
│   └── notification_channels.py # 通知渠道 API (/api/notification-channels)
└── services/
    ├── scheduler.py           # APScheduler 调度服务
    ├── script_executor.py     # 脚本执行器（安全沙箱）
    └── script_sdk.py          # 脚本内置 SDK

---

脚本 SDK 文档

内置函数

# 日志
log(message)                    # 记录日志
print(message)                  # 打印输出

# AI 调用
ai(prompt, system=None, model='deepseek-chat')  # 调用 AI

# 通知发送（直接发送，不走 result）
dingtalk(webhook_url, content, title='通知')
wecom(webhook_url, content)

# HTTP 请求
http_get(url, headers=None, params=None)
http_post(url, data=None, json=None, headers=None)

# 数据库查询（只读）
db_query(sql, params=None)

# 变量存储（跨执行持久化）
get_var(key, default=None)
set_var(key, value)
del_var(key)

# 任务参数
get_param(key, default=None)   # 获取单个参数
get_params()                    # 获取所有参数

# 租户相关
get_tenants()                   # 获取所有租户
get_tenant_config(tenant_id, app_code, key)  # 获取租户配置
get_all_tenant_configs(app_code, key)        # 获取所有租户的配置

# 密钥
get_secret(key)                 # 获取密钥

内置变量

task_id     # 当前任务ID
tenant_id   # 当前租户ID（可能为空）
trace_id    # 追踪ID

内置模块（无需 import）

datetime    # datetime.now(), datetime.strptime()
date        # date.today()
timedelta   # timedelta(days=1)
time        # time.sleep(), time.time()
json        # json.dumps(), json.loads()
re          # re.search(), re.match()
math        # math.ceil(), math.floor()
random      # random.randint(), random.choice()
hashlib     # hashlib.md5()
base64      # base64.b64encode()

---

消息格式（result 变量）

基础格式（默认 markdown）

result = {
    'content': 'Markdown 内容',
    'title': '消息标题'
}

钉钉 ActionCard（交互卡片）

result = {
    'msg_type': 'actionCard',
    'title': '卡片标题',
    'content': '''### 正文内容
| 列1 | 列2 |
|:---:|:---:|
| A | B |
''',
    'btn_orientation': '1',  # 0-竖向 1-横向
    'buttons': [
        {'title': '按钮1', 'url': 'https://...'},
        {'title': '按钮2', 'url': 'https://...'}
    ]
}

钉钉 FeedCard（信息流）

result = {
    'msg_type': 'feedCard',
    'links': [
        {'title': '标题1', 'url': 'https://...', 'pic_url': 'https://...'},
        {'title': '标题2', 'url': 'https://...', 'pic_url': 'https://...'}
    ]
}

钉钉 Link（链接消息）

result = {
    'msg_type': 'link',
    'title': '链接标题',
    'content': '链接描述',
    'url': 'https://...',
    'pic_url': 'https://...'
}

企微 News（图文消息）

result = {
    'msg_type': 'news',
    'articles': [
        {
            'title': '文章标题',
            'description': '文章描述',
            'url': 'https://...',
            'picurl': 'https://...'
        }
    ]
}

企微 Template Card（模板卡片）

result = {
    'msg_type': 'template_card',
    'card_type': 'text_notice',  # text_notice / news_notice / button_interaction
    'title': '卡片标题',
    'content': '卡片内容',
    'horizontal_list': [
        {'keyname': '申请人', 'value': '张三'},
        {'keyname': '金额', 'value': '¥5,000'}
    ],
    'jump_list': [
        {'type': 1, 'title': '查看详情', 'url': 'https://...'}
    ]
}

---

API 端点

定时任务

方法	路径	说明
GET	/api/scheduled-tasks	任务列表
GET	/api/scheduled-tasks/{id}	任务详情
POST	/api/scheduled-tasks	创建任务
PUT	/api/scheduled-tasks/{id}	更新任务
DELETE	/api/scheduled-tasks/{id}	删除任务
POST	/api/scheduled-tasks/{id}/toggle	启用/禁用
POST	/api/scheduled-tasks/{id}/run	立即执行
GET	/api/scheduled-tasks/{id}/logs	执行日志
POST	/api/scheduled-tasks/test-script	测试脚本
GET	/api/scheduled-tasks/sdk-docs	SDK 文档

通知渠道

方法	路径	说明
GET	/api/notification-channels	渠道列表
POST	/api/notification-channels	创建渠道
PUT	/api/notification-channels/{id}	更新渠道
DELETE	/api/notification-channels/{id}	删除渠道
POST	/api/notification-channels/{id}/test	测试渠道

---

前端文件

frontend/src/views/
├── scheduled-tasks/
│   └── index.vue              # 定时任务管理页面
└── notification-channels/
    └── index.vue              # 通知渠道管理页面

---

示例脚本

基础示例

# 无需 import，模块已内置
log('任务开始执行')

now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
prompt = get_param('prompt', '默认提示词')

content = ai(prompt, system='你是一个助手')

result = {
    'title': '每日推送',
    'content': f'**生成时间**: {now}\n\n{content}'
}

log('任务执行完成')

复杂卡片示例

log('生成销售日报')

now = datetime.now()
today = now.strftime('%Y年%m月%d日')

# 模拟数据
revenue = random.randint(50000, 150000)

result = {
    'msg_type': 'actionCard',
    'title': f'销售日报 | {today}',
    'content': f'''### 今日业绩

| 指标 | 数值 |
|:---:|:---:|
| 销售额 | **¥{revenue:,}** |
| 订单数 | **{random.randint(40, 80)}** |

> 点击查看详情
''',
    'buttons': [
        {'title': '查看详情', 'url': 'https://example.com/report'}
    ]
}

---

部署信息

• 测试环境: https://platform.test.ai.ireborn.com.cn
• 数据库: new_qiqi (测试) / new_platform_prod (生产)
• Docker 容器: platform-backend-test / platform-frontend-test