批量处理器技能

概述

该技能支持高效批量处理文档——可并行执行并跟踪进度，一次性完成数百个文件的格式转换、内容变换、信息抽取或分析任务。

使用方法

1. 描述您希望达成的目标
2. 提供所需输入数据或文件
3. 我将自动执行相应操作

示例提示语：

• “将 100 个 PDF 文件批量转换为 Word 文档”
• “提取指定文件夹中所有图像的文本内容”
• “批量重命名并整理文件”
• “批量更新文档页眉/页脚”

领域知识

批量处理模式

输入：[file1, file2, ..., fileN]
         │
         ▼
    ┌─────────────┐
    │  并行工作器 │  ← 并发处理多个文件
    │  Parallel   │
    │  Workers    │
    └─────────────┘
         │
         ▼
输出：[result1, result2, ..., resultN]

Python 实现

from concurrent.futures import ProcessPoolExecutor, as_completed
from pathlib import Path
from tqdm import tqdm

def process_file(file_path: Path) -> dict:
    """处理单个文件。"""
    # 此处填写您的具体处理逻辑
    return {"path": str(file_path), "status": "success"}

def batch_process(input_dir: str, pattern: str = "*.*", max_workers: int = 4):
    """处理目录中所有匹配的文件。"""
    
    files = list(Path(input_dir).glob(pattern))
    results = []
    
    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(process_file, f): f for f in files}
        
        for future in tqdm(as_completed(futures), total=len(files)):
            file = futures[future]
            try:
                result = future.result()
                results.append(result)
            except Exception as e:
                results.append({"path": str(file), "error": str(e)})
    
    return results

# 使用示例
results = batch_process("/documents/invoices", "*.pdf", max_workers=8)
print(f"已处理 {len(results)} 个文件")

错误处理与断点续传

import json
from pathlib import Path

class BatchProcessor:
    def __init__(self, checkpoint_file: str = "checkpoint.json"):
        self.checkpoint_file = checkpoint_file
        self.processed = self._load_checkpoint()
    
    def _load_checkpoint(self):
        if Path(self.checkpoint_file).exists():
            return json.load(open(self.checkpoint_file))
        return {}
    
    def _save_checkpoint(self):
        json.dump(self.processed, open(self.checkpoint_file, "w"))
    
    def process(self, files: list, processor_func):
        for file in files:
            if str(file) in self.processed:
                continue  # 跳过已处理文件
            
            try:
                result = processor_func(file)
                self.processed[str(file)] = {"status": "success", **result}
            except Exception as e:
                self.processed[str(file)] = {"status": "error", "error": str(e)}
            
            self._save_checkpoint()  # 支持断点续传

最佳实践

1. 使用进度条（tqdm）向用户提供实时反馈
2. 为长时间运行的任务实现断点检查（checkpointing）
3. 合理设置工作线程数（通常不超过 CPU 核心数）
4. 记录失败详情，便于后续审查（review）

安装

# 安装所需依赖
pip install python-docx openpyxl python-pptx reportlab jinja2

资源

• 自定义仓库（Custom Repository）
• Claude Office Skills Hub