概述

scripts/crontab_tool.py是一个 Python 脚本，用于管理系统 crontab 定时任务。支持添加、删除、列出任务，可作为公共方法复用，适用于各种定时场景（基金日报、数据备份、提醒推送等）。

为什么使用系统 crontab 而非 OpenClaw cron

对于纯脚本执行（如运行 Python 脚本发送报告），系统 crontab 是更经济、更稳定的选择。

使用方法

添加任务

python3 scripts/crontab_tool.pyadd\--expr"30 20 * * 1-5"\--command"cd /path && python3 xxx.py"\--name"任务名称"\--log-dir /path/to/logs

参数说明：

示例：

# 添加基金日报（指数基金实时估值）python3 scripts/crontab_tool.pyadd\--expr"5 15 * * 1-5"\--command"cd /root/OpenClawWorkspace && python3 scripts/fund_report.py --type index"\--name"指数基金实时估值"\--log-dir /root/OpenClawWorkspace/logs# 添加基金日报（主动基金收盘净值）python3 scripts/crontab_tool.pyadd\--expr"30 20 * * 1-5"\--command"cd /root/OpenClawWorkspace && python3 scripts/fund_report.py --type active"\--name"主动基金收盘净值"\--log-dir /root/OpenClawWorkspace/logs

列出所有任务

python3 scripts/crontab_tool.py list

输出示例：

📋 共 2 条 crontab 任务： ============================================================ 2. 5 15 * * 1-5 cd /root/OpenClawWorkspace && python3 scripts/fund_report.py --type index >> /root/OpenClawWorkspace/logs/指数基金实时估值.log 2>&1 [#指数基金实时估值] 4. 30 20 * * 1-5 cd /root/OpenClawWorkspace && python3 scripts/fund_report.py --type active >> /root/OpenClawWorkspace/logs/主动基金收盘净值.log 2>&1 [#主动基金收盘净值] ============================================================

删除任务

按名称删除（推荐）：

python3 scripts/crontab_tool.py remove--name"指数基金实时估值"

按行号删除：

python3 scripts/crontab_tool.py remove--line3

查看原始 crontab

crontab-l

任务命名约定

每个任务在 crontab 中以注释行 + 命令的形式存储：

# crontab_tool: 指数基金实时估值 5 15 * * 1-5 cd /root/OpenClawWorkspace && python3 scripts/fund_report.py --type index >> /root/OpenClawWorkspace/logs/指数基金实时估值.log 2>&1

注释行# crontab_tool: {name}用于唯一标识任务，方便后续查找和删除。

作为公共方法调用

脚本中的核心函数可以被其他 Python 脚本 import 复用：

fromscripts.crontab_toolimportadd_task,remove_task,list_tasks# 添加任务result=add_task(expr="0 8 * * 1-5",command="cd /path && python3 script.py",name="每日早间提醒",log_dir="/path/to/logs")# 列出任务tasks=list_tasks()fortintasks:print(f"{t['line']}:{t['content']}")# 删除任务remove_task(name="每日早间提醒")

当前已配置的任务

cron 表达式参考

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

常用表达式：

相关文件

#!/usr/bin/env python3""" crontab_tool.py — 系统 crontab 任务管理工具 通过 Python 管理 Linux 系统 crontab 任务，可添加、删除、列出任务。 后期可复用为公共方法，用于添加不同的定时任务（如基金日报、天气提醒、数据备份等）。 使用方法: # 添加任务 python3 crontab_tool.py add --expr "30 20 * * 1-5" --command "cd /root/OpenClawWorkspace && python3 scripts/fund_report.py --type active" --name "主动基金收盘净值" # 添加任务（带日志路径） python3 crontab_tool.py add --expr "5 15 * * 1-5" --command "cd /root/OpenClawWorkspace && python3 scripts/fund_report.py --type index" --name "指数基金实时估值" --log-dir /root/OpenClawWorkspace/logs # 列出所有任务 python3 crontab_tool.py list # 删除任务（按名称匹配） python3 crontab_tool.py remove --name "指数基金实时估值" # 删除任务（按行号） python3 crontab_tool.py remove --line 3 """importsubprocessimportsysimportargparseimportreimportos# ============================================================# 公共方法# ============================================================defget_crontab():"""获取当前用户的 crontab 内容，返回行列表"""result=subprocess.run(["crontab","-l"],capture_output=True,text=True,timeout=10)ifresult.returncode==0:returnresult.stdout.rstrip("\n").split("\n")ifresult.stdout.strip()else[]# crontab -l 可能返回 1（无 crontab 时）if"no crontab"inresult.stderr.lower():return[]raiseRuntimeError(f"获取 crontab 失败:{result.stderr}")defset_crontab(lines):"""将行列表写入当前用户的 crontab"""content="\n".join(lines)+"\n"iflineselse""proc=subprocess.run(["crontab","-"],input=content,capture_output=True,text=True,timeout=10)ifproc.returncode!=0:raiseRuntimeError(f"写入 crontab 失败:{proc.stderr}")returnTruedefadd_task(expr,command,name=None,log_dir=None,comment=True):""" 添加一条 crontab 任务 参数: expr: cron 表达式，如 "30 20 * * 1-5" command: 要执行的命令（字符串） name: 任务名称，用于后续查找和删除（可选，不传则自动生成） log_dir: 日志目录（可选），传入后自动追加 stdout/stderr 重定向 comment: 是否在任务上方添加注释行（默认 True） 返回: dict: {"success": True, "line": 行号, "name": 名称} """lines=get_crontab()# 自动生成名称（取命令的前 30 个字符）ifnotname:name=command[:30].strip()# 构建任务行iflog_dir:os.makedirs(log_dir,exist_ok=True)safe_name=re.sub(r'[^a-zA-Z0-9_\-\u4e00-\u9fff]','_',name)log_file=os.path.join(log_dir,f"{safe_name}.log")task_line=f"{expr}{command}>>{log_file}2>&1"else:task_line=f"{expr}{command}"# 检查是否已存在同名任务（通过注释行识别）name_pattern=f"# crontab_tool:{name}"fori,lineinenumerate(lines):ifline.strip()==name_pattern:return{"success":False,"error":f"任务 '{name}' 已存在（第{i+1}行），请先删除或使用不同的名称"}# 添加任务（注释 + 命令）new_lines=[]ifcomment:new_lines.append(f"# crontab_tool:{name}")new_lines.append(task_line)lines.extend(new_lines)set_crontab(lines)return{"success":True,"line":len(lines)-len(new_lines)+1,"name":name,"expr":expr,"command":command,"log_dir":log_dir}defremove_task(name=None,line=None):""" 删除 crontab 任务 参数: name: 按任务名称删除（匹配注释行 # crontab_tool: {name} 及下一行命令） line: 按行号删除（1-indexed，删除该行） 返回: dict: {"success": True, "removed": 行数} """lines=get_crontab()ifnotlines:return{"success":False,"error":"当前没有 crontab 任务"}removed=0ifnameisnotNone:# 找到注释行并删除注释行 + 下一行命令new_lines=[]skip_next=Falsefori,lineinenumerate(lines):ifskip_next:skip_next=Falseremoved+=1continueifline.strip()==f"# crontab_tool:{name}":skip_next=True# 跳过下一行（命令行）removed+=1continuenew_lines.append(line)ifremoved==0:return{"success":False,"error":f"未找到名为 '{name}' 的任务"}set_crontab(new_lines)return{"success":True,"removed":removed}eliflineisnotNone:ifline<1orline>len(lines):return{"success":False,"error":f"行号{line}超出范围（1-{len(lines)}）"}removed_line=lines.pop(line-1)set_crontab(lines)return{"success":True,"removed":1,"content":removed_line}return{"success":False,"error":"请指定 --name 或 --line"}deflist_tasks():""" 列出当前所有 crontab 任务，解析注释标记 返回: list[dict]: 每个任务包含 line, content, name(若有标记) """lines=get_crontab()tasks=[]current_name=Nonefori,lineinenumerate(lines):line_num=i+1stripped=line.strip()ifstripped.startswith("# crontab_tool:"):current_name=stripped.split("# crontab_tool:",1)[1].strip()elifstripped.startswith("#"):current_name=Noneelifstrippedandnotstripped.startswith("#"):tasks.append({"line":line_num,"content":stripped,"name":current_name})current_name=Nonereturntasks# ============================================================# CLI 入口# ============================================================defmain():parser=argparse.ArgumentParser(description="系统 crontab 任务管理工具",formatter_class=argparse.RawDescriptionHelpFormatter,epilog=""" 使用示例: # 添加基金日报任务 %(prog)s add --expr "5 15 * * 1-5" --command "cd /root/OpenClawWorkspace && python3 scripts/fund_report.py --type index" --name "指数基金实时估值" --log-dir /root/OpenClawWorkspace/logs %(prog)s add --expr "30 20 * * 1-5" --command "cd /root/OpenClawWorkspace && python3 scripts/fund_report.py --type active" --name "主动基金收盘净值" --log-dir /root/OpenClawWorkspace/logs # 列出所有任务 %(prog)s list # 删除任务 %(prog)s remove --name "指数基金实时估值" %(prog)s remove --line 3 """)subparsers=parser.add_subparsers(dest="action",required=True)# addadd_parser=subparsers.add_parser("add",help="添加 crontab 任务")add_parser.add_argument("--expr",required=True,help="cron 表达式，如 '30 20 * * 1-5'")add_parser.add_argument("--command",required=True,help="要执行的命令")add_parser.add_argument("--name",help="任务名称（用于后续查找和删除）")add_parser.add_argument("--log-dir",help="日志目录，传入后自动追加 stdout/stderr 重定向")add_parser.add_argument("--no-comment",action="store_true",help="不在任务上方添加注释行")# removeremove_parser=subparsers.add_parser("remove",help="删除 crontab 任务")remove_parser.add_argument("--name",help="按任务名称删除")remove_parser.add_argument("--line",type=int,help="按行号删除")# listsubparsers.add_parser("list",help="列出所有 crontab 任务")args=parser.parse_args()ifargs.action=="add":result=add_task(expr=args.expr,command=args.command,name=args.name,log_dir=args.log_dir,comment=notargs.no_comment)ifresult["success"]:print(f"✅ 任务添加成功")print(f" 名称:{result['name']}")print(f" 表达式:{result['expr']}")print(f" 命令:{result['command']}")ifresult.get("log_dir"):print(f" 日志:{result['log_dir']}/")print(f" 行号: 第{result['line']}行")else:print(f"❌ 添加失败:{result['error']}")sys.exit(1)elifargs.action=="remove":result=remove_task(name=args.name,line=args.line)ifresult["success"]:print(f"✅ 删除成功，已移除{result['removed']}行")ifresult.get("content"):print(f" 内容:{result['content']}")else:print(f"❌ 删除失败:{result['error']}")sys.exit(1)elifargs.action=="list":tasks=list_tasks()ifnottasks:print("📭 当前没有 crontab 任务")returnprint(f"📋 共{len(tasks)}条 crontab 任务：")print("="*60)fortintasks:tag=f" [#{t['name']}]"ift["name"]else""print(f"{t['line']:>3}.{t['content']}{tag}")print("="*60)if__name__=="__main__":main()