crontab 管理工具(降低openclaw cron (agentTurn) token )

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

# 添加基金日报（指数基金实时估值）python3 scripts/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 scripts/crontab_tool.py add \  --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 --line 3

crontab -l

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

from scripts.crontab_tool import add_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()for t in tasks:    print(f"{t['line']}: {t['content']}")# 删除任务remove_task(name="每日早间提醒")

┌───────────── 分钟 (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"""import subprocessimport sysimport argparseimport reimport os# ============================================================# 公共方法# ============================================================def get_crontab():    """获取当前用户的 crontab 内容，返回行列表"""    result = subprocess.run(        ["crontab", "-l"],        capture_output=True,        text=True,        timeout=10    )    if result.returncode == 0:        return result.stdout.rstrip("\n").split("\n") if result.stdout.strip() else []    # crontab -l 可能返回 1（无 crontab 时）    if "no crontab" in result.stderr.lower():        return []    raise RuntimeError(f"获取 crontab 失败: {result.stderr}")def set_crontab(lines):    """将行列表写入当前用户的 crontab"""    content = "\n".join(lines) + "\n" if lines else ""    proc = subprocess.run(        ["crontab", "-"],        input=content,        capture_output=True,        text=True,        timeout=10    )    if proc.returncode != 0:        raise RuntimeError(f"写入 crontab 失败: {proc.stderr}")    return Truedef add_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 个字符）    if not name:        name = command[:30].strip()    # 构建任务行    if log_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}"    for i, line in enumerate(lines):        if line.strip() == name_pattern:            return {                "success": False,                "error": f"任务 '{name}' 已存在（第 {i+1} 行），请先删除或使用不同的名称"            }    # 添加任务（注释 + 命令）    new_lines = []    if comment:        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    }def remove_task(name=None, line=None):    """    删除 crontab 任务    参数:        name: 按任务名称删除（匹配注释行 # crontab_tool: {name} 及下一行命令）        line: 按行号删除（1-indexed，删除该行）    返回:        dict: {"success": True, "removed": 行数}    """    lines = get_crontab()    if not lines:        return {"success": False, "error": "当前没有 crontab 任务"}    removed = 0    if name is not None:        # 找到注释行并删除注释行 + 下一行命令        new_lines = []        skip_next = False        for i, line in enumerate(lines):            if skip_next:                skip_next = False                removed += 1                continue            if line.strip() == f"# crontab_tool: {name}":                skip_next = True  # 跳过下一行（命令行）                removed += 1                continue            new_lines.append(line)        if removed == 0:            return {"success": False, "error": f"未找到名为 '{name}' 的任务"}        set_crontab(new_lines)        return {"success": True, "removed": removed}    elif line is not None:        if line < 1 or line > 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"}def list_tasks():    """    列出当前所有 crontab 任务，解析注释标记    返回:        list[dict]: 每个任务包含 line, content, name(若有标记)    """    lines = get_crontab()    tasks = []    current_name = None    for i, line in enumerate(lines):        line_num = i + 1        stripped = line.strip()        if stripped.startswith("# crontab_tool:"):            current_name = stripped.split("# crontab_tool:", 1)[1].strip()        elif stripped.startswith("#"):            current_name = None        elif stripped and not stripped.startswith("#"):            tasks.append({                "line": line_num,                "content": stripped,                "name": current_name            })            current_name = None    return tasks# ============================================================# CLI 入口# ============================================================def main():    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)    # add    add_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="不在任务上方添加注释行")    # remove    remove_parser = subparsers.add_parser("remove", help="删除 crontab 任务")    remove_parser.add_argument("--name", help="按任务名称删除")    remove_parser.add_argument("--line", type=int, help="按行号删除")    # list    subparsers.add_parser("list", help="列出所有 crontab 任务")    args = parser.parse_args()    if args.action == "add":        result = add_task(            expr=args.expr,            command=args.command,            name=args.name,            log_dir=args.log_dir,            comment=not args.no_comment        )        if result["success"]:            print(f"✅ 任务添加成功")            print(f"   名称: {result['name']}")            print(f"   表达式: {result['expr']}")            print(f"   命令: {result['command']}")            if result.get("log_dir"):                print(f"   日志: {result['log_dir']}/")            print(f"   行号: 第 {result['line']} 行")        else:            print(f"❌ 添加失败: {result['error']}")            sys.exit(1)    elif args.action == "remove":        result = remove_task(name=args.name, line=args.line)        if result["success"]:            print(f"✅ 删除成功，已移除 {result['removed']} 行")            if result.get("content"):                print(f"   内容: {result['content']}")        else:            print(f"❌ 删除失败: {result['error']}")            sys.exit(1)    elif args.action == "list":        tasks = list_tasks()        if not tasks:            print("📭 当前没有 crontab 任务")            return        print(f"📋 共 {len(tasks)} 条 crontab 任务：")        print("=" * 60)        for t in tasks:            tag = f" [#{t['name']}]" if t["name"] else ""            print(f"  {t['line']:>3}. {t['content']}{tag}")        print("=" * 60)if __name__ == "__main__":    main()