Vitals APIリファレンス

# 指定する必要がある値
client_id = "<クライアントID>"
client_secret = "<クライアントシークレット>"
package_name = "<アプリのパッケージ名>"

BASE_URL = "https://developer.amazon.com/api/appstore"

scope = "adx_reporting::appstore:marketer"
data = {
    "grant_type": "client_credentials",
    "client_id": client_id,
    "client_secret": client_secret,
    "scope": scope
}
auth_response = requests.post("https://api.amazon.com/auth/o2/token", data=data)
auth_token = auth_response.json()["access_token"]

headers = {
    "Authorization": f"Bearer {auth_token}",
    "Content-Type": "application/json"
}

freshness_url = f"{BASE_URL}/vitals/apps/{package_name}/crashMetricSet"
freshness_response = requests.get(freshness_url, headers=headers)
freshness = freshness_response.json()

for info in freshness["freshnessInfo"]:
    print(f"{info['aggregationPeriod']}: {info['latestDataTimestamp']}")

from datetime import datetime, timezone, timedelta


def utc_date_components(local_dt=None):
    """API向けに、datetimeをUTCのyear/month/dayに変換します。

    タイムゾーン情報のないlocal_dtは、システムのローカル時刻として解釈されます。
    あいまいさを避けるには、タイムゾーン付きのdatetimeを渡してください。
    """
    if local_dt is None:
        utc_dt = datetime.now(timezone.utc)
    else:
        if local_dt.tzinfo is None:
            local_dt = local_dt.astimezone()  # システムのローカルタイムゾーンを明示的に適用
        utc_dt = local_dt.astimezone(timezone.utc)
    return {"year": utc_dt.year, "month": utc_dt.month, "day": utc_dt.day}


# 例：UTCでの「昨日」（ローカルタイムゾーンとは無関係）
yesterday_utc = datetime.now(timezone.utc) - timedelta(days=1)
start = utc_date_components(yesterday_utc)

metric_set = "crashMetricSet"   # or "anrMetricSet", "lmkMetricSet"
query_url = f"{BASE_URL}/vitals/apps/{package_name}/{metric_set}:query"

body = {
    # 必須：時間範囲と詳細度
    "timelineSpec": {
        "aggregationPeriod": "DAILY", # DAILYまたはHOURLY
        "startTime": {"year": 2026, "month": 4, "day": 01},
        "endTime":   {"year": 2026, "month": 4, "day": 01},
    },

    # 省略可能：クエリ対象の指標セット内の指標
    "metrics": ["crashRate", "crashCount", "distinctDevices"],

    # 省略可能：ディメンション別の内訳（最大5つ）
    "dimensions": ["versionCode", "countryCode"],

    # 省略可能： AIP-160仕様のフィルター式
    "filter": {"countryCode": ["US", "CA"], "deviceType": ["AMAZON_FIRE_TV"]},

    # 省略可能：ページ分割（デフォルトは1000、最大100000）
    "pageSize": 1000,
    "pageToken": "<前回のレスポンスから取得>",
}

response = requests.post(query_url, headers=headers, json=body)
result = response.json()

all_rows = []
body.pop("pageToken", None)
while True:
    result = requests.post(query_url, headers=headers, json=body).json()
    all_rows.extend(result["rows"])
    token = result.get("nextPageToken")
    if not token:
        break
    body["pageToken"] = token

issues_url = f"{BASE_URL}/vitals/apps/{package_name}/issuesMetricSet:query"
body = {
  "startTime": { "year": 2025, "month": 6, "day": 1 },
  "endTime": { "year": 2025, "month": 6, "day": 15 },
  "reportType": "CRASH",
  "dimensions": ["deviceType", "osVersion"],
  "filter": { "countryCode": ["US", "CA"] },
  "pageSize": 25,
  "pageToken": null
}
response = requests.post(issues_url, headers=headers, json=body)
result = response.json()

# レスポンス構造（発生率指標セットとは異なります）：
# {
#   "rows": [
#     {
#       "crashDescriptor": "67f35eb2...",
#       "reportText": "java.lang.IllegalStateException: ...",
#       "metrics": [
#         {"name": "errorEventCount", "value": "78"},
#         {"name": "affectedDeviceCount", "value": "45"}
#       ]
#     }
#   ],
#   "nextPageToken": "..."
# }

import time

MAX_RETRIES = 3
RETRIABLE_CODES = {429, 500, 502, 503}

def query_with_retry(client, package, metric_set, body):
    for attempt in range(MAX_RETRIES):
        resp = client.query(package, metric_set, body)

        if resp.status_code == 200:
            return resp.json()

        if resp.status_code not in RETRIABLE_CODES:
            break  # 再試行不可のエラー

        wait = (2 ** attempt) + 0.5  # 1.5秒、2.5秒、4.5秒
        print(f"Retry {attempt + 1}/{MAX_RETRIES} after {wait}s (HTTP {resp.status_code})")
        time.sleep(wait)

    # 最後の試行は失敗しました
    error = resp.json() if resp.headers.get("content-type", "").startswith("application/json") else {}
    raise RuntimeError(f"Failed after {MAX_RETRIES} retries: {resp.status_code} {error.get('message', '')}")