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):
    """将日期时间转换为UTC的年/月/日格式，以供API使用。

    不带时区信息的local_dt会被解释为系统本地时间。
    请提供tz-aware，以免系统误判。
    """
    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", # 按日/按小时
        "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  # Non-retriable error

        wait = (2 ** attempt) + 0.5  # 1.5s, 2.5s, 4.5s
        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', '')}")