初心者のApp Engine | QiitaのLGTMを集計 + API化

ちょっとポートフォリオでも作ろうかと考えているため、取りあえず初めの一歩として、Qiitaの活動を可視化できるように、Qiitaの情報を収集 + 活用できるAPIをGoogle App Engineで作ってみました。

**注意**
GAEの手順を可能な限り簡略化しているので、一部 Multi-RegionのCloud Storageが自動生成されて、`¥3 / month`ほどのコストがかかっています。

## 対象読者
以下の内容を書いてます。

* Flaskの簡単な書き方
* Qiita APIの使い方
* App Engine へのデプロイ方法
* App Engine のスケジューリング
* PythonからDatastoreへの保存・参照方法

## GitHub

# 概要
以下の仕組みで作成しています。  

![architechture.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/152224/16da90c6-7cd3-bbff-c41b-9fa125ebbd86.png)

## 目次
* [Qiita APIから集計](#qiita-apiから集計)
* [Cloud Schedulerで、集計処理を定期実行](#cloud-schedulerで集計処理を定期実行)
* [Datastoreの値を返すAPI実装](#datastoreの値を返すapi実装)
* [App Engineにデプロイ](#app-engineにデプロイ)

## やりたいこと
自分のポートフォリオサイトとかで、Qiitaの、LGTM数 / ストック数 / View数の集計値出したいなと思ったため、App Engineで

* Qiita APIから、各記事の数を集計
* GCPのFirestoreで管理
* 管理データをAPIで使えるように

を実装します。

## Project Architecture
```bash=
/
|- main.py (Flask App)
|- app.yaml (GAEの設定)
|- cron.yaml (Cloud Schedulerの設定)
┗ requirements.txt (Python依存ライブラリリスト)
```

# Qiita APIから集計
## 集計クラスの実装
### 初期化クラス
初期化クラスでは

1. Qiita APIのアカウント設定
2. 自分のQiita記事のリストを取得 (`_get_post_list()`)
3. Datastore Python Clientの設定

を行っています。

**Point** 
 
* Qiitaのアクセストークンは、App Engineの環境変数で管理(設定方法は後述)

```python
class QiitaWorker(object):
    
    def __init__(self):
        self.access_token = os.environ.get('QIITA_TOKEN')
        _username = 'sasayabaku'

        self.base_url = "https://qiita.com/api/v2/"
        self.items_url = self.base_url + "users/" + _username + "/items?per_page=100"
        self.item_url = self.base_url + "items/"

        self.headers = {
            "content-type": "application/json",
            "Authorization": "Bearer " + self.access_token
        }

        self.total = {
            "likes": 0,
            "views": 0,
            "stocks": 0
        }

        self._get_post_list()

        self.datastore_client = datastore.Client()
        
    def _get_post_list(self):
        json = requests.get(self.items_url, headers=self.headers)
        self.items_list = [item['id'] for item in json.json()]
```

### Qiita 各記事の情報集計
初期化クラスで取得した、記事リストからQiita APIでLGTM / ストック数 / View数 を取得して、`for文`で合計値として集計。

```python
class QiitaWorker(object):

    <・・中略・・>

    def collect(self):
        for item in tqdm(self.items_list):
            _url = self.item_url + str(item)

            # 各記事の情報を取得
            res_json = requests.get(_url, headers=self.headers).json()
            stock_json = self._get_item_stock(item)
            item_new_document = {
                "id": item,
                "likes": res_json['likes_count'],
                "views": res_json['page_views_count'],
                "stocks": len(stock_json)
            }
            
            # 1つの変数に合計値を追加
            self.total['likes'] += item_new_document['likes']
            self.total['views'] += item_new_document['views']
            self.total['stocks'] += item_new_document['stocks']
            

    def _get_item_stock(self, item):
        # ストック数を取得するAPIの呼び出し
        _url = self.item_url + str(item) + "/stockers"
        stock_json = requests.get(_url, headers=self.headers).json()
        return stock_json
```

### Datastoreにアップロード

集計したデータを、DataStoreに保存します。  
以下を参考にしました。

* **PythonでDatastoreを扱う**  
[Python3でDatastore を動かしてみた - Qiita](https://qiita.com/tfuruya/items/fa529c6cf54adaa81aa7)

```python
class QiitaWorker(object):

    <・・中略・・>
    
    def update_datastore(self):
        kind = "Aggregate"
        task_key = self.datastore_client.key(kind)

        task = datastore.Entity(key=task_key)

        task['date'] = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9)))

        task['likes'] = self.total['likes']
        task['views'] = self.total['views']
        task['stocks'] = self.total['stocks']

        self.datastore_client.put(task)
```

### API部分の実装
App Engineの関数をCronで回すために、APIとして実装します。先ほどの集計関数を実行するAPIを以下で実装。Web App開発で、localhostからの呼び出しを連発する想定なので、CORS対策も入れています。

```python
app = Flask(__name__)
CORS(app)

@app.route('/')
def worker():
    worker = QiitaWorker()
    worker.collect()
    worker.update_datastore()

    return jsonify({'message': 'work success'}), 200
```

## Cloud Storage | App保存先設定
App Engineのコードは、Cloud Storageに保存されます。  
特に指定がない場合は、`プロジェクト名`に準拠した名前のバケットが自動作成されそこに管理されますが、今回は、`リージョン指定 + Standard Storage`設定にするために、自前で作成したバケットに保存するようにします。

> gae-example って名前のバケット名にします

![storage_create.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/152224/a0ce2c8d-d406-6c2f-f2b8-08a998c96fd7.png)


## App Engineにデプロイ

作成したFlask Appを App Engineにデプロイ。

App Engineをほとんど触ったことがなかったので、以下を参考にしました.

* **Hello Worldを返すクイックスタート**
[App Engine スタンダード環境での Python 3 のクイックスタート](https://cloud.google.com/appengine/docs/standard/python3/quickstart?hl=ja)

* **App Engineにデプロイする手順マニュアル**
[アプリケーションのテストとデプロイ  |  Python 3 の App Engine スタンダード環境  |  Google Cloud](https://cloud.google.com/appengine/docs/standard/python3/testing-and-deploying-your-app?hl=ja)

* **Qiita APIの使い方**
[Qiita API v2 の概要(非公式) - Qiita](https://qiita.com/tag1216/items/b0b90e30c7e581aa2b00)

```bash
gcloud app deploy --bucket=gs://qiita-worker
```

### アプリケーション設定(app.yamlの記載)
Qiitaトークンをコードに埋め込みたくなかったので、環境変数で管理したかったため、app.yamlの `env_variables`で設定します。

```yaml
runtime: python38
instance_class: F1
env_variables:
    QIITA_TOKEN: 'Qiitaのアクセストークン'
```

### デプロイコマンド
```bash
gcloud app deploy --bucket=gs://qiita-worker
```

# Cloud Schedulerで、集計処理を定期実行

ここまでで、Qiitaから集計してDatastoreに保存するアプリケーションが完成しました。  
ただ、毎回集計のたびにAPIをコールする必要があるため、Cloud Schedulerで定期実行するように設定します。

記述には以下を参考にしました。

* **(公式)App Engineでcronを設定する方法**
[cron.yaml を使用したジョブのスケジューリング](https://cloud.google.com/appengine/docs/standard/go/scheduling-jobs-with-cron-yaml?hl=ja)

* **シンプルに説明したBlog**
[GAEでcronを使って定時実行処理をする - morimo7’s blog](https://morimo7.hatenablog.com/entry/2019/09/09/235126)

`cron.yaml`をプロジェクト内に作成します。
今回は、`/(ルートエンドポイント)`にアクセスするため、`url`部分は空白で記載してます。  
あと、`01:00`と記載していますが、UTCフォーマットなため、日本時間では9時間+の`10:00`に実行するスケジュールになってます。(タイムゾーンは、リージョンに依存するのか？)

```yaml
cron:
  - description: "daily qiita summary log"
    url: ""
    schedule: every day 01:00
```

### App Engineにデプロイ
`cron.yaml`は、アプリケーションとは別にデプロイする必要があります。  
同様にデプロイします。  

```bash
gcloud app deploy cron.yaml --bucket=gs://gae-example
```

### Cloud Schedulerの確認
これで、スケジューリングの設定が完了したので、`Cloud Scheduler`をチェックすると、`APP ENGINEのジョブ`に入ってるので、`今すぐ実行`で動作確認をします。

![cloud scheduler.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/152224/7aa4d0d5-fa82-58da-0bd9-f1fc05d5f82e.png)


その後、DataStoreを確認して、APIで集計した値が格納されているのを確認します。

![Datastore.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/152224/de35e06e-db43-0a75-9c34-eef77f286302.png)


この処理が、cron.yamlで設定したスケジュールで定期実行されます。


# Datastoreの値を返すAPI実装

ここまでで、自動で毎日Qiitaの情報を集計できるようになったため、次はこの情報をポートフォリオ等のアプリケーションで活用できるようにAPIを実装します。具体的には以下の処理を行っています。

1. Datastoreから、対象のデータを取得
2. JavaScriptで扱えるように、DateをISO形式に変換
3. Datastoreは、順不同なため、日付ごとにソートをかける

```python
@app.route('/get_work')
def get_work():

    datastore_client = datastore.Client()

    query = datastore_client.query(kind="Aggregate")
    datastore_response = query.fetch()
    data_list = list(datastore_response)

    data = list([])

    for item in data_list:
        _json = {
            "date": item['date'].isoformat(),
            "likes": int(item['likes']),
            "stocks": int(item['stocks']),
            "views": int(item['views'])
        }

        data.append(_json)

    data = sorted(data, key=lambda x: x['date'])

    return jsonify({"data": data}), 200
```

すると以下のレスポンスが取得できます。

```json
{
    "data": [
        {
            "date": "2021-02-15T07:31:52.400326+00:00",
            "likes": 633,
            "stocks": 325,
            "views": 564128
        },
        {
            "date": "2021-02-16T01:00:34.956754+00:00",
            "likes": 633,
            "stocks": 325,
            "views": 564364
        },
        {
            "date": "2021-02-17T01:01:24.152121+00:00",
            "likes": 634,
            "stocks": 325,
            "views": 564797
        },
    ]
}
```

## 再度 App Engine にデプロイ

```bash
gcloud app deploy --bucket=gs://gae-example
```

`https://<Project名>.an.r.appspot.com/get_work`に、リクエストしてデータが取得できるようになりました。


# まとめ
App Engine + Cloud Schedulerで、Qiitaの活動を集計して、Datastoreに保存、それらを取得するAPIを実装しました。
今後どう活用するか考えます!