Python+Google Cloud Platform+Line Botを使った定期作業の自動実行
签到報酬アプリを例に、毎日自動サインインスクリプトを作成する
Photo by Paweł Czerwiński
起源
私はずっとPythonで小さなツールを作る習慣があります。仕事では自動でデータをスクレイピングしたり、レポートを作成したりする真面目なものもあれば、スケジュールで自動的に情報を取得したり、手動で行う作業をスクリプトに任せたりする気軽なものもあります。
これまで「自動化」というと、私はいつも単純にパソコンを一台用意してPythonスクリプトをずっと動かしていました。利点は簡単で便利なことですが、欠点はネットに繋がった機器が必要で電源も必要なことです。ラズベリーパイでも少量の電気代やネット代がかかりますし、遠隔で起動や停止を操作するのも（実際には可能ですが）面倒です。今回は仕事の合間に、無料でクラウド上で実行する方法を調べてみました。
目標
Pythonスクリプトをクラウドに移行して実行し、定期的に自動実行し、ネット経由でオン/オフ可能にする。
本記事では、私が考えたちょっとした工夫を使って、チェックイン報酬型アプリの自動チェックインスクリプトを例に挙げます。これにより毎日自動でチェックインができ、わざわざアプリを開く必要がありません。実行後には通知も送られます。
完了通知！
本章の順序
-
Proxyman を使った Man in the middle 攻撃による API スニッフィング
-
Pythonスクリプトを作成し、APPのAPIリクエストを偽造する（サインイン動作をシミュレート）
-
PythonスクリプトをGoogle Cloudにデプロイする
-
Google Cloudで自動スケジュールを設定する
-
敏感な分野に関わるため、本記事ではどのサインイン報酬型アプリかは明かしません。ご自身で応用してご利用ください。
-
もしPythonでの自動実行連携だけ知りたい場合は、前半のMan in the middle攻撃によるAPI解析部分を飛ばして、第3章からご覧ください。
使用したツール
-
Proxyman ：Man in the middle攻撃によるAPIスニッフィング
-
Python ：スクリプトの作成
-
Linebot ：スクリプト実行結果を自分に通知する
-
Google Cloud Function ：Pythonスクリプトのホスティングサービス
-
Google Cloud Scheduler ：自動スケジューリングサービス
1.Proxymanを使ったMan in the middle攻撃によるAPIのスニッフィング
以前、「APPはHTTPSで通信しているのにデータが盗まれた。」という記事を書きました。内容は似ていますが、今回はmitmproxyの代わりにProxymanを使います。同じく無料ですが、より使いやすいです。
-
公式サイト https://proxyman.io/ から Proxyman ツールをダウンロードしてください。
-
ダウンロード後、Proxymanを起動し、Root証明書をインストールします（Man in the middle攻撃でhttpsトラフィックの内容を解析するため）。
「Certificate 」->「このMacに証明書をインストール」->「インストール済み＆信頼済み」
パソコンのRoot証明書をインストールした後、スマホに切り替える場合：
「Certificate 」->「iOSに証明書をインストール」->「実機デバイス…」
指示に従って、スマホにプロキシを設定し、証明書のインストールと有効化を完了してください。
- スマホでAPI通信内容を解析したいアプリを開く
この時、Mac上のProxymanに嗅ぎ取ったトラフィックが表示されます。デバイスのIPの下にある確認したいAPPのAPIドメインをクリックしてください。初めて確認する場合は、まず「Enable only this domain」をクリックしないと、その後のトラフィックが解析されません。
「Enable only this domain」をオンにすると、新たにキャプチャされたトラフィックの元のRequestおよびResponse情報が表示されます：
この方法でAPPのサインイン操作時にどのAPIエンドポイントが呼ばれ、どのようなデータが送信されたかを解析し、その情報を記録しておき、後でPythonで直接リクエストを模倣します。
⚠️注意すべき点として、APPのtoken情報は変更されることがあり、そのため将来的にPythonの模擬リクエストが無効になる可能性があります。APP tokenの交換方法もよく理解しておく必要があります。
⚠️Proxymanが正常に動作していることを確認した上で、Proxymanを経由しているにもかかわらずAPPがリクエストを送信できない場合、APPがSSL Pinningを実装している可能性があります。現時点では解決策がなく、諦めるしかありません。
⚠️アプリ開発者がスニッフィング対策を知りたい場合は、以前の記事を参照してください。
ここでは、取得した情報が以下の通りと仮定します：
POST /usercenter HTTP/1.1
Host: zhgchg.li
Content-Type: application/x-www-form-urlencoded
Cookie: PHPSESSID=dafd27784f94904dd586d4ca19d8ae62
Connection: keep-alive
Accept: */*
User-Agent: (iPhone12,3;iOS 14.5)
Content-Length: 1076
Accept-Language: zh-tw
Accept-Encoding: gzip, deflate, br
AuthToken: 12345
action=checkIn
2. Pythonスクリプトを作成し、APPのAPIリクエストを偽造する（サインイン動作のシミュレーション）
Pythonスクリプトを作成する前に、Postman を使ってパラメータを調整し、どのパラメータが必要か、または有効期限で変わるかを確認できますが、そのままコピーしても問題ありません。
checkIn.py:
import requests
import json
def main(args):
results = {}
try:
data = { "action" : "checkIn" }
headers = { "Cookie" : "PHPSESSID=dafd27784f94904dd586d4ca19d8ae62",
"AuthToken" : "12345",
"User-Agent" : "(iPhone12,3;iOS 14.5)"
}
request = requests.post('https://zhgchg.li/usercenter', data = data, headers = headers)
result = json.loads(request.content)
if result['status_code'] == 200:
return "チェックイン成功！"
else:
return result['message']
except Exception as e:
return str(e)
⚠️
main(args)の args の使い方は後で説明します。ローカルでテストする場合は、直接main(True)としてください。
Requests パッケージを使ってHTTPリクエストを実行します。もし以下のようなエラーが出た場合：
ImportError: No module named requests
まずは pip install requests でパッケージをインストールしてください。
実行結果をLinebotで通知：
この部分は非常にシンプルに作成しており、参考程度で自分にのみ通知しています。
-
Line Developers Console 開発者 にアクセスして有効化してください。
-
プロバイダーを作成する
- 「Messaging API チャンネルを作成」を選択してください
次のステップで基本情報を入力し、「Create」を押して作成を送信します。
- 作成後、最初の「Basic settings」タブの下にある「Your user ID」セクションで、自分のユーザーIDを確認できます。
- 作成したら、「Messaging API」タブを選択し、QRコードをスキャンしてボットを友だちに追加します。
- 下にスクロールして「Channel access token」セクションを見つけ、「Issue」をクリックしてトークンを発行します。
- コピーして生成されたトークンがあれば、そのトークンを使ってユーザーにメッセージを送信できます。
User IdとTokenがあれば、自分にメッセージを送ることができます。
他の機能を作らないため、python line sdkはインストールせずに、直接httpリクエストを送信します。
以前のPythonスクリプトと連携して…
checkIn.py:
import requests
import json
def main(args):
results = {}
try:
data = { "action" : "checkIn" }
headers = { "Cookie" : "PHPSESSID=dafd27784f94904dd586d4ca19d8ae62",
"AuthToken" : "12345",
"User-Agent" : "(iPhone12,3;iOS 14.5)"
}
request = requests.post('https://zhgchg.li/usercenter', data = data, headers = headers)
result = json.loads(request.content)
if result['status_code'] == 200:
sendLineNotification("チェックイン成功！")
return "チェックイン成功！"
else:
sendLineNotification(result['message'])
return result['message']
except Exception as e:
sendLineNotification(str(e))
return str(e)
def sendLineNotification(message):
data = {
"to" : "ここにあなたのUser IDを入れてください",
"messages" : [
{
"type" : "text",
"text" : message
}
]
}
headers = {
"Content-Type" : "application/json",
"Authorization" : "ここにチャネルアクセストークンを入れてください"
}
request = requests.post('https://api.line.me/v2/bot/message/push',json = data, headers = headers)
通知が正常に送信されたか確認する：
成功しました！
ちょっとしたエピソードですが、通知部分はもともとGmailのSMTPを使ってメールで送ろうと思っていましたが、Google Cloudにデプロイしたところ使えないことが分かりました…
3. PythonスクリプトをGoogle Cloudに移行する
前半の基本説明が終わり、ここから本題に入ります；Pythonスクリプトをクラウドに移行します。
この部分は最初Google Cloud Runを使おうと思いましたが、使ってみると複雑すぎて実際に調べるのが面倒だったので、私のニーズには機能が多すぎました。そこで Google Cloud Function のサーバーレスプランを使いました。実際にはサーバーレスのウェブサービス構築によく使われています。
-
Google Cloud を使ったことがない方は、まず コンソール にアクセスして、プロジェクトを作成し、請求情報を設定してください。
-
プロジェクトコンソールのホーム画面で、リソースの部分にある「Cloud Functions」をクリックします。
- 上部の「関数を作成」を選択してください
- 基本情報を入力してください
⚠️「 **トリガーURL」をメモしてください
リージョン選択：
-
US-WEST1、US-CENTRAL1、US-EAST1は Cloud Storage サービスの無料枠を利用できます。 -
asia-east2（香港）は比較的近いですが、わずかなCloud Storage料金が発生します。
⚠️Cloud Functionsを作成する際は、Cloud Storageにコードを保存する必要があります。
⚠️詳細な料金体系については文末をご参照ください。
トリガー条件の選択： HTTP
認証： 要件に応じて、外部からリンクをクリックしてスクリプトを実行できるように「未認証の呼び出しを許可」を選択しました。認証が必要な場合は、後でSchedulerサービスの設定も必要です。
変数、ネットワークおよび高度な設定は変数で設定し、Pythonで使用できます（パラメータが変更されてもPythonコードを修正する必要がありません）：
Pythonでの呼び出し方法：
import os
def main(request):
return os.environ.get('test', 'デフォルト値')
他の設定は変更せず、そのまま「保存」->「次へ」。
- 実行環境は「Python 3.x」を選択し、作成したPythonスクリプトを貼り付け、エントリーポイントを「main」に変更してください。
補足 main(args)、前述の通り、このサービスは主にサーバーレスウェブ用です。そのため、argsは実際にはRequestオブジェクトで、そこからHTTP GETクエリやHTTP POSTボディのデータを取得できます。具体的な方法は以下の通りです：
GETクエリ情報の取得：
request_args = args.args
example: ?name=zhgchgli => request_args = [“name”:”zhgchgli”]
POSTボディデータの取得：
request_json = request.get_json(silent=True)
example: name=zhgchgli => request_json = [“name”:”zhgchgli”]
PostmanでPOSTをテストする場合は、「Raw+JSON」でデータを送信してください。そうしないとデータが送信されません：
- コード部分に問題がなければ、「requirements.txt」に使用するパッケージの依存関係を入力します：
私たちはAPIを呼び出すために「request」というパッケージを使用しますが、このパッケージは標準のPythonライブラリには含まれていません。なので、ここで追加する必要があります：
requests>=2.25.1
ここではバージョン ≥ 2.25.1 を指定していますが、指定せずに requests とだけ入力して最新版をインストールしても構いません。
- 問題がなければ、「デプロイ」をクリックしてデプロイを開始します。
デプロイ完了までに約1〜3分かかります。
- デプロイ完了後、控えておいた「トリガーURL」から実行して正しく動作しているか確認するか、「アクション」->「関数のテスト」でテストを行ってください。
もし 500 Internal Server Error が表示された場合は、プログラムにエラーがあることを意味します。名前をクリックして「ログ」を確認し、原因を特定してください：
UnboundLocalError: ローカル変数 'db' が代入前に参照されました
- 名前をクリックしてから「編集」を押すと、スクリプトの内容を変更できます。
テストに問題がなければ完了です！Pythonスクリプトを無事にクラウドへ移行できました。
変数に関する補足事項
私たちのニーズに応じて、サインインAPPのtokenを保存・読み取りできる場所が必要です。tokenは無効になる可能性があるため、再取得して次回実行時に使用するために書き込む必要があります。
外部から動的に変数をスクリプトに渡す方法は以下の通りです：
-
[Read Only] 前述した実行時環境変数
-
[Temp] Cloud Functions は実行時にファイルの読み書きが可能な /tmp ディレクトリを提供しますが、実行終了後に削除されます。詳細は公式ドキュメントをご参照ください。
-
[読み取り専用] GET/POST データ送信
-
[読み取り専用] 添付ファイルの挿入
プログラム内で相対パス ./ を使うと読み取れますが、読み取りのみで動的な変更はできません。変更する場合はコンソールで行い、再デプロイが必要です。
読み取りや動的な変更を行いたい場合は、Cloud SQL、Google Storage、Firebase Cloud Firestoreなどの他のGCPサービスと連携する必要があります。
- [Read & Write] ここでは Firebase Cloud Firestore を選択しました。現在、このプランのみが無料枠を利用可能だからです。
入門手順 に従って Firebase プロジェクトを作成した後、Firebase コンソールにアクセスしてください：
左側メニューで「 Cloud Firestore 」->「 コレクションを追加 」を見つけます。
入力コレクション ID。
入力データの内容。
コレクションには複数のドキュメントが含まれ、それぞれのドキュメントは独自のフィールド内容を持つことができます。非常に柔軟に使用可能です。
Pythonでの使用方法：
まずは GCPコンソール -> IAMと管理 -> サービスアカウント にアクセスし、以下の手順で認証用の秘密鍵ファイルをダウンロードしてください：
まずアカウントを選択してください：
下方「新增金鑰」->「新しいキーを作成」
「JSON」を選択してファイルをダウンロードしてください。
この JSON ファイルを Python プロジェクトのディレクトリに置いてください。
ローカル開発環境の場合：
pip install --upgrade firebase-admin
firebase-admin パッケージをインストールします。
Cloud Functions 上で requirements.txt に firebase-admin を追加する必要があります。
環境が整ったら、先ほど追加したデータを読み取ってみましょう：
firebase_admin.py:
import firebase_admin
from firebase_admin import credentials
from firebase_admin import firestore
if not firebase_admin._apps:
cred = credentials.Certificate('./身份驗證.json')
firebase_admin.initialize_app(cred)
# initialize_app を重複して実行すると以下のエラーが発生します
# providing an app name as the second argument. In most cases you only need to call initialize_app() once. But if you do want to initialize multiple apps, pass a second argument to initialize_app() to give each app a unique name.
# そのため、安全のために initialize_app を実行する前に初期化済みか確認しています
db = firestore.client()
ref = db.collection(u'example') # コレクション名
stream = ref.stream()
for data in stream:
print("id:"+data.id+","+data.to_dict())
もし Cloud Functions 上であれば、認証用の JSON ファイルをアップロードする必要はなく、使用時に接続コードを以下のように変更して使うこともできます：
cred = credentials.ApplicationDefault()
firebase_admin.initialize_app(cred, {
'projectId': project_id,
})
db = firestore.client()
# Firestoreクライアントを初期化
もし
Failed to initialize a certificate credential.と表示された場合は、認証用のJSONファイルが正しいかどうか確認してください。
追加や削除などの操作については、公式ドキュメント を参照してください。
4. Google Cloudで自動スケジュールを設定する
スクリプトができたら、次は自動実行させて最終目標を達成しましょう。
-
Google Cloud Scheduler コンソールのホームページに移動してください
-
上部の「ジョブを作成」
- 作業の基本情報を入力してください
実行頻度： crontab の入力方法と同じです。crontab の文法に不慣れな場合は、crontab.guru という便利なサイト を直接利用してください：
設定した構文の実際の意味をわかりやすく翻訳します。（next をクリックすると次回実行時間を確認できます）
ここでは
15 1 * * *に設定しています。なぜなら、サインインは毎日一回だけ実行すればよく、毎日午前1時15分に実行するためです。
URL部分： 先ほど控えた「トリガーURL」を入力してください
タイムゾーン： 「台湾」と入力し、台北標準時を選択してください
HTTP メソッド： 前述のPythonコード通り、Getで問題ありません。
もし前に「認証」を設定している場合は、「SHOW MORE」を展開して認証設定を行ってください。
すべて入力したら、「作成」を押します。
- 作成に成功したら、「今すぐ実行」を選択して正常に動作するかテストできます。
- 実行結果や最終実行日を確認できます
⚠️ ご注意ください。実行結果が「失敗」となるのは、ウェブのステータスコードが400〜500の範囲、またはPythonプログラムにエラーがある場合のみです。
大成功！
私たちはルーチンタスクのPythonスクリプトをクラウドにアップロードし、自動スケジュールを設定して自動実行する目標を達成しました。
料金体系
もう一つ重要なポイントは料金体系です。Google CloudやLinebotは完全無料のサービスではないため、料金の仕組みを理解することが大切です。さもなければ、小さなスクリプトのために多額の費用を支払うことになり、むしろパソコンをずっと起動させておく方が良いでしょう。
Linebot
参考 公式料金 情報では、1ヶ月500件まで無料です。
Google Cloud Functions
参考 公式料金 の情報によると、月間200万回の呼び出し、400,000 GB秒および200,000 GHz秒の計算時間、5 GBのインターネット送信トラフィックが無料です。
Google Firebase Cloud Firestore
参考 公式料金 の情報では、1 GB の容量、月間 10 GB のトラフィック、1日あたり 50,000 回の読み取り、20,000 回の書き込み/削除が含まれています。軽量な使用には十分です！
Google Cloud Scheduler
参考 公式料金 の情報によると、アカウントごとに3つの無料ジョブを設定できます。
スクリプトには、上記の無料使用量で十分すぎるほどです！
Google Cloud Storage は条件付きで無料
あちこち避けても、料金が発生する可能性のあるサービスは避けられません。
Cloud Functions を作成すると、自動的に2つの Cloud Storage バケットが作成されます：
もし先ほど Cloud Functions で US-WEST1、US-CENTRAL1、または US-EAST1 のいずれかのリージョンを選択した場合、無料利用枠を利用できます：
私は US-CENTRAL1 を選択しました。最初の Cloud Storage 実体の地域が US-CENTRAL1 であることは間違いありませんが、2つ目は 米国内の複数の地域 と記載されています；これは料金が発生すると予想しています。
参考 公式価格 情報によると、ホストの地域によって価格が異なります。
コードはそれほど多くないので、月額最低料金はおそらく0.0X0円程度だと思います（？
⚠️上記の情報は2021/02/21時点の記録であり、実際の価格は現在のものを優先してください。参考程度にご利用ください。
料金予算管理通知
念のため…もし無料使用量を超えて課金が始まった場合、通知を受け取りたいです。プログラムのエラーなどで請求額が急増しても気づかない事態を避けるためです。。。
-
コンソール にアクセスしてください。
-
「課金機能」カードを見つけてください：
「 詳細な請求履歴を見る 」をクリックしてください。
- 左側メニューを展開し、「 予算とアラート 」機能に入る
- 上部の「 設定予算 」をクリックしてください
- カスタム名を入力してください
次のステップ。
- 金額、入力欄に「目標金額」を設定できます。$1や$10などを入力でき、小さな支出にはあまりお金をかけたくありません。
次のステップ。
アクションでは、予算が何パーセントに達したときに通知をトリガーするかを設定できます。
チェック 「請求書管理者とユーザーにメールでアラートを送信する」をオンにすると、条件が発生した際にすぐに通知を受け取れます。
「完了」をクリックして保存を送信します。
予算を超えた場合、すぐに通知を受け取ることができ、さらなる費用の発生を防げます。
まとめ
人間のエネルギーは有限です。現代の情報洪水の中で、あらゆるプラットフォームやサービスが私たちの限られたエネルギーを搾り取ろうとしています。もし自動化スクリプトで日常生活の負担を分散できれば、小さな積み重ねが大きな力となり、重要なことにより多くのエネルギーを集中できるようになります！
関連記事
自動化に関する最適化のご依頼も歓迎します。こちらからご連絡ください。よろしくお願いします。
Post MediumからZMediumToMarkdownによって変換されました。
コメント