Python×Google Cloud Platform×Line Bot：自動化例行任務で作業効率アップ｜毎日自動サインインスクリプト実装法

PythonとGoogle Cloud Platform、Line Botを活用し、毎日のサインイン作業を自動化。手動作業の負担を減らし、効率的に報酬獲得を実現する具体的手順を解説します。

公開日 2021年02月20日更新日 2024年04月13日

著者 ZhgChgLi | Senior Software Engineer

40 分読了時間

Python×Google Cloud Platform×Line Bot：自動化例行任務で作業効率アップ｜毎日自動サインインスクリプト実装法

閲覧数: 14,445+

正體中文简体中文 English 日本語

本記事は AI による翻訳をもとに作成されています。表現が不自然な箇所がありましたら、ぜひコメントでお知らせください。

記事一覧

Python+Google Cloud Platform+Line Botを使った定期作業の自動実行

サインイン報酬アプリを例に、毎日自動サインインスクリプトを作成する

$Photo by [Paweł Czerwiński](https://unsplash.com/@pawel_czerwinski?utm_source=unsplash&utm_medium=referral&utm_content=creditCopyText){:target="_blank"}$

Photo by Paweł Czerwiński

起源

ずっとPythonで小さなツールを作る習慣があります。真面目なものでは、仕事で自動的にデータをスクレイピングしてレポートを作成したり、不真面目なものでは、スケジュールで自動的に欲しい情報を取得したり、手動で行っていた作業をスクリプトに任せたりしています。

ずっと「自動化」というと、私は単純にパソコンを一台立ち上げて Python スクリプトをずっと動かしていました。利点は簡単で便利なことですが、欠点はネットに繋がった機器が必要なことです。ラズベリーパイでも微量の電気代やネット代がかかりますし、遠隔で起動や停止を操作することも難しい（実際は可能ですが面倒です）。今回は仕事の合間に無料かつクラウド上でできる方法を調べてみました。

目標

Python スクリプトをクラウドに移行して実行し、定期的に自動実行し、ネット経由でオン／オフ可能にする。

本記事では、私が考えたちょっとした工夫を使い、チェックイン報酬型のアプリ向けに自動チェックインを行うスクリプトの例を紹介します。これにより毎日自動でチェックインができ、わざわざアプリを開く必要がなく、実行後には通知が届きます。

完了の通知！

本章の順序

Proxyman を使った Man in the middle attack による 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を使います。どちらも無料ですが、Proxymanのほうが使いやすいです。

公式サイト https://proxyman.io/ から Proxyman ツールをダウンロードしてください
ダウンロード後、Proxyman を起動し、Root 証明書をインストールします（HTTPS トラフィックの内容を解析するための中間者攻撃用）。

「Certificate」->「このMacに証明書をインストール」->「インストール済み＆信頼済み」

パソコンの Root 証明書をインストールした後、スマホに切り替える場合：

「Certificate」->「iOSに証明書をインストール」->「実機デバイス…」

指示に従って、スマートフォンにプロキシを設定し、証明書のインストールと有効化を完了してください。

スマホでAPI通信内容を解析したいAPPを開く

この時、Mac 上の Proxyman に嗅ぎ取ったトラフィックが表示されます。デバイスの IP の下で確認したい APP の API ドメインをクリックしてください。初めて確認する場合は、先に「Enable only this domain」をクリックしないと、その後のトラフィックがデコードされません。

「Enable only this domain」を有効にすると、新たにキャプチャされたトラフィックで元のリクエストやレスポンスの情報が表示されます：

この方法で、APPのサインイン操作時にどのAPIエンドポイントが呼ばれ、どのデータが送られたかを解析し、その情報を記録します。後でPythonで直接リクエストを模倣します。

⚠️注意すべき点として、一部のAPPのトークン情報は変更されることがあり、そのため将来的にPythonでの模擬リクエストが無効になる可能性があります。APPトークンの更新方法についてもよく理解しておく必要があります。

⚠️Proxymanが正常に動作していても、Proxyman経由でAPPがリクエストを送信できない場合、APPがSSLピニングを実装している可能性があります；現時点での解決策はなく、諦めるしかありません。

⚠️アプリ開発者がスニッフィング対策を知りたい場合は、以前の記事を参照してください。

ここでは、以下の情報を取得したと仮定します：

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

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 開発者 にアクセスして有効化してください
Provider を作成する

「Messaging API チャンネルを作成」を選択する

次に、基本情報を入力した後、「Create」を押して作成を送信します。

作成後、最初の「Basic settings」タブの下にある「Your user ID」セクションで、自分のユーザーIDを確認できます。

作成後、「Messaging API」タブを選択し、QRコードをスキャンしてボットを友だちに追加します。

下にスクロールして「Channel access token」セクションを見つけ、「Issue」をクリックしてトークンを発行します。

コピーして生成された Token があれば、その Token を使ってユーザーにメッセージを送信できます。

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" : "ここにあなたのユーザー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', 'DEFAULT VALUE')

他の設定は変更せず、そのまま「保存」->「次へ」をクリックしてください。

実行環境で「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」でPOSTデータを送信してください。そうしないとデータが送信されません：

コード部分に問題がなければ、「requirements.txt」に使用するパッケージの依存関係を入力します：

我們使用「request」這個套件幫我們打 API，此套件不在原生 Python 庫裡面；所以我們要在這裡加上去：

「request」パッケージを使ってAPIを呼び出しますが、このパッケージはPython標準ライブラリに含まれていません。なので、ここで追加する必要があります：

requests>=2.25.1

ここではバージョン ≥ 2.25.1 を指定していますが、指定せずに requests と入力して最新版をインストールしても構いません。

問題がなければ、「デプロイ」をクリックしてデプロイを開始します。

デプロイが完了するまでに約1〜3分かかります。

デプロイ完了後、控えておいた「トリガーURL」から実行して正しく動作するか確認するか、「アクション」->「関数のテスト」を使ってテストを行います。

もし 500 Internal Server Error が表示された場合、プログラムにエラーがあることを意味します。名前をクリックして「ログ」を確認し、原因を特定してください：

UnboundLocalError: ローカル変数 'db' が代入前に参照されました

名前をクリックして入った後、「編集」を押してスクリプト内容を変更できます

テストに問題がなければ完了です！Pythonスクリプトを無事にクラウドに移行できました。

変数に関する補足部分

依照私たちの要件として、サインインAPPのトークンを保存・読み取りできる場所が必要です。トークンは無効になる可能性があるため、再取得して次回実行時に使用できるように書き込む必要があります。

外部から動的に変数をスクリプトに渡す方法は以下の通りです：

[Read Only] 前述した実行時環境変数
[Temp] Cloud Functions は実行時にファイルの読み書きが可能な /tmp ディレクトリを提供していますが、処理終了後に削除されます。詳細は公式ドキュメントをご参照ください。
[Read Only] GET/POST データ送信
[Read Only] 添付ファイルを追加

プログラム内で相対パス ./ を使うと読み取れますが、読み取りのみで動的な変更はできません。変更する場合はコンソール上で行い、再デプロイが必要です。

読み取りや動的な変更を行いたい場合は、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()

もし 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」を展開して認証設定を行ってください。

すべて入力したら、「作成」を押します。

作成に成功したら、「今すぐ実行」を選んで正常に動作するかテストできます。

実行結果と前回の実行日を確認できます

⚠️ ご注意ください。実行結果が「失敗」となるのは、web ステータスコードが 400〜500 の場合、または Python プログラムにエラーがある場合のみです。

完了！

私たちは、定期的なタスクの Python スクリプトをクラウドにアップロードし、自動スケジュール設定で自動実行する目標を達成しました。

料金体系

まだ非常に重要な部分があります。それは料金体系です。Google CloudやLinebotは完全無料のサービスではないため、料金体系を理解することが重要です。そうしないと、小さなスクリプトのために多額の費用を支払うことになり、むしろパソコンをずっと起動して実行させておいたほうが良い場合もあります。