App Storeのレビュー1,000件をClaude APIで分析したら、自分のアプリの本当の問題が見えてきた

import anthropic
import json
import os
from typing import Optional
 
client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
 
def analyze_reviews_batch(reviews: list[dict]) -> list[dict]:
    """
    App Storeレビューのリストを受け取り、
    感情・カテゴリ・改善提案の構造化データを返す
    """
    # 1回のAPIコールで最大20件をまとめて処理（コスト削減）
    reviews_text = "\n---\n".join([
        f"評価: {r['rating']}星\nタイトル: {r['title']}\n本文: {r['body']}"
        for r in reviews[:20]
    ])
 
    message = client.messages.create(
        model="claude-haiku-4-5-20251001",  # 分析はHaikuで十分なコスパ
        max_tokens=2048,
        messages=[
            {
                "role": "user",
                "content": f"""以下のApp Storeレビューを分析し、JSON配列で返してください。
各レビューに対して:
- sentiment: "positive" / "neutral" / "negative"
- category: "ui" / "bug" / "feature_request" / "performance" / "billing" / "other"
- improvement_hint: 具体的な改善点（あれば）、なければ null
- priority_score: 1〜5（5が最高優先度。具体的で再現性の高い問題ほど高くする）
 
レビュー:
{reviews_text}
 
JSON配列のみを返してください。説明文は不要です。"""
            }
        ]
    )
 
    try:
        results = json.loads(message.content[0].text)
        return results
    except json.JSONDecodeError:
        # パース失敗時は空のリストを返してスキップ
        return []
 
def aggregate_by_priority(analyzed_reviews: list[dict]) -> dict:
    """
    分析結果を優先度スコアで集計し、
    次のスプリントで対応すべき改善点を上位5件抽出する
    """
    # priority_score 4以上の improvement_hint を抽出
    high_priority = [
        r["improvement_hint"]
        for r in analyzed_reviews
        if r.get("priority_score", 0) >= 4
        and r.get("improvement_hint")
    ]
 
    # Claude Sonnet 4.6で優先度上位5件を要約・整理
    if not high_priority:
        return {"top_issues": [], "summary": "高優先度の改善点は見つかりませんでした"}
 
    issues_text = "\n".join(f"- {h}" for h in high_priority)
 
    summary_message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=512,
        messages=[
            {
                "role": "user",
                "content": f"""以下のApp Storeレビューから抽出した改善ヒントをまとめ、
重複を排除して上位5件に絞ってください。
開発者がすぐに対応できる形式で、箇条書きで出力してください。
 
改善ヒント一覧:
{issues_text}"""
            }
        ]
    )
 
    return {
        "top_issues": high_priority[:10],
        "summary": summary_message.content[0].text
    }

import time
 
def process_all_reviews(reviews: list[dict]) -> list[dict]:
    """全レビューをバッチで処理する"""
    all_results = []
    batch_size = 20
 
    for i in range(0, len(reviews), batch_size):
        batch = reviews[i:i + batch_size]
        results = analyze_reviews_batch(batch)
        all_results.extend(results)
 
        # レート制限を避けるため1秒待機
        # Haikuの場合、TPMが高いので通常は不要だが安全のため入れる
        if i + batch_size < len(reviews):
            time.sleep(1)
 
        print(f"処理済み: {min(i + batch_size, len(reviews))}/{len(reviews)}")
 
    return all_results

import requests
 
def fetch_reviews_rss(app_id: str, page: int = 1) -> list[dict]:
    """
    iTunes RSS APIを使ってApp Storeレビューを取得する。
    最大10ページ（500件）まで取得可能。
    """
    url = f"https://itunes.apple.com/jp/rss/customerreviews/page={page}/id={app_id}/sortby=mostrecent/json"
    response = requests.get(url)
 
    if response.status_code != 200:
        return []
 
    data = response.json()
    entries = data.get("feed", {}).get("entry", [])
 
    # 最初のentryはアプリ情報なのでスキップ
    return [
        {
            "rating": int(e["im:rating"]["label"]),
            "title": e["title"]["label"],
            "body": e["content"]["label"]
        }
        for e in entries[1:]  # index 0をスキップ
        if "im:rating" in e
    ]