Install
openclaw skills install @thcjp/cron-scheduler-pro-freeopenclaw skills install @thcjp/cron-scheduler-pro-free将重复性意图转化为可信赖的执行契约。本地优先,隐私安全,四种调度模式。
重复的任务只需配置一次,然后放心交给调度引擎执行。本技能提供本地优先的定时任务调度框架,支持每日、每周、每月、间隔四种调度模式,配合完整的任务生命周期管理,让Agent按计划自动执行。
┌─────────────────────────────────────────────────┐
│ 定时调度引擎 (免费版) │
├─────────────────────────────────────────────────┤
│ │
│ ┌────────────────────────────────────────┐ │
│ │ 调度模式层 │ │
│ │ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ │ │
│ │ │ 每日 │ │ 每周 │ │ 每月 │ │ 间隔 │ │ │
│ │ │ daily│ │weekly│ │monthly│ │interval│ │ │
│ │ └──────┘ └──────┘ └──────┘ └──────┘ │ │
│ └────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌────────────────────────────────────────┐ │
│ │ 任务管理层 │ │
│ │ 创建 → 激活 → 暂停 → 恢复 → 归档 │ │
│ │ active paused archived │ │
│ └────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌────────────────────────────────────────┐ │
│ │ 持久化存储层 │ │
│ │ jobs.json │ runs.json │ stats │ │
│ │ (任务定义) │ (执行历史) │ (统计) │ │
│ └────────────────────────────────────────┘ │
└─────────────────────────────────────────────────┘
python3 --version
import json
from pathlib import Path
from datetime import datetime, timedelta
STORE_DIR = Path.home() / "workspace" / "scheduler" / "cron"
STORE_DIR.mkdir(parents=True, exist_ok=True)
JOBS_FILE = STORE_DIR / "jobs.json"
if not JOBS_FILE.exists():
JOBS_FILE.write_text("[]", encoding="utf-8")
def add_job(name, schedule_type, schedule_config, task_desc):
"""添加定时任务"""
jobs = json.loads(JOBS_FILE.read_text(encoding="utf-8"))
job = {
"id": f"job_{len(jobs) + 1:04d}",
"name": name,
"schedule_type": schedule_type, # daily/weekly/monthly/interval
"schedule_config": schedule_config,
"task": task_desc,
"status": "active",
"created_at": datetime.now().isoformat(),
"next_run": None # 待计算
}
jobs.append(job)
JOBS_FILE.write_text(json.dumps(jobs, ensure_ascii=False, indent=2), encoding="utf-8")
print(f"任务已创建:{job['id']} - {job['name']}")
return job
add_job(
name="每日晨报",
schedule_type="daily",
schedule_config={"time": "08:00"},
task_desc="生成每日工作晨报,包含待办事项和日程提醒"
)
add_job(
name="API健康检查",
schedule_type="interval",
schedule_config={"every_minutes": 30},
task_desc="检查API端点健康状态,异常时通知"
)
配置完整的任务管理流程:
详细代码示例已移至
references/detail.md
配置执行历史记录与任务查看:
from datetime import datetime
import json
class FullScheduler(CronScheduler):
"""带执行历史的完整调度器"""
def record_run(self, job_id, status, result=""):
"""记录执行历史"""
runs = json.loads(self.runs_file.read_text(encoding="utf-8"))
run = {
"job_id": job_id,
"run_time": datetime.now().isoformat(),
"status": status, # success / failed / skipped
"result": result
}
runs.append(run)
self.runs_file.write_text(
json.dumps(runs, ensure_ascii=False, indent=2), encoding="utf-8")
jobs = self._load_jobs()
for j in jobs:
if j["id"] == job_id:
j["last_run"] = run["run_time"]
if j["status"] == "active":
j["next_run"] = self._calc_next_run(
j["schedule_type"], j["schedule_config"])
break
self._save_jobs(jobs)
def get_run_history(self, job_id=None, limit=20):
"""查看执行历史"""
runs = json.loads(self.runs_file.read_text(encoding="utf-8"))
if job_id:
runs = [r for r in runs if r["job_id"] == job_id]
runs = runs[-limit:]
print(f"\n{'任务ID':<12} {'执行时间':<26} {'状态':<10} {'结果':<30}")
print("-" * 80)
for r in runs:
print(f"{r['job_id']:<12} {r['run_time']:<26} "
f"{r['status']:<10} {r['result'][:30]}")
return runs
def get_next_runs(self, limit=5):
"""查看接下来要执行的任务"""
jobs = self._load_jobs()
active = [j for j in jobs if j["status"] == "active" and j.get("next_run")]
active.sort(key=lambda j: j["next_run"])
print(f"\n接下来 {min(limit, len(active))} 个待执行任务:")
print(f"{'ID':<12} {'名称':<16} {'执行时间':<26} {'任务描述':<30}")
print("-" * 80)
for j in active[:limit]:
print(f"{j['id']:<12} {j['name']:<16} "
f"{j['next_run']:<26} {j['task'][:30]}")
return active[:limit]
fs = FullScheduler()
fs.record_run("job_0001", "success", "晨报已生成并发送")
fs.get_next_runs()
fs.get_run_history()
| 模式 | 配置参数 | 适用场景 | 示例 |
|---|---|---|---|
| daily(每日) | time: 执行时间 | 每日固定任务 | 每日08:00生成晨报 |
| weekly(每周) | weekday: 星期几(0-6), time: 时间 | 每周固定任务 | 每周五17:00生成周报 |
| monthly(每月) | day: 日期(1-31), time: 时间 | 每月固定任务 | 每月1日02:00数据备份 |
| interval(间隔) | every_minutes: 间隔分钟数 | 轮询监控类任务 | 每30分钟健康检查 |
输入: 用户提供调度模式所需的指令和必要参数。 处理: 按照skill规范执行调度模式操作,遵循单一意图原则。 输出: 返回调度模式的执行结果,包含操作状态和输出数据。
创建 手动暂停
┌──────┐ ┌──────────┐
│ ▼ │ ▼
│ ┌─────────┐ │ ┌─────────┐
│ │ active │◀──────│ paused │
│ │ (活跃) │ │ │ (暂停) │
│ └────┬────┘ │ └────┬────┘
│ │ │ │
│ │ 归档 │ │ 归档
│ ▼ │ ▼
│ ┌─────────┐ │ ┌─────────┐
└──▶│archived │◀──────│archived │
│ (归档) │ │ (归档) │
└─────────┘ └─────────┘
输入: 用户提供任务生命周期所需的指令和必要参数。 处理: 按照skill规范执行任务生命周期操作,遵循单一意图原则。 输出: 返回任务生命周期的执行结果,包含操作状态和输出数据。
input_params参数,支持创建/查询/导出操作| 操作 | 方法 | 说明 |
|---|---|---|
| 创建任务 | add_job(name, type, config, task) | 创建并激活新任务 |
| 列出任务 | list_jobs(status=None) | 按状态过滤查看 |
| 暂停任务 | pause_job(job_id) | 临时停止执行 |
| 恢复任务 | resume_job(job_id) | 恢复暂停的任务 |
| 归档任务 | archive_job(job_id) | 永久停用但保留历史 |
| 查看详情 | show_job(job_id) | 查看单个任务信息 |
| 查看下次执行 | get_next_runs(limit) | 查看待执行队列 |
| 执行历史 | get_run_history(job_id) | 查看执行记录 |
输入: 用户提供任务管理命令所需的指令和必要参数。 处理: 按照skill规范执行任务管理命令操作,遵循单一意图原则。 输出: 返回任务管理命令的执行结果,包含操作状态和输出数据。
所有数据存储在本地,路径为 ~/workspace/scheduler/cron/:
| 文件 | 用途 |
|---|---|
jobs.json | 任务定义与配置 |
runs.json | 执行历史记录 |
stats.json | 统计数据(专业版) |
输入: 用户提供数据存储所需的指令和必要参数。 处理: 按照skill规范执行数据存储操作,遵循单一意图原则。 输出: 返回数据存储的执行结果,包含操作状态和输出数据。 能力覆盖范围:本skill的核心能力覆盖以下场景关键词:本地优先的定时任、务调度引擎免费版、支持每日、间隔四种调度模式、与任务生命周期管、定时调度引擎免费、版是一套面向、Agent、的本地优先定时任、务调度框架、将重复性意图转化、为结构化的执行契、按计划自动执行任、when、需要项目管理、任务规划、进度跟踪、团队协作时使用、不适用于实际人员、绩效评估、适用于独立开发者、企业团队和自动化、工作流场景等。这些关键词对应description中声明的使用场景,均已在上述能力点中提供对应的操作支持。
角色:独立开发者
场景描述:每天早上8点自动生成工作晨报,包含昨日完成事项和今日待办。
scheduler = CronScheduler()
scheduler.add_job(
name="每日晨报",
schedule_type="daily",
schedule_config={"time": "08:00"},
task="读取昨日工作日志,生成晨报,包含:1.昨日完成事项 2.今日待办 3.日程提醒"
)
角色:运维工程师
场景描述:每30分钟检查一次API端点健康状态,异常时通知。
scheduler.add_job(
name="API健康检查",
schedule_type="interval",
schedule_config={"every_minutes": 30},
task="检查以下API端点状态:/health, /api/v1/status, /api/v1/metrics。异常时发送通知"
)
角色:系统管理员
场景描述:每月1日凌晨2点执行全量数据备份。
scheduler.add_job(
name="月度全量备份",
schedule_type="monthly",
schedule_config={"day": 1, "time": "02:00"},
task="执行全量数据备份:1.数据库导出 2.文件系统快照 3.验证备份完整性 4.清理30天前旧备份"
)
所有数据存储在本地 ~/workspace/scheduler/cron/ 目录下,包含三个JSON文件:jobs.json(任务定义)、runs.json(执行历史)、stats.json(统计数据)。不涉及云端同步,确保数据隐私。可通过修改 STORE_DIR 变量自定义存储路径。
根据任务特性选择:(1) 固定每天某个时间执行用 daily;(2) 固定每周某天执行用 weekly;(3) 固定每月某天执行用 monthly;(4) 每隔N分钟轮询类任务用 interval。如果需要cron表达式级别的精确控制(如"每季度第一个周一"),建议升级专业版。
不会。暂停时 next_run 字段保留,恢复时自动重新计算。暂停期间不会触发任何执行。归档的任务保留所有历史数据但不再参与调度。
Agent在每次会话开始时调用 get_next_runs() 查看待执行队列,检查 next_run 时间是否已到。如果当前时间超过 next_run,则执行该任务并调用 record_run() 记录结果,引擎自动计算下次执行时间。
免费版支持管理最多50个活跃任务。任务数据存储在本地JSON文件中,50个以内的任务读写性能良好。如果需要管理更多任务或需要数据库级查询能力,建议升级专业版。
| 依赖项 | 类型 | 是否必需 | 获取方式 |
|---|---|---|---|
| Python标准库 | 内置 | 必需 | Python自带(json/pathlib/datetime) |
| LLM API | API | 必需 | 由Agent平台内置LLM提供 |
本技能基于原始开源定时调度作品改进,保留原始版权声明:
本改进作品在原始作品基础上进行了深度差异化改造,包括但不限于:
原始MIT-0 license允许使用、复制、修改和分发,无需保留版权声明。本改进作品在保留原始版权声明的基础上添加自有署名,完全符合MIT license要求。
本免费体验版限制以下高级功能:
解锁全部功能请使用专业版:cron-scheduler-pro-pro
| 错误场景 | 原因 | 处理方式 |
|---|---|---|
| 配置错误 | 参数缺失或格式错误 | 检查依赖说明中的配置要求 |
| 运行时错误 | 运行环境不满足 | 确认运行环境符合依赖说明 |
| 网络错误 | 连接超时或不可达 | 执行ping命令测试网络连通性,检查防火墙和代理设置连接后执行ping命令测试网络连通性,检查防火墙和代理设置连接后重新执行命令,参考国内替代方案 |
输入:用户提供操作指令和必要参数
输出:返回执行结果,包含操作状态和输出数据
用户: 执行核心功能
Skill: 正在执行核心功能...
Skill: 执行完成,结果如下: 操作成功