Play Store 段階公開を Claude Code で自動監視する — v2.1.0 クラッシュ対応から学んだリリース戦略

# google_play_vitals.py
# Google Play Developer API から Crash-free Rate を取得する
 
from google.oauth2 import service_account
from googleapiclient.discovery import build
from datetime import datetime, timedelta
import json
 
PACKAGE_NAME = "net.dolice.beautifulwallpapers"
SERVICE_ACCOUNT_FILE = "service_account.json"
 
SCOPES = ["https://www.googleapis.com/auth/androidpublisher"]
 
def get_crash_metrics(days_back: int = 7) -> dict:
    """指定期間のクラッシュ指標を取得する"""
    credentials = service_account.Credentials.from_service_account_file(
        SERVICE_ACCOUNT_FILE, scopes=SCOPES
    )
    service = build("androidpublisher", "v3", credentials=credentials)
    
    end_date = datetime.now()
    start_date = end_date - timedelta(days=days_back)
    
    # Crash-free Users Rate を取得
    request = service.vitals().crashrate().get(
        packageName=PACKAGE_NAME,
        # メトリクス: CRASH_RATE = 全ユーザー中クラッシュが発生したユーザーの割合
        # つまり Crash-free Rate = 1 - crash_rate
    )
    
    result = request.execute()
    return result
 
def parse_crash_free_rate(vitals_data: dict) -> float:
    """API レスポンスから Crash-free Users Rate を計算する"""
    # crash_rate が返ってくる（クラッシュが発生したユーザーの割合）
    # Crash-free Rate = 100 - crash_rate_percentage
    crash_rate = vitals_data.get("crashRate", {}).get("rate", 0)
    crash_free_rate = (1 - crash_rate) * 100
    return round(crash_free_rate, 3)

# csv_reader.py
# Play Console から手動エクスポートした CSV を読み込む（API不使用の場合）
 
import csv
from pathlib import Path
 
def read_vitals_csv(csv_path: str) -> list[dict]:
    """Play Console の Android Vitals CSV を読み込む"""
    rows = []
    with open(csv_path, encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        for row in reader:
            rows.append({
                "date": row.get("Date", ""),
                "crash_free_rate": float(row.get("Crash-free users (%)", 0)),
                "anr_free_rate": float(row.get("ANR-free users (%)", 0)),
                "crash_count": int(row.get("Crashes", 0)),
                "version": row.get("App version", "")
            })
    return rows

# go_no_go_analyzer.py
# 段階公開の Go/No-Go 判定ロジック
 
import math
from dataclasses import dataclass
from enum import Enum
 
class Decision(Enum):
    PROCEED = "PROCEED"       # 次のステップへ進む
    WATCH = "WATCH"           # 様子見（警戒）
    PAUSE = "PAUSE"           # 一時停止
    ROLLBACK = "ROLLBACK"     # ロールバック推奨
 
@dataclass
class RolloutMetrics:
    crash_free_rate: float    # 例: 99.8 (%)
    anr_free_rate: float      # 例: 99.9 (%)
    crash_count: int          # 期間中の総クラッシュ数
    active_users: int         # 対象ユーザー数
    rollout_percentage: int   # 現在の公開割合 (5, 25, 50, 100)
    days_elapsed: int         # 公開からの経過日数
 
def wilson_lower_bound(successes: int, total: int, z: float = 1.96) -> float:
    """
    Wilson スコア区間の下限を計算する
    クラッシュなしユーザーを「成功」として信頼区間を推定する
    z=1.96 は 95% 信頼区間に相当する
    """
    if total == 0:
        return 0.0
    p_hat = successes / total
    denominator = 1 + z**2 / total
    center = p_hat + z**2 / (2 * total)
    margin = z * math.sqrt(p_hat * (1 - p_hat) / total + z**2 / (4 * total**2))
    return (center - margin) / denominator * 100
 
def analyze_rollout(metrics: RolloutMetrics) -> tuple[Decision, str]:
    """
    段階公開の Go/No-Go を判定する
    Returns: (判定結果, 理由の説明)
    """
    # サンプルサイズの計算（Crash-free Users 数）
    crash_free_users = int(metrics.active_users * metrics.crash_free_rate / 100)
    
    # Wilson 下限（最悪ケースの推定）
    wilson_lower = wilson_lower_bound(crash_free_users, metrics.active_users)
    
    # 段階ごとの最小経過日数（十分なサンプルを確保するため）
    min_days = {5: 1, 25: 2, 50: 3, 100: 5}.get(metrics.rollout_percentage, 3)
    
    reasons = []
    
    # 1. 絶対的な閾値チェック
    if metrics.crash_free_rate < 99.5:
        return Decision.ROLLBACK, f"Crash-free Users が {metrics.crash_free_rate:.3f}% — 即時ロールバック推奨（基準: 99.5% 未満）"
    
    if metrics.anr_free_rate < 99.8:
        return Decision.ROLLBACK, f"ANR-free Users が {metrics.anr_free_rate:.3f}% — ロールバック推奨（基準: 99.8% 未満）"
    
    # 2. Wilson 下限による信頼区間チェック
    if wilson_lower < 99.5:
        reasons.append(f"Wilson 下限 {wilson_lower:.3f}% が警戒ゾーン（サンプルサイズ: {metrics.active_users:,}）")
    
    # 3. 観測期間が不十分な場合
    if metrics.days_elapsed < min_days:
        reasons.append(f"観測期間 {metrics.days_elapsed}日（推奨: {min_days}日以上）")
    
    # 4. 警告ゾーン
    if metrics.crash_free_rate < 99.7:
        reasons.append(f"Crash-free Users {metrics.crash_free_rate:.3f}% — 警告ゾーン（基準: 99.7%）")
    
    if reasons:
        return Decision.WATCH, "、".join(reasons)
    
    return Decision.PROCEED, f"すべての指標が基準を満たしています（Crash-free: {metrics.crash_free_rate:.3f}%、ANR-free: {metrics.anr_free_rate:.3f}%）"

# report_generator.py
# Claude API を使って段階公開レポートを生成する
 
import anthropic
from datetime import datetime
 
client = anthropic.Anthropic()  # ANTHROPIC_API_KEY は環境変数から
 
def generate_rollout_report(
    app_version: str,
    metrics_history: list[dict],
    decisions: list[tuple],
    final_outcome: str
) -> str:
    """
    段階公開の履歴から運用レポートを生成する
    metrics_history: 各段階での指標データのリスト
    decisions: 各段階での判定結果と理由のリスト
    """
    
    # プロンプト構築
    history_text = "\n".join([
        f"- {d['rollout_pct']}% 公開（{d['days_elapsed']}日経過）: "
        f"Crash-free {d['crash_free_rate']:.3f}%, ANR-free {d['anr_free_rate']:.3f}%, "
        f"クラッシュ {d['crash_count']} 件 → {decisions[i][0].value}"
        for i, d in enumerate(metrics_history)
    ])
    
    prompt = f"""以下は Android アプリ（バージョン {app_version}）の段階公開モニタリング記録です。
 
## 段階別データ
{history_text}
 
## 最終結果
{final_outcome}
 
この記録を元に、以下の観点で簡潔なレポートを日本語で作成してください：
1. 今回のリリースで特筆すべき指標の動き（あった場合）
2. Go/No-Go 判定が「WATCH」または「PAUSE」になった原因（あった場合）
3. 次のリリースに向けた改善提案（具体的に1〜2点）
 
文体は「〜でした」「〜でしょう」の敬体で、技術的に正確に書いてください。
余計な前置きは不要です。直接内容から始めてください。"""
 
    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": prompt}
        ]
    )
    
    return message.content[0].text
 
def save_report(version: str, report: str) -> None:
    """レポートをファイルに保存する"""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M")
    filename = f"rollout_report_{version}_{timestamp}.md"
    
    with open(f"reports/{filename}", "w", encoding="utf-8") as f:
        f.write(f"# Play Store 段階公開レポート: {version}\n\n")
        f.write(f"生成日時: {datetime.now().strftime('%Y-%m-%d %H:%M JST')}\n\n")
        f.write(report)
    
    print(f"✅ レポート保存: reports/{filename}")

# monitor.py
# 段階公開モニタリングのメインスクリプト（統合版）
 
import time
import logging
from pathlib import Path
from typing import Optional
from dataclasses import dataclass
 
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler("monitor.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)
 
@dataclass
class MonitorConfig:
    package_name: str
    check_interval_hours: int = 6     # 6時間ごとにチェック
    rollout_percentage: int = 5       # 現在の公開割合
    crash_free_warning: float = 99.7  # 警告閾値
    crash_free_stop: float = 99.5     # 停止閾値
    anr_free_stop: float = 99.8       # ANR 停止閾値
    notification_webhook: Optional[str] = None  # Slack webhook URL
 
def run_monitor(config: MonitorConfig) -> None:
    """
    段階公開モニタリングのメインループ
    エラー時は指数バックオフでリトライする
    """
    retry_count = 0
    max_retries = 3
    
    while True:
        try:
            logger.info(f"[{config.package_name}] 監視チェック開始 ({config.rollout_percentage}% 公開中)")
            
            # データ取得（CSV または API）
            metrics = fetch_current_metrics(config)
            
            if metrics is None:
                logger.warning("メトリクス取得失敗 — データ欠損の可能性。次回チェックまで待機")
                time.sleep(config.check_interval_hours * 3600)
                continue
            
            # Go/No-Go 判定
            decision, reason = analyze_rollout(metrics)
            
            logger.info(f"判定: {decision.value} — {reason}")
            
            # 通知送信（PAUSE/ROLLBACK の場合は必ず）
            if decision in (Decision.PAUSE, Decision.ROLLBACK):
                send_alert(
                    config=config,
                    decision=decision,
                    metrics=metrics,
                    reason=reason
                )
                
                if decision == Decision.ROLLBACK:
                    logger.critical("🚨 ロールバック推奨 — 即時対応が必要です")
                    # ロールバック推奨時はモニタリングを停止して手動対応を待つ
                    break
            
            retry_count = 0  # 成功したらリトライカウントをリセット
            
        except Exception as e:
            retry_count += 1
            wait_seconds = min(300 * (2 ** retry_count), 3600)  # 最大1時間
            logger.error(f"エラー発生 (試行 {retry_count}/{max_retries}): {e}")
            
            if retry_count >= max_retries:
                logger.critical("最大リトライ回数に達しました。手動確認が必要です。")
                send_alert(config=config, decision=Decision.PAUSE, 
                          metrics=None, reason=f"監視システムエラー: {e}")
                break
            
            logger.info(f"{wait_seconds}秒後に再試行します")
            time.sleep(wait_seconds)
            continue
        
        # 次のチェックまで待機
        logger.info(f"次のチェックまで {config.check_interval_hours} 時間待機")
        time.sleep(config.check_interval_hours * 3600)
 
def send_alert(config: MonitorConfig, decision: Decision, 
               metrics: Optional[RolloutMetrics], reason: str) -> None:
    """Slack または標準出力にアラートを送信する"""
    if not config.notification_webhook:
        print(f"⚠️  アラート: {decision.value} — {reason}")
        return
    
    import requests
    
    emoji = {"PROCEED": "✅", "WATCH": "👀", "PAUSE": "⏸️", "ROLLBACK": "🚨"}.get(decision.value, "ℹ️")
    
    message = {
        "text": f"{emoji} *Play Store 段階公開アラート*",
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*{emoji} {decision.value}* — {config.package_name}\n{reason}"
                }
            }
        ]
    }
    
    if metrics:
        message["blocks"].append({
            "type": "section",
            "fields": [
                {"type": "mrkdwn", "text": f"*Crash-free*\n{metrics.crash_free_rate:.3f}%"},
                {"type": "mrkdwn", "text": f"*ANR-free*\n{metrics.anr_free_rate:.3f}%"},
                {"type": "mrkdwn", "text": f"*公開割合*\n{metrics.rollout_percentage}%"},
                {"type": "mrkdwn", "text": f"*クラッシュ数*\n{metrics.crash_count:,}"}
            ]
        })
    
    try:
        response = requests.post(config.notification_webhook, json=message, timeout=10)
        response.raise_for_status()
    except Exception as e:
        logger.error(f"通知送信失敗: {e}")

import glob
 
def find_latest_vitals_csv(download_dir: str) -> str | None:
    pattern = f"{download_dir}/crashrate_*.csv"
    files = sorted(glob.glob(pattern))
    return files[-1] if files else None