任务自动化

将重复性工作转化为自动化工作流，节省时间、减少错误、提升运营规模。

自动化类型

1. 文件操作

批量重命名：

import os
import re

def batch_rename(directory, pattern, replacement):
    """重命名匹配正则模式的文件"""
    for filename in os.listdir(directory):
        if re.match(pattern, filename):
            new_name = re.sub(pattern, replacement, filename)
            os.rename(
                os.path.join(directory, filename),
                os.path.join(directory, new_name)
            )
            print(f"已重命名: {filename} -> {new_name}")

批量格式转换：

from PIL import Image
import os

def convert_images(input_dir, output_dir, format='webp'):
    """将所有图像转换为指定格式"""
    os.makedirs(output_dir, exist_ok=True)
    
    for filename in os.listdir(input_dir):
        if filename.lower().endswith(('.png', '.jpg', '.jpeg')):
            img = Image.open(os.path.join(input_dir, filename))
            name = os.path.splitext(filename)[0]
            img.save(os.path.join(output_dir, f"{name}.{format}"), format.upper())
            print(f"已转换: {filename}")

文件归类整理：

import os
import shutil

def organize_by_type(directory):
    """按文件类型将文件移入对应文件夹"""
    extensions = {
        'images': ['.jpg', '.jpeg', '.png', '.gif', '.webp'],
        'documents': ['.pdf', '.doc', '.docx', '.txt', '.md'],
        'videos': ['.mp4', '.mov', '.avi', '.mkv'],
        'audio': ['.mp3', '.wav', '.flac'],
        'code': ['.py', '.js', '.ts', '.go', '.rs'],
    }
    
    for filename in os.listdir(directory):
        filepath = os.path.join(directory, filename)
        if os.path.isfile(filepath):
            ext = os.path.splitext(filename)[1].lower()
            for folder, exts in extensions.items():
                if ext in exts:
                    target = os.path.join(directory, folder)
                    os.makedirs(target, exist_ok=True)
                    shutil.move(filepath, os.path.join(target, filename))
                    print(f"已将 {filename} 移至 {folder}/")
                    break

2. 数据处理

批量数据转换：

import pandas as pd

def process_csv_batch(input_dir, output_file, transform_func):
    """批量处理多个 CSV 文件并合并输出"""
    dfs = []
    
    for filename in os.listdir(input_dir):
        if filename.endswith('.csv'):
            df = pd.read_csv(os.path.join(input_dir, filename))
            df = transform_func(df)
            dfs.append(df)
    
    combined = pd.concat(dfs, ignore_index=True)
    combined.to_csv(output_file, index=False)
    print(f"已将 {len(dfs)} 个文件处理并合并为 {output_file}")

数据流水线：

def create_pipeline(steps):
    """创建可复用的数据流水线"""
    def pipeline(data):
        result = data
        for step in steps:
            result = step(result)
        return result
    return pipeline

# 示例用法：
pipeline = create_pipeline([
    lambda x: x.dropna(),
    lambda x: x.drop_duplicates(),
    lambda x: x[x['value'] > 0],
    lambda x: x.sort_values('date')
])

clean_data = pipeline(raw_data)

3. API 操作

限速 API 客户端：

import time
from functools import wraps

def rate_limit(calls_per_second=2):
    """用于限制 API 调用频率的装饰器"""
    min_interval = 1.0 / calls_per_second
    last_call = [0.0]
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_call[0]
            if elapsed < min_interval:
                time.sleep(min_interval - elapsed)
            last_call[0] = time.time()
            return func(*args, **kwargs)
        return wrapper
    return decorator

@rate_limit(2)  # 每秒最多调用 2 次
def api_call(endpoint, data):
    return requests.post(endpoint, json=data)

批量 API 调用：

def batch_api_calls(items, endpoint, batch_size=100):
    """以批次方式执行 API 调用"""
    results = []
    
    for i in range(0, len(items), batch_size):
        batch = items[i:i + batch_size]
        
        # 执行批次请求
        response = requests.post(endpoint, json={'items': batch})
        
        if response.status_code == 200:
            results.extend(response.json())
        else:
            print(f"第 {i//batch_size} 批次失败: {response.status_code}")
        
        time.sleep(1)  # 实施速率限制
    
    return results

指数退避重试：

import time
import random

def retry_with_backoff(func, max_retries=3, base_delay=1):
    """对失败调用采用指数退避策略进行重试"""
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            
            delay = base_delay * (2 ** attempt) + random.random()
            print(f"第 {attempt + 1}/{max_retries} 次重试，{delay:.1f} 秒后执行: {e}")
            time.sleep(delay)

4. 定时任务

Cron 作业：

# 每小时执行一次
0 * * * * /path/to/script.sh

# 每天上午 9 点执行
0 9 * * * /path/to/script.sh

# 每周一上午 9 点执行
0 9 * * 1 /path/to/script.sh

# 工作日（周一至周五）每小时执行一次
0 * * * 1-5 /path/to/script.sh

Python 调度器：

import schedule
import time

def job():
    print("正在运行定时任务...")

schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("09:00").do(job)
schedule.every().monday.do(job)

while True:
    schedule.run_pending()
    time.sleep(60)

OpenClaw Cron：

openclaw cron add 
  --name "Daily Report" 
  --schedule "0 9 * * *" 
  --task "Generate daily report and send to slack"

工作流模式

顺序流水线

def sequential_workflow(steps):
    """按顺序执行各步骤"""
    results = []
    
    for i, step in enumerate(steps):
        try:
            result = step['action'](**step.get('params', {}))
            results.append({'step': i, 'status': 'success', 'result': result})
        except Exception as e:
            results.append({'step': i, 'status': 'error', 'error': str(e)})
            if step.get('stop_on_error', True):
                break
    
    return results

并行执行

from concurrent.futures import ThreadPoolExecutor, as_completed

def parallel_workflow(tasks, max_workers=5):
    """并行执行任务"""
    results = []
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(task['action'], **task.get('params', {})): task 
                   for task in tasks}
        
        for future in as_completed(futures):
            task = futures[future]
            try:
                result = future.result()
                results.append({'task': task['name'], 'status': 'success', 'result': result})
            except Exception as e:
                results.append({'task': task['name'], 'status': 'error', 'error': str(e)})
    
    return results

条件工作流

def conditional_workflow(steps):
    """根据条件决定是否执行步骤"""
    context = {}
    
    for step in steps:
        # 检查条件
        if 'condition' in step:
            if not step['condition'](context):
                print(f"跳过 {step['name']}：条件不满足")
                continue
        
        # 执行步骤
        result = step['action'](**step.get('params', {}), context=context)
        context[step['name']] = result
    
    return context

错误处理

优雅降级

def robust_operation(data, fallback=None):
    """尝试主操作，失败时启用备用方案"""
    try:
        return primary_operation(data)
    except SpecificError as e:
        print(f"主操作失败: {e}，尝试备用方案")
        return fallback_operation(data) if fallback else None
    except Exception as e:
        print(f"所有方案均失败: {e}")
        return fallback

错误通知

def notify_on_error(func, notify_func):
    """出错时触发通知的装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            notify_func(f"函数 {func.__name__} 执行出错: {e}")
            raise
    return wrapper

@notify_on_error(send_slack_message)
def important_operation():
    # ...

监控

进度追踪

from tqdm import tqdm

def process_with_progress(items):
    """带进度条处理数据项"""
    results = []
    for item in tqdm(items, desc="Processing"):
        results.append(process(item))
    return results

日志记录

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    filename='automation.log'
)

logger = logging.getLogger(__name__)

def logged_operation(data):
    logger.info(f"开始执行操作，输入数据共 {len(data)} 条")
    try:
        result = process(data)
        logger.info(f"操作完成，共生成 {len(result)} 条结果")
        return result
    except Exception as e:
        logger.error(f"操作失败: {e}")
        raise

最佳实践

1. 从简单开始

手动 → 脚本 → 定时执行 → 监控告警

避免过度设计。先从手动流程入手，再逐步实现自动化。

2. 保证幂等性

def safe_operation(data):
    """可安全多次执行的操作"""
    # 检查是否已处理
    if already_processed(data):
        return get_cached_result(data)
    
    # 执行处理
    result = process(data)
    
    # 标记为已完成
    mark_processed(data)
    
    return result

3. 添加检查点

def long_running_workflow(data):
    """在关键节点保存执行进度"""
    checkpoint_file = "workflow_checkpoint.json"
    
    # 若检查点存在，则加载
    if os.path.exists(checkpoint_file):
        with open(checkpoint_file) as f:
            state = json.load(f)
            start_from = state['step']
            data = state['data']
    else:
        start_from = 0
    
    # 带检查点执行流程
    for i, step in enumerate(steps[start_from:], start=start_from):
        result = step(data)
        
        # 保存检查点
        with open(checkpoint_file, 'w') as f:
            json.dump({'step': i + 1, 'data': result}, f)
    
    # 清理检查点文件
    os.remove(checkpoint_file)
    return result

4. 充分测试

def test_automation():
    """使用模拟数据测试自动化逻辑"""
    test_data = create_mock_data()
    
    # 预演执行（dry run）
    result = automation(test_data, dry_run=True)
    
    # 验证结果
    assert result['status'] == 'success'
    assert len(result['output']) == expected_count
    
    print("所有测试通过！")

5. 全面文档化

def automated_task(config):
    """
    处理每日销售数据并生成报告。
    
    参数:
        config: 字典，包含以下键：
            - input_dir: 存放 CSV 文件的目录路径
            - output_file: 输出报告的文件路径
            - notify: 完成后发送通知的邮箱地址
    
    返回值:
        字典，包含以下键：
            - status: 'success' 或 'error'
            - records_processed: 已处理记录数
            - output_file: 生成报告的文件路径
    
    示例:
        result = automated_task({
            'input_dir': '/data/sales',
            'output_file': '/reports/daily.csv',
            'notify': 'team@company.com'
        })
    """
    # 实现逻辑...

常见应用场景

每日报告自动化

def daily_report():
    # 1. 获取数据
    data = fetch_from_sources()
    
    # 2. 数据处理
    processed = process_data(data)
    
    # 3. 生成报告
    report = generate_report(processed)
    
    # 4. 分发报告
    send_email(report)
    upload_to_slack(report)
    
    # 5. 存档报告
    archive_report(report)

数据同步

def sync_data():
    # 1. 获取上次同步状态
    last_sync = get_last_sync_time()
    
    # 2. 获取自上次同步以来的变更
    changes = fetch_changes_since(last_sync)
    
    # 3. 应用变更
    for change in changes:
        apply_change(change)
    
    # 4. 更新同步状态
    update_sync_time()

清理自动化

def cleanup():
    # 1. 删除旧文件
    remove_old_files(days=30)
    
    # 2. 清空临时目录
    clear_temp_dirs()
    
    # 3. 归档旧日志
    archive_logs()
    
    # 4. 优化数据库
    optimize_database()