Claude Code で App Store Connect と AdMob の売上を毎朝集計するパイプライン — 壁紙アプリ 4 本で 8 週運用した実装ノート

# ❌ 動かなかった最初のコード
import jwt
import time
 
def make_token(key_path: str, key_id: str, issuer_id: str) -> str:
    with open(key_path, "rb") as f:
        private_key = f.read()
    headers = {"alg": "ES256", "kid": key_id, "typ": "JWT"}
    payload = {
        "iss": issuer_id,
        "iat": int(time.time()),
        "exp": int(time.time()) + 1200,
        "aud": "appstoreconnect-v1",
    }
    return jwt.encode(payload, private_key, algorithm="ES256", headers=headers)

# ✅ 動いた最終形
import jwt
import time
 
def make_token(key_path: str, key_id: str, issuer_id: str) -> str:
    with open(key_path, "rb") as f:
        private_key = f.read()
    now = int(time.time())
    payload = {
        "iss": issuer_id,
        "iat": now,
        "exp": now + 1200,
        "aud": "appstoreconnect-v1",
    }
    return jwt.encode(
        payload,
        private_key,
        algorithm="ES256",
        headers={"kid": key_id, "typ": "JWT"},
    )

# 落とし穴 1: gzip を展開せずそのまま CSV パーサに渡すと文字化けする
# 落とし穴 2: その日のセールスがゼロだと 404 ではなく 200 で本文が空
 
def fetch_daily_sales(token: str, vendor_id: str, report_date: str) -> bytes | None:
    url = "https://api.appstoreconnect.apple.com/v1/salesReports"
    params = {
        "filter[frequency]": "DAILY",
        "filter[reportDate]": report_date,
        "filter[reportType]": "SALES",
        "filter[reportSubType]": "SUMMARY",
        "filter[vendorNumber]": vendor_id,
        "filter[version]": "1_0",
    }
    headers = {"Authorization": f"Bearer {token}"}
    r = requests.get(url, params=params, headers=headers, timeout=30)
    if r.status_code == 404:
        # Apple の集計が未確定 (前日朝に呼ぶと UTC-8 でまだ未確定の場合がある)
        return None
    r.raise_for_status()
    # gzip 展開
    return gzip.decompress(r.content)

def fetch_with_retry(fn, *args, max_retry: int = 4) -> bytes | None:
    for attempt in range(max_retry):
        try:
            result = fn(*args)
            if result is not None:
                # ヘッダ行のみの空レポートを除外
                lines = result.decode("utf-8").splitlines()
                if len(lines) <= 1:
                    return None
                return result
        except requests.HTTPError as e:
            if e.response.status_code in (429, 500, 502, 503):
                wait = 2 ** attempt
                time.sleep(wait)
                continue
            raise
    return None

from googleapiclient.discovery import build
from google.oauth2 import service_account
 
SCOPES = ["https://www.googleapis.com/auth/admob.readonly"]
 
def get_admob_client(sa_path: str):
    creds = service_account.Credentials.from_service_account_file(
        sa_path, scopes=SCOPES
    )
    return build("admob", "v1beta", credentials=creds)

report_spec = {
    "dateRange": {
        "startDate": {"year": 2026, "month": 5, "day": 22},
        "endDate":   {"year": 2026, "month": 5, "day": 22},
    },
    "dimensions": ["DATE", "APP", "AD_UNIT", "COUNTRY", "FORMAT"],
    "metrics": [
        "AD_REQUESTS", "MATCHED_REQUESTS",
        "IMPRESSIONS", "CLICKS",
        "ESTIMATED_EARNINGS",  # JPY で返る
        "IMPRESSION_CTR",
    ],
    "localizationSettings": {
        "currencyCode": "JPY",
        "languageCode": "ja",
    },
    "timeZone": "Asia/Tokyo",
}

-- JST 基準の日次サマリ
CREATE TABLE daily_summary (
    jst_date TEXT NOT NULL,
    app_id TEXT NOT NULL,
    asc_revenue_jpy INTEGER,        -- App Store Connect (PT を JST に補正済み)
    admob_earnings_jpy INTEGER,     -- AdMob (元から JST 指定)
    admob_impressions INTEGER,
    admob_ecpm_jpy REAL,
    subscriber_count INTEGER,
    PRIMARY KEY (jst_date, app_id)
);

-- 前日比 25% 以上の変動を抽出
WITH yesterday AS (
  SELECT app_id, asc_revenue_jpy, admob_earnings_jpy
  FROM daily_summary
  WHERE jst_date = date('now', '-1 day', 'localtime')
),
day_before AS (
  SELECT app_id, asc_revenue_jpy AS prev_asc, admob_earnings_jpy AS prev_admob
  FROM daily_summary
  WHERE jst_date = date('now', '-2 day', 'localtime')
)
SELECT
  y.app_id,
  y.asc_revenue_jpy, d.prev_asc,
  ROUND( CAST(y.asc_revenue_jpy AS REAL) / NULLIF(d.prev_asc, 0) - 1.0 , 3) AS asc_pct,
  y.admob_earnings_jpy, d.prev_admob,
  ROUND( CAST(y.admob_earnings_jpy AS REAL) / NULLIF(d.prev_admob, 0) - 1.0 , 3) AS admob_pct
FROM yesterday y JOIN day_before d ON y.app_id = d.app_id
WHERE ABS(CAST(y.asc_revenue_jpy AS REAL) / NULLIF(d.prev_asc, 0) - 1.0) >= 0.25
   OR ABS(CAST(y.admob_earnings_jpy AS REAL) / NULLIF(d.prev_admob, 0) - 1.0) >= 0.25;

# 1) 仮想環境を作る
python3 -m venv .venv && source .venv/bin/activate
pip install pyjwt cryptography requests google-api-python-client \
            google-auth google-auth-httplib2
 
# 2) ディレクトリ
mkdir -p revenue_pipeline/{secrets,raw,scripts,db}
 
# 3) ASC キーと AdMob サービスアカウントを secrets/ に置く
 
# 4) Claude Code に依頼するプロンプト例
#    「revenue_pipeline/scripts/asc_fetcher.py を作って。
#     - JWT 認証は PyJWT 2.8 で
#     - Sales Report (DAILY/SUMMARY) を 7 日分取得
#     - gzip 展開して raw/asc_YYYY-MM-DD.tsv に保存
#     - 200 でヘッダ行のみは None 扱いで再キュー
#     - 失敗時は指数バックオフで 4 回まで再試行」