task-automation

任务自动化

此技能可帮助 AI Agent 设计并实现重复任务和工作流的自动化。Agent 会识别适合自动化的手动流程，选择合适的自动化模式（脚本、文件监视器、cron 作业、CI/CD 触发器、API 轮询），编写实现并验证其能否正常工作。目标是消除繁琐事务——这类工作重复且需要手动完成，其工作量会随任务量线性增长——并以可靠、无需人工干预的自动化取而代之。

工作流

1. 分析任务： 了解用户想要自动化的内容，包括触发条件（什么会启动任务）、涉及的步骤、输入和输出，以及目前手动执行的频率。判断任务由事件驱动（由某项变化触发）还是由时间驱动（按计划运行）。

2. 选择自动化模式： 根据触发类型和环境选择合适的自动化方案。常见模式包括：用于一次性或顺序任务的 shell 脚本；用于响应文件变化的文件监视器（fswatch、inotifywait、chokidar）；用于定期计划任务的 cron 作业或 systemd 计时器；用于代码相关自动化的 CI/CD 流水线触发器；用于响应外部服务事件的 API 轮询或 webhook 监听器。

3. 设计实现方案： 详细规划自动化方案：定义输入和配置、错误处理策略（重试逻辑、告警、回退行为）、日志记录方式，以及所需的机密或凭证管理。考虑幂等性——自动化应能安全地多次运行，且不会产生副作用。

4. 编写自动化代码： 使用合适的工具和语言实现自动化。优先选择成熟且广泛受支持的工具：脚本使用 bash/Python，调度使用 crontab，CI 触发器使用 GitHub Actions 或 GitLab CI，事件监听器使用标准 webhook 框架。

5. 测试和验证： 先在安全环境中运行自动化。验证其能否正确处理正常情况，然后测试边界情况：空输入、网络故障、权限错误和并发执行。确认日志记录了足够的信息以便调试。

6. 部署和监控： 为自动化设置适当的权限，并将其部署到目标环境。设置监控或告警，以便及时发现故障。记录自动化的用途、配置，以及需要时如何将其禁用。

用法

描述要自动化的任务，包括触发条件、需要执行的操作，以及运行环境。Agent 会选择合适的模式并实现自动化。

Automate the following: whenever a new CSV file is added to the ~/data/incoming/
directory, validate the CSV headers, transform the data into JSON, and move the
result to ~/data/processed/. Log any files that fail validation to ~/data/errors/.

示例

示例 1：使用处理流水线的文件监视器

用户请求：

自动处理目录中的传入 CSV 文件。

实现：

#!/usr/bin/env python3
"""File watcher that processes incoming CSVs into JSON.

Usage: python csv_watcher.py
Watches: ~/data/incoming/ for new .csv files
Outputs: ~/data/processed/*.json and ~/data/errors/error.log
"""

import os
import csv
import json
import time
import logging
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

INCOMING = Path.home() / "data" / "incoming"
PROCESSED = Path.home() / "data" / "processed"
ERRORS = Path.home() / "data" / "errors"
REQUIRED_HEADERS = ["id", "name", "email", "amount"]

logging.basicConfig(
    filename=ERRORS / "error.log",
    level=logging.ERROR,
    format="%(asctime)s - %(message)s"
)

for d in [INCOMING, PROCESSED, ERRORS]:
    d.mkdir(parents=True, exist_ok=True)


class CSVHandler(FileSystemEventHandler):
    def on_created(self, event):
        if not event.src_path.endswith(".csv"):
            return
        filepath = Path(event.src_path)
        try:
            with open(filepath, "r") as f:
                reader = csv.DictReader(f)
                headers = reader.fieldnames or []
                missing = set(REQUIRED_HEADERS) - set(headers)
                if missing:
                    raise ValueError(f"Missing headers: {missing}")
                rows = list(reader)

            output = PROCESSED / filepath.with_suffix(".json").name
            with open(output, "w") as f:
                json.dump(rows, f, indent=2)

            filepath.unlink()  # Remove processed file
            print(f"Processed: {filepath.name} -> {output.name}")
        except Exception as e:
            logging.error(f"Failed to process {filepath.name}: {e}")
            filepath.rename(ERRORS / filepath.name)


if __name__ == "__main__":
    observer = Observer()
    observer.schedule(CSVHandler(), str(INCOMING), recursive=False)
    observer.start()
    print(f"Watching {INCOMING} for new CSV files...")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

示例 2：基于 cron 的 API 轮询并发送 Slack 通知

用户请求：

每隔 6 小时检查一次我们的 API 健康端点；如果有任何服务降级，就在 Slack 上发送通知。

实现：

通过 crontab -e 添加的 Cron 条目：

0 */6 * * * /usr/bin/python3 /opt/scripts/health_check.py >> /var/log/health_check.log 2>&1

脚本：

#!/usr/bin/env python3
"""Poll API health endpoint and alert Slack on degraded services.

Runs every 6 hours via cron. Exits 0 on success, 1 on alert sent, 2 on script error.
"""

import os
import json
import urllib.request

HEALTH_URL = "https://api.example.com/health"
SLACK_WEBHOOK = os.environ["SLACK_WEBHOOK_URL"]


def check_health():
    req = urllib.request.Request(HEALTH_URL, headers={"Accept": "application/json"})
    with urllib.request.urlopen(req, timeout=10) as resp:
        data = json.loads(resp.read())
    return data  # e.g., {"services": {"auth": "ok", "payments": "degraded", "db": "ok"}}


def send_slack_alert(degraded_services):
    service_list = "\n".join(f"- *{name}*: {status}" for name, status in degraded_services)
    payload = json.dumps({
        "text": f":warning: *Service Health Alert*\n{service_list}"
    }).encode()
    req = urllib.request.Request(
        SLACK_WEBHOOK,
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST"
    )
    urllib.request.urlopen(req)


if __name__ == "__main__":
    health = check_health()
    degraded = [
        (name, status)
        for name, status in health.get("services", {}).items()
        if status != "ok"
    ]
    if degraded:
        send_slack_alert(degraded)
        print(f"Alert sent for {len(degraded)} degraded service(s)")
        exit(1)
    else:
        print("All services healthy")
        exit(0)

最佳实践

• 确保自动化操作具有幂等性。 使用相同输入运行同一自动化操作两次，应产生相同结果且没有副作用。这样可以避免任务意外重新触发时造成数据损坏。
• 记录所有日志，有选择地发送警报。 编写详细日志以便调试，但只针对可采取行动的故障发送警报。收件箱里满是“全部正常”的通知，会让人逐渐忽视警报。
• 将配置外置。 将文件路径、URL、阈值和凭据存储在环境变量或配置文件中，不要在脚本中硬编码。这样可以让自动化操作更易移植，也更便于管理机密信息。
• 为计划任务使用锁文件或互斥锁。 如果上一次运行尚未结束，Cron 任务可能会重叠。使用 flock 或 PID 文件，确保同一时间只有一个实例运行。
• 对自动化脚本进行版本控制。 将自动化操作视为生产代码：存储在 Git 中、审查更改并标记发行版本。自动化操作出错造成的损害可能比功能出错更大。
• 提供手动干预方式。 每项自动化操作都应记录如何暂停、跳过或手动运行。在事故期间，这一点至关重要，因为自动化操作可能会干扰手动修复。

边缘情况

• 多步骤自动化中的部分失败： 如果 5 个步骤中的第 3 步失败，自动化操作不应悄悄跳过该步骤。实现检查点机制，让自动化操作可以从最后一个成功步骤恢复，而不是从头开始。
• 并发写入文件： 文件监视器可能会在文件尚未写入完成时触发。处理文件前，增加短暂延迟或检查文件是否稳定（大小连续 N 秒未变化）。
• 凭据过期： API 令牌和 OAuth 凭据会过期。为长期运行的自动化操作加入令牌刷新逻辑，并在刷新失败时发送警报，而不是悄无声息地停止运行。
• Cron 的时区问题： Cron 默认使用系统时区。对于全球团队，请明确使用 UTC，或记录所用时区。注意夏令时切换可能导致任务重复运行或跳过。
• 轮询 API 时的速率限制： 如果轮询间隔太短或运行多个实例，API 轮询可能触发速率限制。实现指数退避，并跟踪速率限制响应头。
• 输入为空或格式错误： 由外部数据（文件、webhook、API 响应）触发的自动化操作，应在处理前验证输入架构。遇到错误时应优雅失败并提供明确的错误消息，而不是生成损坏的输出。