Gemini API × App Store Connect API でレビュー対応を自動化する：5,000万DL運用の経験から書いたASOインテリジェンス実装メモ

def get_app_ids(client: AppStoreConnectClient) -> list[dict]:
    """運営中のアプリ一覧を取得"""
    data = client.get("/apps", params={"limit": 50})
    apps = []
    for app in data.get("data", []):
        apps.append({
            "id": app["id"],
            "bundle_id": app["attributes"]["bundleId"],
            "name": app["attributes"]["name"],
        })
    return apps
 
# 使用例
# [{'id': '123456789', 'bundle_id': 'com.example.app', 'name': 'My App'}, ...]

from datetime import datetime, timedelta, timezone
from dataclasses import dataclass
 
@dataclass
class Review:
    """レビューデータクラス"""
    id: str
    app_id: str
    app_name: str
    rating: int
    title: str
    body: str
    reviewer_nickname: str
    created_date: datetime
    territory: str  # 国・地域コード
    developer_response: str | None = None
 
def fetch_recent_reviews(
    client: AppStoreConnectClient,
    app_id: str,
    app_name: str,
    days: int = 7
) -> list[Review]:
    """指定期間のレビューを取得"""
    reviews = []
    cursor = None
    since = datetime.now(timezone.utc) - timedelta(days=days)
 
    while True:
        params = {
            "limit": 200,
            "sort": "-createdDate",
            "filter[territory]": "",  # 全地域
        }
        if cursor:
            params["cursor"] = cursor
 
        data = client.get(f"/apps/{app_id}/customerReviews", params=params)
 
        for item in data.get("data", []):
            attrs = item["attributes"]
            created = datetime.fromisoformat(attrs["createdDate"].replace("Z", "+00:00"))
 
            # 指定期間より古いレビューが出たら終了
            if created < since:
                return reviews
 
            # 開発者返信の取得
            response_text = None
            if item.get("relationships", {}).get("response", {}).get("data"):
                response_data = client.get(f"/customerReviews/{item['id']}/response")
                if response_data.get("data"):
                    response_text = response_data["data"]["attributes"].get("responseBody")
 
            reviews.append(Review(
                id=item["id"],
                app_id=app_id,
                app_name=app_name,
                rating=attrs["rating"],
                title=attrs.get("title", ""),
                body=attrs.get("body", ""),
                reviewer_nickname=attrs.get("reviewerNickname", "匿名"),
                created_date=created,
                territory=attrs.get("territory", "JPN"),
                developer_response=response_text,
            ))
 
        # ページネーション処理
        next_link = data.get("links", {}).get("next")
        if not next_link:
            break
        cursor = next_link.split("cursor=")[-1].split("&")[0]
 
    return reviews

import google.generativeai as genai
import json
from typing import Literal
 
genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
model = genai.GenerativeModel("gemini-2.5-flash")
 
ANALYSIS_SYSTEM_PROMPT = """
あなたはApp Storeレビュー分析の専門家です。
以下のレビューを分析し、必ず指定のJSONフォーマットで返答してください。
 
分析項目:
- sentiment: 感情（positive/negative/neutral/mixed）
- category: 問題カテゴリ（bug/ui_ux/feature_request/performance/praise/support/other）
- severity: 緊急度（1=低 〜 5=高、negativeのみ）
- key_issue: 主要な問題または賞賛点（30文字以内・日本語）
- needs_response: 返信が必要か（true/false）
- response_priority: 返信優先度（1=低 〜 5=高）
- language: レビューの言語コード（ja/en/zh/ko/fr等）
 
返答はJSON配列のみ。説明文は不要。
"""
 
def analyze_reviews_batch(reviews: list[Review]) -> list[dict]:
    """レビューをバッチ処理で分析（コスト最適化）"""
 
    # Gemini APIへのバッチリクエスト用フォーマット
    review_batch = []
    for r in reviews:
        review_batch.append({
            "id": r.id,
            "rating": r.rating,
            "title": r.title,
            "body": r.body[:500],  # 長すぎるレビューはトリム
            "territory": r.territory,
            "has_response": r.developer_response is not None,
        })
 
    prompt = f"""
以下の{len(review_batch)}件のApp Storeレビューを分析してください。
 
レビューデータ:
{json.dumps(review_batch, ensure_ascii=False, indent=2)}
 
各レビューのidに対応する分析結果をJSON配列で返してください。
フォーマット: [{{"id": "...", "sentiment": "...", "category": "...", ...}}]
"""
 
    response = model.generate_content(
        prompt,
        generation_config=genai.GenerationConfig(
            temperature=0.1,  # 分析タスクは低温度で安定させる
            response_mime_type="application/json",
        ),
        system_instruction=ANALYSIS_SYSTEM_PROMPT,
    )
 
    try:
        results = json.loads(response.text)
        return results
    except json.JSONDecodeError:
        # JSONパースエラー時のフォールバック
        print(f"⚠️ JSONパースエラー: {response.text[:200]}")
        return [{"id": r["id"], "sentiment": "neutral", "category": "other", "severity": 1,
                 "key_issue": "分析失敗", "needs_response": False, "response_priority": 1, "language": "ja"}
                for r in review_batch]
 
# バッチサイズ: Gemini 2.5 Flash は大きなコンテキストに対応しているが
# 50件程度に分割すると精度が安定する
def analyze_all_reviews(reviews: list[Review], batch_size: int = 50) -> list[dict]:
    """全レビューを分割して分析"""
    all_results = []
    for i in range(0, len(reviews), batch_size):
        batch = reviews[i:i + batch_size]
        results = analyze_reviews_batch(batch)
        all_results.extend(results)
        print(f"✅ 分析完了: {min(i + batch_size, len(reviews))}/{len(reviews)} 件")
    return all_results

REPLY_GENERATION_PROMPT = """
あなたはApp Storeで成功しているアプリ開発者です。
ユーザーレビューへの誠実で温かみのある返信を生成してください。
 
返信のルール:
1. レビューと同じ言語で返信する
2. 200文字以内（App Storeの制限に合わせる）
3. 定型文にならないよう、レビューの具体的な内容に言及する
4. ネガティブレビューには: 謝罪→原因説明（可能な場合）→改善予定→サポート案内
5. ポジティブレビューには: 感謝→開発への意欲→次のアップデート予告（ある場合）
6. バグ報告には: 謝罪→確認方法→修正状況→連絡先
7. 開発者の「私」「私たち」などの一人称は使わない（Appleのガイドラインに従い中立的に）
 
絶対に使わない表現:
- 「ご不便をおかけして申し訳ございません」（使い古された定型文）
- 「貴重なご意見ありがとうございます」（機械的に聞こえる）
- 過度な敬語の重複
"""
 
def generate_reply(review: Review, analysis: dict) -> str:
    """個別レビューへの返信文を生成"""
 
    # 返信不要と判断されたレビューはスキップ
    if not analysis.get("needs_response", True):
        return ""
 
    # 既に返信済みはスキップ
    if review.developer_response:
        return ""
 
    # アプリ情報をコンテキストに含める（実際の運用では動的に設定）
    app_context = f"""
アプリ名: {review.app_name}
アプリジャンル: 壁紙・ライフスタイル（実際の運用では設定ファイルから読み込む）
最新バージョン: 3.2.1
最近の主な変更: iOS 18対応、パフォーマンス改善、新カテゴリ追加
"""
 
    prompt = f"""
{app_context}
 
レビュー情報:
- 星: {review.rating}/5
- タイトル: {review.title or "（タイトルなし）"}
- 本文: {review.body}
- 投稿地域: {review.territory}
- 分析結果: {json.dumps(analysis, ensure_ascii=False)}
 
このレビューへの返信文を生成してください。
200文字以内、{analysis.get('language', 'ja')}で返信すること。
返信文のみ出力（前置き不要）。
"""
 
    response = model.generate_content(
        prompt,
        generation_config=genai.GenerationConfig(temperature=0.7),
        system_instruction=REPLY_GENERATION_PROMPT,
    )
 
    return response.text.strip()
 
def submit_reply(client: AppStoreConnectClient, review_id: str, reply_text: str) -> bool:
    """App Store Connect APIで返信を送信"""
    try:
        # 既存の返信を確認
        existing = client.get(f"/customerReviews/{review_id}/response")
 
        if existing.get("data"):
            # 既存返信の更新（PATCH）
            client.patch(
                f"/customerReviewResponses/{existing['data']['id']}",
                {"data": {"type": "customerReviewResponses",
                          "id": existing['data']['id'],
                          "attributes": {"responseBody": reply_text}}}
            )
        else:
            # 新規返信（POST）
            import httpx
            headers = {"Authorization": f"Bearer {client.token}", "Content-Type": "application/json"}
            body = {"data": {"type": "customerReviewResponses",
                             "attributes": {"responseBody": reply_text},
                             "relationships": {"review": {"data": {"type": "customerReviews", "id": review_id}}}}}
            httpx.post(f"{AppStoreConnectClient.BASE_URL}/customerReviewResponses",
                      headers=headers, json=body).raise_for_status()
 
        return True
    except Exception as e:
        print(f"❌ 返信送信エラー (review_id: {review_id}): {e}")
        return False

import sqlite3
from enum import Enum
 
class ReplyStatus(Enum):
    PENDING = "pending"        # 生成済み、未確認
    APPROVED = "approved"      # 確認済み、送信待ち
    SENT = "sent"              # 送信済み
    REJECTED = "rejected"      # 却下（手動返信に切り替え）
    SKIPPED = "skipped"        # 返信不要と判断
 
def setup_database(db_path: str = "reviews.db"):
    """SQLiteデータベースのセットアップ"""
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS reviews (
            id TEXT PRIMARY KEY,
            app_id TEXT,
            app_name TEXT,
            rating INTEGER,
            title TEXT,
            body TEXT,
            territory TEXT,
            created_date TEXT,
            sentiment TEXT,
            category TEXT,
            severity INTEGER,
            key_issue TEXT,
            generated_reply TEXT,
            reply_status TEXT DEFAULT 'pending',
            sent_at TEXT,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.commit()
    return conn

import httpx
from datetime import datetime
 
ITUNES_SEARCH_API = "https://itunes.apple.com"
 
class CompetitorMonitor:
    """競合アプリモニタリングクラス"""
 
    def __init__(self, competitor_app_ids: list[str], country: str = "jp"):
        self.competitor_ids = competitor_app_ids
        self.country = country
 
    def fetch_competitor_info(self, app_id: str) -> dict:
        """iTunes Search APIで競合情報を取得"""
        response = httpx.get(
            f"{ITUNES_SEARCH_API}/lookup",
            params={"id": app_id, "country": self.country, "lang": "ja_jp"},
            timeout=10
        )
        data = response.json()
        if data["resultCount"] == 0:
            return {}
 
        result = data["results"][0]
        return {
            "app_id": app_id,
            "name": result.get("trackName"),
            "seller": result.get("sellerName"),
            "current_version": result.get("version"),
            "user_rating": result.get("averageUserRating", 0),
            "rating_count": result.get("userRatingCount", 0),
            "current_version_rating": result.get("averageUserRatingForCurrentVersion", 0),
            "release_notes": result.get("releaseNotes", ""),
            "release_date": result.get("currentVersionReleaseDate"),
            "price": result.get("price", 0),
            "description": result.get("description", "")[:500],
            "keywords_hint": result.get("sellerUrl", ""),  # SEO情報の手がかり
            "fetched_at": datetime.utcnow().isoformat(),
        }
 
    def analyze_competitor_update(self, current: dict, previous: dict | None) -> str:
        """Geminiで競合のアップデートを分析"""
        if not previous:
            return "新規追跡開始"
 
        if current.get("current_version") == previous.get("current_version"):
            return "バージョン変更なし"
 
        prompt = f"""
競合アプリのアップデートを分析してください。
 
アプリ名: {current['name']}
 
前バージョン: {previous.get('current_version', '不明')}
新バージョン: {current['current_version']}
 
リリースノート:
{current.get('release_notes', 'なし')}
 
評価変化:
- 総合評価: {previous.get('user_rating', 0):.1f} → {current['user_rating']:.1f}
- 評価数: {previous.get('rating_count', 0):,} → {current['rating_count']:,}
 
以下の観点で100文字程度で分析してください:
1. 主な変更点（新機能/バグ修正/UI変更）
2. 自社アプリへの影響・対策
"""
        response = model.generate_content(prompt, generation_config=genai.GenerationConfig(temperature=0.3))
        return response.text.strip()
 
    def run_monitoring(self) -> list[dict]:
        """全競合アプリを監視"""
        results = []
        for app_id in self.competitor_ids:
            try:
                info = self.fetch_competitor_info(app_id)
                if info:
                    results.append(info)
                    print(f"✅ {info['name']} の情報を取得")
            except Exception as e:
                print(f"❌ App ID {app_id} の取得失敗: {e}")
        return results

def generate_weekly_aso_report(
    reviews: list[Review],
    analyses: list[dict],
    competitor_data: list[dict],
    my_apps: list[dict],
) -> str:
    """週次ASOインテリジェンスレポートをMarkdownで生成"""
 
    # レビュー統計の集計
    total = len(reviews)
    if total == 0:
        return "今週のレビューはありません。"
 
    rating_dist = {1: 0, 2: 0, 3: 0, 4: 0, 5: 0}
    sentiment_dist = {"positive": 0, "negative": 0, "neutral": 0, "mixed": 0}
    category_dist = {}
 
    for review, analysis in zip(reviews, analyses):
        rating_dist[review.rating] = rating_dist.get(review.rating, 0) + 1
        sentiment = analysis.get("sentiment", "neutral")
        sentiment_dist[sentiment] = sentiment_dist.get(sentiment, 0) + 1
        cat = analysis.get("category", "other")
        category_dist[cat] = category_dist.get(cat, 0) + 1
 
    avg_rating = sum(r.rating for r in reviews) / total
 
    # 緊急対応が必要なレビュー（severity 4以上のnegative）
    urgent = [
        (r, a) for r, a in zip(reviews, analyses)
        if a.get("sentiment") == "negative" and a.get("severity", 0) >= 4
    ]
 
    stats_json = json.dumps({
        "total_reviews": total,
        "avg_rating": round(avg_rating, 2),
        "rating_distribution": rating_dist,
        "sentiment_distribution": sentiment_dist,
        "category_distribution": category_dist,
        "urgent_issues_count": len(urgent),
        "urgent_issues": [
            {"title": r.title, "body": r.body[:100], "key_issue": a.get("key_issue")}
            for r, a in urgent[:5]
        ],
        "competitor_updates": [
            {"name": c["name"], "new_version": c.get("current_version"),
             "rating": c.get("user_rating"), "release_notes_snippet": c.get("release_notes", "")[:150]}
            for c in competitor_data if c.get("release_notes")
        ],
    }, ensure_ascii=False)
 
    prompt = f"""
以下のApp Storeデータをもとに、週次ASOインテリジェンスレポートをMarkdown形式で生成してください。
 
データ:
{stats_json}
 
レポート構成:
1. ## 今週のサマリー（3行以内）
2. ## レビュー分析（評価分布・感情・カテゴリ別洞察）
3. ## 緊急対応事項（severity 4-5のネガティブレビュー）
4. ## 競合動向（アップデート内容と自社への影響）
5. ## 来週のアクションプラン（具体的・実行可能な施策3件）
 
日本語で、実行可能なインサイトに焦点を当てて記述してください。
"""
 
    response = model.generate_content(
        prompt,
        generation_config=genai.GenerationConfig(temperature=0.4, max_output_tokens=2000),
    )
 
    return response.text
 
def send_slack_report(report_text: str, urgent_count: int):
    """Slackへレポートを送信"""
    color = "#ff0000" if urgent_count > 0 else "#36a64f"
    emoji = "🚨" if urgent_count > 0 else "📊"
 
    payload = {
        "attachments": [{
            "color": color,
            "title": f"{emoji} 週次ASOインテリジェンスレポート",
            "text": report_text[:2900],  # Slack の文字数制限
            "footer": f"Gemini Lab ASO System | {datetime.now().strftime('%Y-%m-%d %H:%M')} JST",
            "mrkdwn_in": ["text"],
        }]
    }
    httpx.post(os.getenv("SLACK_WEBHOOK_URL"), json=payload, timeout=10)
    print("✅ Slack通知を送信しました")

# .github/workflows/aso-monitor.yml
name: ASO Intelligence Monitor
 
on:
  schedule:
    - cron: '0 0 * * *'  # 毎日UTC 0:00（JST 9:00）
    - cron: '0 22 * * 0'  # 毎週日曜UTC 22:00（JST 月曜7:00）に週次レポート
  workflow_dispatch:       # 手動実行も可能
 
jobs:
  monitor:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - run: pip install -r requirements.txt
      - run: python aso_monitor.py --mode=daily
        env:
          APP_STORE_KEY_ID: ${{ secrets.APP_STORE_KEY_ID }}
          APP_STORE_ISSUER_ID: ${{ secrets.APP_STORE_ISSUER_ID }}
          APP_STORE_PRIVATE_KEY: ${{ secrets.APP_STORE_PRIVATE_KEY }}
          GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}

import time
from functools import wraps
 
def with_retry(max_retries: int = 3, backoff_factor: float = 2.0):
    """指数バックオフ付き再試行デコレータ"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    wait = backoff_factor ** attempt
                    print(f"⚠️ 試行 {attempt + 1}/{max_retries} 失敗: {e}. {wait}秒後に再試行...")
                    time.sleep(wait)
        return wrapper
    return decorator
 
@with_retry(max_retries=3)
def safe_analyze_reviews(reviews):
    return analyze_reviews_batch(reviews)

#!/usr/bin/env python3
"""
ASO Intelligence System — メインエントリポイント
使い方: python aso_main.py [--mode=daily|weekly|reply]
"""
 
import argparse
import sys
from datetime import date
 
def main():
    parser = argparse.ArgumentParser(description="ASO Intelligence System")
    parser.add_argument("--mode", choices=["daily", "weekly", "reply", "report"], default="daily")
    parser.add_argument("--dry-run", action="store_true", help="返信を実際には送信しない")
    args = parser.parse_args()
 
    client = AppStoreConnectClient()
    conn = setup_database()
 
    print(f"🚀 ASO Intelligence System 起動 (mode: {args.mode})")
 
    # アプリ一覧取得
    apps = get_app_ids(client)
    print(f"📱 対象アプリ: {len(apps)} 件")
 
    # 競合アプリ（設定ファイルから読み込む）
    competitor_ids = ["YOUR_COMPETITOR_APP_ID_1", "YOUR_COMPETITOR_APP_ID_2"]
    monitor = CompetitorMonitor(competitor_ids)
 
    all_reviews = []
    all_analyses = []
 
    days = 1 if args.mode == "daily" else 7
 
    for app in apps:
        # レビュー取得
        reviews = fetch_recent_reviews(client, app["id"], app["name"], days=days)
        print(f"📝 {app['name']}: {len(reviews)} 件のレビューを取得")
 
        if not reviews:
            continue
 
        # 感情分析
        analyses = analyze_all_reviews(reviews)
        all_reviews.extend(reviews)
        all_analyses.extend(analyses)
 
        if args.mode in ["daily", "reply"]:
            # 自動返信生成と送信
            for review, analysis in zip(reviews, analyses):
                if analysis.get("response_priority", 0) >= 4:
                    reply = generate_reply(review, analysis)
                    if reply and not args.dry_run:
                        success = submit_reply(client, review.id, reply)
                        status = "✅ 送信" if success else "❌ 失敗"
                        print(f"  {status}: [{review.rating}★] {review.title[:20]}...")
 
    if args.mode == "weekly" or args.mode == "report":
        # 競合情報取得
        competitor_data = monitor.run_monitoring()
 
        # 週次レポート生成
        report = generate_weekly_aso_report(all_reviews, all_analyses, competitor_data, apps)
 
        urgent_count = sum(
            1 for a in all_analyses
            if a.get("sentiment") == "negative" and a.get("severity", 0) >= 4
        )
 
        # Slack通知
        send_slack_report(report, urgent_count)
 
        # レポートをファイルに保存
        report_path = f"reports/aso_report_{date.today()}.md"
        with open(report_path, "w") as f:
            f.write(report)
        print(f"📄 レポート保存: {report_path}")
 
    print("✅ 処理完了")
 
if __name__ == "__main__":
    main()

def get_token(self) -> str:
    cached = redis.get("ascc:jwt")
    if cached:
        return cached.decode()
    token = self._generate_token()  # exp = now + 1200
    redis.set("ascc:jwt", token, ex=1080)  # キャッシュは 18 分
    return token