mcpField
B
batch-processor

batch-processor

github.com/claude-office-skills文档处理5454.8k 次安装39 周前更新
代码仓库

关于 batch-processor

batch-processor 是用于批量处理文档的技能，可并行转换、整理、提取或分析文件，并跟踪处理进度。适合处理大量文件的文档工作流。

主要功能

  1. 01并行处理多个文件
  2. 02转换和整理文档
  3. 03提取或分析文件内容
  4. 04跟踪进度并保存检查点
  5. 05记录失败以便后续查看

适用场景

  1. 01批量将 PDF 转换为 Word 文档
  2. 02从文件夹中的图像提取文本
  3. 03批量重命名并整理文件
  4. 04批量更新文档页眉和页脚

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/claude-office-skills/skills --skill batch-processor

安装后在 Claude Code 中输入/batch-processor确认已加载。

详细介绍

批量处理技能

概述

此技能可高效批量处理文档——通过并行执行和进度跟踪，转换、转变、提取或分析数百个文件。

使用方法

  1. 描述你想完成的任务
  2. 提供所需的输入数据或文件
  3. 我会执行相应操作

示例提示：

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

领域知识

批处理模式
Input: [file1, file2, ..., fileN]
         │
         ▼
    ┌─────────────┐
    │  Parallel   │  ← Process multiple files concurrently
    │  Workers    │
    └─────────────┘
         │
         ▼
Output: [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:
    """Process a single file."""
    # Your processing logic here
    return {"path": str(file_path), "status": "success"}

def batch_process(input_dir: str, pattern: str = "*.*", max_workers: int = 4):
    """Process all matching files in directory."""
    
    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

# Usage
results = batch_process("/documents/invoices", "*.pdf", max_workers=8)
print(f"Processed {len(results)} files")
错误处理与恢复
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  # Skip already processed
            
            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()  # Resume-safe

最佳实践

  1. 使用进度条（tqdm）向用户反馈进度
  2. 为长时间运行的任务实现检查点机制
  3. 设置合理的工作线程数量（CPU 核心数）
  4. 记录失败情况以供后续检查

安装

# Install required dependencies
pip install python-docx openpyxl python-pptx reportlab jinja2

资源

评论

支持 Markdown