アプリのダウンロード数収集をGCPで実装してみる

なぜこんなことをすることになったか

先日、アプリストアからダウンロード数を取得・集計するスクリプトが逝ってしまいました。すでにいないエンジニア(しかも他チーム)が作ったもので原因すらわかりません。

ついにこの日が来てしまったという感じですが、 ダウンロード数は、弊社のマーケティングチームが追っている大事な数字です。そこで、社内のデータを集める役割の自分は新たな仕組みを新たに構築することになりました。

ちょっと端折り端折り書きますが、誰かの役に立てば幸いです。

方針としては、とにかくサーバレス（自分が構築している分析基盤は基本このポリシーです！）な実装で数字を取得したいと思います。 今回は、最終的にGCPのBigQueryにデータを格納することを想定しているので、自然とGCPのサービスを使います。

アプリのダウンロード数は、Androidは Google Play Store 、iOSは Apple Store Connect から取得できます。

では、それぞれやっていきたいと思います。

Google Play

これはすごく楽でした。

なんと BigQueryにData Transfer Service というサービスがありまして、 以下の手順で自動的にBigQueryにテーブル連携できました。

1. BigQuery の画面の左上の 「Tranfers」 を選択

1. 「Add Transfer」 を選択

1. 「Select an option」 から 「Google Play」 を選択
2. 項目を埋めて 「Add」

「Cloud Storage bucket」 には、IDに基づく Google Cloud Storage のバケットパスを入れる必要があります。 IDは、Google Play Console に接続して、「レポートをダウンロード」から対象アプリを選択するとレポートの最下部に "gs://pubsite_prod_rev_[ID]/reviews/" の記載があるのでこれを利用します。

これだけで指定したデータセットにレポートのデータが格納されます。

Apple Connect Store

こちらはちょっと面倒でした。。

Apple Store Connect からダウンロード数などのレポートを取得する方法は２つあります。

1. 専用ツール(jar) [参考]
2. API

今回は、実装しやすそうなので　API　にしました。

APIの説明については、こちらにあります。

大まかな手順としては、

1. App Store Connect API　の API private キーを作成する [マニュアル]
2. private キーを使ってJWTトークンを生成する　[マニュアル]
3. App Store Connect API　をコールして、レポートを取得する

となります。

Google Cloud Functions に python3 で実装していきます。

JWTトークンの生成

def __get_apple_store_connect_api_token():

    jst = datetime.timezone(datetime.timedelta(hours=+9), 'JST')
    now_ut = int(datetime.datetime.now(jst).timestamp()) + 20 * 60

    headers = {"alg": "ES256",
               "kid": str(os.environ['key_id']),
               "typ": "JWT"}
    payload = {"iss": str(os.environ['issuer_id']),
               "exp": now_ut,
               "aud": "appstoreconnect-v1"}

    # https://pyjwt.readthedocs.io/en/latest/installation.html
    private_key = os.environ['private_key']
    private_key = re.sub(r'\\n',r'\n',private_key)

    token = jwt.encode(payload,
                       private_key,
                       algorithm='ES256',
                       headers=headers).decode()

    logging.debug('Apple Store Connect API token: ' + token)

    return token

説明

ポイントだけいきますと、ヘッダーとペイロードとして、以下ように「key ID」と「Issuer ID」を入れます。 また、「ext」 は、20分先の時間を設定します。（上のマニュアル参照）

ほかはこの通りで大丈夫です。

    jst = datetime.timezone(datetime.timedelta(hours=+9), 'JST')
    now_ut = int(datetime.datetime.now(jst).timestamp()) + 20 * 60

    headers = {"alg": "ES256",
               "kid": str(os.environ['key_id']),
               "typ": "JWT"}
    payload = {"iss": str(os.environ['issuer_id']),
               "exp": now_ut,
               "aud": "appstoreconnect-v1"}

Private key　の文字列をつかって、以下の用にJWTトークンを生成します。 (Google Cloud Functionsの環境変数の扱いの仕様上、途中で文字整形しています。)

    private_key = os.environ['private_key']
    private_key = re.sub(r'\\n',r'\n',private_key)

    token = jwt.encode(payload,
                       private_key,
                       algorithm='ES256',
                       headers=headers).decode()

APIからレポート取得

def __get_reports(token,date_str=None):
    base_url = 'https://api.appstoreconnect.apple.com/v1/salesReports'

    if not date_str:
        jst = datetime.timezone(datetime.timedelta(hours=+9), 'JST')
        date_str = (datetime.datetime.now(jst) - datetime.timedelta(days=1)).strftime('%Y-%m-%d')
    query_parameters = {
        "filter[frequency]": "DAILY",
        "filter[reportSubType]": "SUMMARY",
        "filter[reportType]": "SALES",
        "filter[vendorNumber]": os.environ['vendor_number'],
        "filter[reportDate]": date_str
    }

    url = base_url

    logging.debug("url: " + url)
    logging.debug("query_params[filter[reportDate]]: " + date_str)

    headers = {'Authorization': 'Bearer ' + token,
                       'Accept': '*/*'}

    req = urllib.request.Request('{}?{}'.format(url, urllib.parse.urlencode(query_parameters)), headers=headers)
    try:
        with urllib.request.urlopen(req) as res:
            content = res.read()
        code = 200
    except (urllib.error.HTTPError, urllib.error.URLError) as e:
        logging.error("Error occered in urllib.request.Request().\n code: " + str(e.code) + \
                     "\n reason: " + e.reason)
        code = e.code
        content = e.reason

    return (code, content, date_str)

説明

ここはやっていることは簡単でurllibを使ってAPIを叩いて、レポートデータをとってきています。

その際に、JWTトークンを header に仕込んでおきます。

    headers = {'Authorization': 'Bearer ' + token,
                       'Accept': '*/*'}

    req = urllib.request.Request('{}?{}'.format(url, urllib.parse.urlencode(query_parameters)), headers=headers)

Google Cloud Functions にデプロイ

今回は、 Google Cloud Functions を使って、API I/Fを作ることにしました。 このAPIを叩くとApple Connect StoreのAPIを叩いて、取れたレポートを Google Cloud Storage に保存する動きをします。

一部省略していたり、エラーハンドリングとか省いていますが、 以下のソースコードをデプロイします。

import logging, os, sys, re
import pprint
import datetime

import urllib.request, urllib.error

from google.cloud import storage

import jwt

def get_apple_reports(request):

    token = __get_apple_store_connect_api_token()

    date_str = request.args.get('date')
    (code, content, date_str) = __get_reports(token, date_str)

    filename_header = 'test/'
    filename = filename_header + 'S_D_' + os.environ['vendor_number'] \
                    + '_' + date_str.replace('-','') + '.txt.gz'

    __write_to_gcs(content,filename)

    return (None,200,None)


def __get_apple_store_connect_api_token():
    <省略>

def __get_reports(token,date_str=None):
    <省略>

def __write_to_gcs(content,filename):
    storage_client = storage.Client()
    bucket_name = 'bucket-name'
    bucket = storage_client.get_bucket(bucket_name)
    blob = bucket.blob(filename)
    blob.upload_from_string(content)

if __name__ == '__main__':
    request = {"args": {"date_str": None}}
    get_apple_reports(request)

WebUIでみるとこんな感じです。

実行スケジューラ Google Cloud Scheduler

上の　Functionｓ　を定期実行したいので、最近リリースされた Google Cloud Scheduler を使います。Cloud Scheduler は、cronライクにHTTPリクエストを発出したり、　Google App Engine　のエンドポイントを叩いたりできます。

以下は、Cloud Schedulerの設定WebUIですが、赤枠のところに Cloud Functions のエンドポイントを設定します。

これで毎日自動的に Apple Store Connect のレポートが Google Cloud Storage に溜まっていきます。ちなみに、今回は省略しますが、このあと定期的にBigQueryにこのファイルをロードして、テーブルデータを更新して運用しています。

よく考えたら Functions　の中で直接BigQueryに Ingest　したらよかったですね。。。

Mobility Technologies では共に日本のモビリティを進化させていくエンジニアを募集しています。話を聞いてみたいという方は、是非 募集ページ からご相談ください！