Gemini Vision APIで壁紙アプリの画像を自動カテゴリ分類する — 3000枚を30カテゴリに処理した実践記録

import google.generativeai as genai
import json
from pathlib import Path
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.0-flash")
 
# アプリ独自の30カテゴリを定義
CATEGORIES = [
    "自然風景", "ミニマリスト", "ダークテーマ", "グラデーション",
    "抽象アート", "都市景観", "サンセット", "オーシャン",
    "フォレスト", "スペース・宇宙", "フラワー・植物", "アニマル",
    "テクスチャ", "パターン", "タイポグラフィ", "建築",
    "食べ物", "スポーツ", "テクノロジー", "ゲーム",
    "アニメ・イラスト", "水彩・アート", "モノクロ", "ヴィンテージ",
    "ネオン・サイバーパンク", "和風・日本", "北欧スタイル", "ローファイ",
    "クリスマス・季節", "その他"
]
 
def classify_wallpaper(image_path: str) -> dict:
    """壁紙画像を分類して結果を返す"""
    
    image_file = genai.upload_file(image_path)
    
    prompt = f"""以下の壁紙画像を分析し、最も適切なカテゴリを選んでください。
 
カテゴリ一覧:
{json.dumps(CATEGORIES, ensure_ascii=False)}
 
以下のJSON形式で回答してください:
{{
  "primary_category": "メインカテゴリ名",
  "secondary_category": "サブカテゴリ名（オプション）",
  "confidence": 0.0〜1.0の数値,
  "description_ja": "この壁紙の雰囲気を30文字以内で",
  "description_en": "30-char atmosphere description in English",
  "tags": ["タグ1", "タグ2", "タグ3"]
}}
 
JSONのみ返してください。"""
 
    response = model.generate_content([image_file, prompt])
    
    try:
        result = json.loads(response.text.strip())
        return result
    except json.JSONDecodeError:
        # JSON解析失敗の場合は手動確認キューへ
        return {"primary_category": "その他", "confidence": 0.0, "error": response.text}

import time
import asyncio
from pathlib import Path
 
async def classify_batch(image_dir: str, output_file: str):
    """ディレクトリ内の全画像を分類してJSONに保存"""
    
    images = list(Path(image_dir).glob("*.jpg")) + list(Path(image_dir).glob("*.png"))
    results = []
    manual_review_queue = []
    
    print(f"処理対象: {len(images)}枚")
    
    for i, img_path in enumerate(images):
        result = classify_wallpaper(str(img_path))
        result["filename"] = img_path.name
        results.append(result)
        
        # 信頼度0.7未満は手動確認へ
        if result.get("confidence", 0) < 0.7:
            manual_review_queue.append(img_path.name)
        
        # 進捗表示
        if (i + 1) % 50 == 0:
            print(f"処理済み: {i+1}/{len(images)} ({len(manual_review_queue)}件が手動確認待ち)")
        
        # レート制限対策: 0.5秒待機
        time.sleep(0.5)
    
    # 結果を保存
    with open(output_file, "w", encoding="utf-8") as f:
        json.dump({
            "total": len(images),
            "manual_review_count": len(manual_review_queue),
            "manual_review_queue": manual_review_queue,
            "results": results
        }, f, ensure_ascii=False, indent=2)
    
    print(f"完了。手動確認が必要: {len(manual_review_queue)}枚")