任务自动化器（Task Automator）

概述

OpenClaw 的通用任务自动化能力。支持文件操作、数据处理、API 调用，并可创建带调度功能的自定义工作流。

典型应用场景

• 文件操作：批量重命名、格式转换、文件归类
• 数据处理：CSV/JSON/Excel 数据操作与清洗
• API 集成：连接多个服务，实现跨系统数据同步
• 定时任务：类 Cron 自动化，支持周期性执行
• 网页自动化：网页抓取、状态监控、异常告警
• 电商运营：订单处理、库存同步、价格更新

快速上手

运行一个简单任务

python scripts/run_task.py --task file_organizer --config tasks/organize.json

调度一个周期性任务

python scripts/schedule_task.py --task data_backup --cron "0 2 * * *"

创建并运行一个工作流

python scripts/run_workflow.py --workflow ecommerce_sync

内置任务

1. 文件归类器（File Organizer）

按文件类型、日期或自定义规则对文件进行归类整理。

配置文件（tasks/organize.json）：

{
  "source": "~/Downloads",
  "destination": "~/Organized",
  "rules": [
    {"extension": ".pdf", "folder": "Documents"},
    {"extension": ".jpg", "folder": "Images"},
    {"extension": ".mp4", "folder": "Videos"}
  ]
}

2. 数据格式转换器（Data Converter）

在 CSV、JSON、Excel 格式之间相互转换。

使用方式：

python scripts/convert_data.py --input data.csv --output data.json --format json

3. API 同步器（API Sync）

在两个 API 之间同步数据。

配置文件（tasks/api_sync.json）：

{
  "source": {
    "type": "api",
    "url": "https://api.source.com/data",
    "auth": "bearer_token"
  },
  "destination": {
    "type": "api",
    "url": "https://api.dest.com/items",
    "auth": "api_key"
  },
  "mapping": {
    "source_field": "dest_field"
  }
}

4. 网站监控器（Web Monitor）

监控指定网站状态，并在满足条件时触发告警。

配置文件（tasks/monitor.json）：

{
  "urls": [
    {"url": "https://example.com/product", "check": "price < 100"}
  ],
  "alert": {
    "type": "email",
    "to": "you@example.com"
  }
}

5. 电商订单处理器（E-commerce Order Processor）

处理来自淘宝 / 抖音店铺的订单。

配置文件（tasks/order_process.json）：

{
  "stores": ["taobao", "douyin"],
  "actions": [
    "fetch_new_orders",
    "update_inventory",
    "generate_shipping_labels",
    "send_confirmation_email"
  ]
}

脚本说明

run_task.py

执行单个自动化任务。

参数说明：

• --task —— 任务名称
• --config —— 任务配置文件路径
• --dry-run —— 模拟执行（不真实运行）
• --verbose —— 启用详细日志输出

schedule_task.py

调度周期性任务。

参数说明：

• --task —— 任务名称
• --cron —— Cron 表达式（例如："0 2 * * *"）
• --config —— 任务配置文件路径

run_workflow.py

执行多步骤工作流。

参数说明：

• --workflow —— 工作流名称
• --steps —— 仅执行指定步骤（以逗号分隔）
• --continue-on-error —— 出错时继续执行后续步骤

创建自定义任务

第一步：编写任务脚本

# scripts/tasks/my_task.py
from base_task import BaseTask

class MyTask(BaseTask):
    def run(self, config):
        # 在此处添加你的自动化逻辑
        self.log("Starting task...")
        
        # 执行处理逻辑
        result = self.process(config)
        
        # 返回执行结果
        return {"status": "success", "data": result}

第二步：定义配置结构

{
  "name": "my_task",
  "description": "What this task does",
  "config_schema": {
    "required": ["input_path", "output_path"],
    "properties": {
      "input_path": {"type": "string"},
      "output_path": {"type": "string"}
    }
  }
}

第三步：注册任务

将以下内容添加至 tasks/registry.json：

{
  "my_task": {
    "script": "tasks/my_task.py",
    "config": "tasks/my_task.json"
  }
}

工作流（Workflows）

工作流用于串联多个任务，形成有依赖关系的执行序列。

示例：电商每日同步工作流

name: ecommerce_daily_sync
steps:
  - task: fetch_orders
    stores: [taobao, douyin]
  - task: update_inventory
    sync_stores: true
  - task: generate_reports
    format: excel
  - task: send_summary
    channel: email

任务调度

使用 Cron 表达式定义调度规则：

与电商平台集成

淘宝 / 抖音店铺自动化

# 每日订单同步
python scripts/run_task.py --task order_sync --config tasks/taobao_sync.json

# 库存更新
python scripts/run_task.py --task inventory_update --config tasks/inventory.json

# 价格监控
python scripts/run_task.py --task price_monitor --config tasks/prices.json

最佳实践

1. 上线前务必使用 --dry-run 测试
2. 全面记录日志，便于问题排查
3. 优雅处理异常，建议加入重试机制
4. 敏感信息使用环境变量管理（如 API Key）
5. 合理规划调度时间，避免触发平台调用频率限制
6. 持续监控任务健康状态，通过状态检查及时发现失败任务

安全规范

• API 密钥等敏感信息必须存储于环境变量中
• 推荐使用 .env 文件管理配置（切勿提交至版本控制）
• 对所有输入参数进行校验
• 对文件路径做白名单过滤或规范化处理，防止目录遍历攻击
• 对 API 请求实施速率限制

故障排查

• 任务静默失败：检查 logs/ 目录下的日志文件
• API 请求被限频：在请求间增加延迟（如 time.sleep()）
• 文件未找到：确认路径为绝对路径，或相对于 workspace 的正确相对路径
• 权限拒绝：检查目标文件/目录的读写权限设置