跳到主要内容

定时任务

CountBot 内置基于 APScheduler 的定时任务系统，支持 Cron 表达式和单次任务。

概述​

定时任务由 backend/modules/cron/scheduler.py 管理:

  • 基于 AsyncIOScheduler
  • 支持 Cron 表达式和单次任务
  • 任务配置存储在 SQLite 数据库
  • 通过 cron 技能在 Web UI 中管理

CronJob​

每个定时任务对应一个 CronJob 模型:

from backend.modules.cron.scheduler import CronJob

job = CronJob(
    name="每日新闻",
    description="每天早上 8 点发送新闻摘要",
    cron_expression="0 8 * * *",
    prompt="请帮我整理今天的新闻摘要",
    target_channels=["telegram"],
    enabled=True,
    task_type="cron",
)
字段说明
id自动生成
name任务名称
description任务描述
cron_expressionCron 表达式
prompt发送给 AgentLoop 的指令
target_channels输出渠道列表
enabled是否启用
task_type固定值 cron
last_run_time上次运行时间
last_run_status上次运行状态
next_run_time下次运行时间

使用方式​

1. 通过技能管理(推荐)​

在 Web UI 中使用 cron-manager 技能:

请帮我创建一个定时任务，每天早上 8 点发送新闻摘要到 Telegram

技能会自动创建任务并添加到调度器。

2. 直接调用 API​

# 创建任务
curl -X POST http://localhost:8000/api/cron/jobs \
  -H "Content-Type: application/json" \
  -d '{
    "name": "每日新闻",
    "prompt": "整理今日新闻",
    "cron_expression": "0 8 * * *",
    "target_channels": ["telegram"]
  }'

# 查看任务列表
curl http://localhost:8000/api/cron/jobs

# 启用任务
curl -X POST http://localhost:8000/api/cron/jobs/1/enable

# 禁用任务
curl -X POST http://localhost:8000/api/cron/jobs/1/disable

# 删除任务
curl -X DELETE http://localhost:8000/api/cron/jobs/1

# 立即运行任务
curl -X POST http://localhost:8000/api/cron/jobs/1/run

Cron 表达式​

CountBot 使用 APScheduler 的 CronTrigger，支持标准 5 字段 Cron 表达式:

┌────────── 分钟 (0-59)
│ ┌──────── 小时 (0-23)
│ │ ┌────── 日 (1-31)
│ │ │ ┌──── 月 (1-12)
│ │ │ │ ┌── 星期 (0-6, 0=周日)
│ │ │ │ │
* * * * *

常用示例​

表达式含义
0 8 * * *每天早上 8 点
0 */2 * * *每 2 小时
*/15 * * * *每 15 分钟
0 9 * * 1-5工作日早上 9 点
0 0 1 * *每月 1 号零点

执行流程​

调度器触发
  ↓
创建 session_id="cron:<job_id>"
  ↓
AgentLoop.process_direct()
  ↓
生成响应
  ↓
发送到 target_channels
  ↓
更新 last_run_time / last_run_status

任务执行失败不会停止调度器，错误会记录到日志。

数据库存储​

定时任务存储在 cron_jobs 表中:

CREATE TABLE cron_jobs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    description TEXT,
    cron_expression TEXT NOT NULL,
    prompt TEXT NOT NULL,
    target_channels TEXT DEFAULT '[]',
    enabled BOOLEAN DEFAULT 1,
    task_type TEXT DEFAULT 'cron',
    last_run_time DATETIME,
    last_run_status TEXT,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
);

target_channels 以 JSON 格式存储。

API 参考​

方法路径说明
GET/api/cron/jobs获取任务列表
GET/api/cron/jobs/{id}获取任务详情
POST/api/cron/jobs创建任务
PUT/api/cron/jobs/{id}更新任务
DELETE/api/cron/jobs/{id}删除任务
POST/api/cron/jobs/{id}/enable启用任务
POST/api/cron/jobs/{id}/disable禁用任务
POST/api/cron/jobs/{id}/run立即运行

相关文件​

文件说明
backend/modules/cron/scheduler.pyCronScheduler 实现
backend/modules/cron/models.pyCronJob 数据模型
backend/api/cron.pyCron API 路由
workspace/skills/cron-manager/SKILL.mdCron 管理技能