batch-processor

批次處理技能

概述

此技能可高效批次處理文件——轉換、轉檔、擷取或分析數百個檔案，支援平行執行與進度追蹤。

使用方式

1. 描述你想完成的工作
2. 提供所需的輸入資料或檔案
3. 我將執行對應的操作

範例提示：

• 「將 100 個 PDF 轉換為 Word 文件」
• 「從資料夾中的所有圖片擷取文字」
• 「批次重新命名並整理檔案」
• 「大量更新文件的頁首/頁尾」

領域知識

批次處理模式

輸入: [file1, file2, ..., fileN]
         │
         ▼
    ┌─────────────┐
    │  平行工作   │  ← 同時處理多個檔案
    │  執行緒     │
    └─────────────┘
         │
         ▼
輸出: [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. 針對長時間任務實作檢查點機制
3. 設定合理的工作執行緒數量（CPU 核心數）
4. 記錄失敗項目以便後續檢視

安裝

# 安裝必要相依套件
pip install python-docx openpyxl python-pptx reportlab jinja2

資源

• 自訂儲存庫
• Claude Office Skills Hub