壁紙アプリの一括カテゴリ分類を 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)
 
IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp"}
 
def collect_images(image_dir: str) -> list:
    """対象ディレクトリから画像ファイルだけを拾う"""
    return sorted(
        p for p in Path(image_dir).iterdir()
        if p.is_file() and p.suffix.lower() in IMAGE_EXTS
    )
 
def batch_classify(image_dir: str, output_file: str):
    """ディレクトリ内の全画像を分類してJSONで保存する"""
    
    image_files = collect_images(image_dir)
    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} に保存")

image_files = list(Path(image_dir).glob("*.{jpg,jpeg,png,webp}"))

from pathlib import Path
 
# a.jpg / b.png / c.webp / d.jpeg を置いたディレクトリで実行
p = Path(".")
print(p.glob("*.{jpg,jpeg,png,webp}"))          # → 空
print([x.name for x in p.iterdir()
       if x.suffix.lower() in {".jpg", ".jpeg", ".png", ".webp"}])
# → ['a.jpg', 'b.png', 'c.webp', 'd.jpeg']

image_files = collect_images(image_dir)
if not image_files:
    raise SystemExit(f"対象画像が見つかりません: {image_dir}")
print(f"対象: {len(image_files)}件")

# 対策：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()

def load_done(output_file: str) -> dict:
    """既存の結果を読み込む（無ければ空で始める）"""
    path = Path(output_file)
    if not path.exists():
        return {}
    with path.open(encoding="utf-8") as f:
        done = json.load(f)
    # エラーで終わった分はやり直す
    return {k: v for k, v in done.items() if "error" not in v}
 
def batch_classify_resumable(image_dir: str, output_file: str, save_every: int = 20):
    """途中から再開できるバッチ処理"""
    results = load_done(output_file)
    targets = [p for p in collect_images(image_dir) if p.name not in results]
 
    print(f"対象: {len(targets)}件（処理済みスキップ: {len(results)}件）")
 
    for i, img_path in enumerate(targets, 1):
        try:
            results[img_path.name] = classify_wallpaper(str(img_path))
        except Exception as e:
            print(f"  失敗: {img_path.name} — {e}")
            results[img_path.name] = {"category": "未分類", "error": str(e)}
 
        # 定期的に書き出す（最後にまとめて書くと落ちたとき全損する）
        if i % save_every == 0 or i == len(targets):
            with open(output_file, "w", encoding="utf-8") as f:
                json.dump(results, f, ensure_ascii=False, indent=2)
            print(f"  → {i}/{len(targets)} 保存")