壁紙アプリの一括カテゴリ分類を Claude Vision API で自動化した話 — 実運用で見えた精度と限界

import anthropic
import base64
import json
from pathlib import Path
 
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
 
# アプリで使用するカテゴリ定義（実際のカテゴリ体系に合わせる）
CATEGORIES = [
    "自然・風景", "都市・建築", "抽象・パターン",
    "動物・生き物", "花・植物", "海・水辺",
    "夜景・夕暮れ", "宇宙・天体", "季節・行事",
    "シンプル・ミニマル"
]
 
def classify_wallpaper(image_path: str) -> dict:
    """1枚の壁紙画像をカテゴリ分類する"""
    
    with open(image_path, "rb") as f:
        image_data = base64.standard_b64encode(f.read()).decode("utf-8")
    
    # 拡張子から media_type を判定
    ext = Path(image_path).suffix.lower()
    media_type_map = {".jpg": "image/jpeg", ".jpeg": "image/jpeg",
                      ".png": "image/png", ".webp": "image/webp"}
    media_type = media_type_map.get(ext, "image/jpeg")
    
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",  # コスト重視でHaikuを使用
        max_tokens=256,
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": media_type,
                        "data": image_data
                    }
                },
                {
                    "type": "text",
                    "text": f"""この画像は壁紙アプリ用の画像です。
以下のカテゴリから最も適切なものを1つ選び、JSON形式で回答してください。
 
カテゴリ: {json.dumps(CATEGORIES, ensure_ascii=False)}
 
回答形式:
{{"category": "カテゴリ名", "confidence": 0.0〜1.0, "reason": "選んだ理由を1文で"}}
 
他の形式での回答は不要です。"""
                }
            ]
        }]
    )
    
    result_text = response.content[0].text.strip()
    return json.loads(result_text)
 
def batch_classify(image_dir: str, output_file: str):
    """ディレクトリ内の全画像を分類してJSONで保存する"""
    
    image_files = list(Path(image_dir).glob("*.{jpg,jpeg,png,webp}"))
    results = {}
    
    for i, img_path in enumerate(image_files):
        print(f"[{i+1}/{len(image_files)}] {img_path.name}")
        try:
            result = classify_wallpaper(str(img_path))
            results[img_path.name] = result
        except Exception as e:
            print(f"  ❌ エラー: {e}")
            results[img_path.name] = {"category": "未分類", "error": str(e)}
    
    with open(output_file, "w", encoding="utf-8") as f:
        json.dump(results, f, ensure_ascii=False, indent=2)
    
    print(f"\n✅ 完了: {len(results)}件を {output_file} に保存")

# 対策：JSON部分だけを抽出する
import re
 
def safe_parse_json(text: str) -> dict:
    """レスポンスからJSON部分を安全に抽出する"""
    # コードブロック内のJSONを試みる
    code_match = re.search(r'```(?:json)?\n(.*?)\n```', text, re.DOTALL)
    if code_match:
        return json.loads(code_match.group(1))
    
    # 通常のJSONオブジェクトを試みる
    json_match = re.search(r'\{.*\}', text, re.DOTALL)
    if json_match:
        return json.loads(json_match.group())
    
    raise ValueError(f"JSONが見つかりません: {text[:100]}")

import time
 
def batch_classify_with_rate_limit(image_files: list, max_per_minute: int = 40):
    """レート制限を考慮したバッチ処理"""
    results = {}
    interval = 60.0 / max_per_minute  # リクエスト間隔（秒）
    
    for i, img_path in enumerate(image_files):
        start = time.time()
        
        result = classify_wallpaper(str(img_path))
        results[str(img_path)] = result
        
        elapsed = time.time() - start
        sleep_time = max(0, interval - elapsed)
        if sleep_time > 0:
            time.sleep(sleep_time)
        
        if (i + 1) % 50 == 0:
            print(f"  → {i+1}件処理済み")
    
    return results

from PIL import Image
import io
 
def resize_for_classification(image_path: str, max_size: int = 512) -> bytes:
    """分類用に画像をリサイズする（アスペクト比保持）"""
    with Image.open(image_path) as img:
        img.thumbnail((max_size, max_size), Image.LANCZOS)
        buffer = io.BytesIO()
        img.save(buffer, format="JPEG", quality=85)
        return buffer.getvalue()