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

リソース

• カスタムリポジトリ
• Claude Office Skills Hub