GCPでTwitter連携アプリを作る

OAuth認証を行い、Twitterからタイムラインデータを取得するところまでを行います。

事前準備

TwitterAPIをアプリから利用するにはアカウント申請を行い、以下情報を入手する必要があります。
・consumer_key
・consumer_secret

申請方法は以下などを参考にしてください。
・Twitter API 登録 (アカウント申請方法) から承認されるまでの手順まとめ

今回のように、OAuthを利用すると以下画像のようにリクエストが投げられます。
Twitterにアプリを登録する際に入力するCallbackURLsとは、画像に示した点になることに留意してください。

自分の環境

● MacBook-Pro内に構築したコンテナ（Python3.7）
● GoogleCloudPlatform AppEngine
● 作業日：2019年6月

利用するPythonフレームワーク

有名なフレームワーク２つのどちらを採用するかで悩みましたが、今後アプリを作る際にGoogleStorageにデータを保存したかったため、Flaskを採用しました。
● Django
・RDBありきなフレームワーク
・SQLを書かなくてもテーブルが作成できる。とても便利。
・GCPではRDBとしてMySqlが使えるが有料となってしまう
・DBがない場合はSqliteが使えるようだが、GCPにデプロイしたらReadOnlyエラーとなってしまい使用できない
● Flask
・データ保存先を自由に決められる

手順

GCPのチュートリアルで”HelloWorld”を出力するアプリはFlaskを利用して作られています。よって、そのソースを修正し、デプロイを行いました。
・HelloWorldアプリ：python-docs-samples/appengine/standard_python37/hello_world/

１．必要なライブラリをインストール

pip install requests_oauthlib

２．ソース修正（内容後述）
・ main.py

３．利用ライブラリ内容をファイルに出力

pip freeze > requirements.txt

４．デプロイ

gcouud app deploy

ソース修正

ソース内の以下は適宜書き換えてください
・consumer_key
・consumer_secret

main.py

import logging
# ログレベルを DEBUG に変更
logging.basicConfig(level=logging.DEBUG)
# ログの出力名を設定
logger = logging.getLogger('LoggingTest')
# ログのコンソール出力の設定
sh = logging.StreamHandler()
logger.addHandler(sh)
# ログの出力形式の設定
formatter = logging.Formatter('%(asctime)s:%(lineno)d:%(levelname)s:%(message)s')
sh.setFormatter(formatter)
import os
import json
from urllib.parse import parse_qsl
from flask import Flask, jsonify, request,redirect
from requests_oauthlib import OAuth1Session
app = Flask(__name__)
app.config['JSON_AS_ASCII'] = False # Unicodeエスケープ対策
consumer_key = "XXXXXXXXXXXXXXXXXX"
consumer_secret = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
base_url = 'https://api.twitter.com/'
request_token_url = base_url + 'oauth/request_token'
authenticate_url = base_url + 'oauth/authenticate'
access_token_url = base_url + 'oauth/access_token'
base_json_url = 'https://api.twitter.com/1.1/%s.json'
user_timeline_url = base_json_url % ('statuses/user_timeline')
home_timeline_url = base_json_url % ('statuses/home_timeline')

# Twitterの認証画面にリダイレクトさせる
@app.route('/twitter/request_token', methods=['GET'])
def get_twitter_request_token():
   # Twitter Application Management で設定したコールバックURLsのどれか
   oauth_callback = request.args.get('oauth_callback')
   twitter = OAuth1Session(consumer_key, consumer_secret)
   response = twitter.post(
       request_token_url,
       params={'oauth_callback': oauth_callback}
   )
   request_token = dict(parse_qsl(response.content.decode("utf-8")))
   # リクエストトークンから認証画面のURLを生成
   authenticate_endpoint = '%s?oauth_token=%s' \
       % (authenticate_url, request_token['oauth_token'])
   request_token.update({'authenticate_endpoint': authenticate_endpoint})
   return redirect(authenticate_endpoint)

# アクセストークンを取得し、ユーザのタイムラインを返却する
@app.route('/twitter/access_token', methods=['GET'])
def get_twitter_access_token():
   ## アクセストークン取得処理
   oauth_token = request.args.get('oauth_token')
   oauth_verifier = request.args.get('oauth_verifier')
   twitter = OAuth1Session(
       consumer_key,
       consumer_secret,
       oauth_token,
       oauth_verifier,
   )
   response = twitter.post(
       access_token_url,
       params={'oauth_verifier': oauth_verifier}
   )
   access_token = dict(parse_qsl(response.content.decode("utf-8")))
   logger.debug(access_token)
   # タイムライン取得処理
   params = {
       'user_id': access_token['user_id'],
       'exclude_replies': True,
       'count': 20,
       'trim_user': False,
   }
   twitter = OAuth1Session(
       consumer_key,
       consumer_secret,
       access_token['oauth_token'],
       access_token['oauth_token_secret'],
   )
   timeline = twitter.get(home_timeline_url, params=params)
   results = json.loads(timeline.text.encode())
   access_token = dict(parse_qsl(timeline.content.decode("utf-8")))
   tweetList = []
   for tweet in results:
       print(tweet['text'])
       tweetList.append(tweet['text'])
   return jsonify(tweetList)

if __name__ == "__main__":
   port = os.environ.get('PORT', 8080)
   app.run(
       host='0.0.0.0',
       port=port,
       debug=True,
   )

・requrement.txt（例）

astroid==2.2.5
certifi==2019.3.9
chardet==3.0.4
Click==7.0
Flask==1.0.2
idna==2.8
isort==4.3.18
itsdangerous==1.1.0
Jinja2==2.10.1
lazy-object-proxy==1.4.0
MarkupSafe==1.1.1
mccabe==0.6.1
oauthlib==3.0.1
pylint==2.3.1
requests==2.21.0
requests-oauthlib==1.2.0
six==1.12.0
typed-ast==1.3.5
urllib3==1.24.3
Werkzeug==0.15.2
wrapt==1.11.1

後は以下にアクセスし、動作を確認してみてください。
・{GCPのアプリドメイン}/twitter/request_token

参考

・[Python] OAuth認証でTwitter連携/ログインを実装する

＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝
「役に立った！」「時間が削減に繋がった」という方、
よろしければ、以下からサポートよろしくお願いします。
投げ銭ページ
＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝