Gmail API を使ってメールを取得すると結構大変だった話（前編）

link管理

链接快照平台

输入网页链接，自动生成快照
标签化管理网页链接

Gmail APIの認証について

Gmail APIは認証と認可にOAuth2.0を採用しています。公式ドキュメント
公式ドキュメントの内容を要約すると、以下の手順でGmail APIを呼ぶことができます。

一見シンプルですが、注意点がいくつかあります。

一つ目は、1.の認証情報についてです。
おそらくサービスアカウントのプライベートキーを使用するのが最も簡単だと思いますが、今回は使用できません。なぜなら、Gmail APIを呼ぶには、対象のGoogleアカウント本人の認証情報が必要だからです。
この記事では、自前のOAuthクライアントIDを立ててそのクライアントシークレットを使用します。

二つ目は、アクセストークンの更新についてです。
2.で取得できるアクセストークンの有効期限は1時間しかありません。（公式ドキュメントには有効期限について具体的に書かれていませんが、この記事の方法で取得したアクセストークンのexpiryはどれも1時間後でした）
したがって、Cloud Functionsにデプロイしたアプリケーションをメンテナンスなしで動かすためには、アクセストークンを更新する処理が必須になります。
幸い取得したアクセストークンのJSONにはリフレッシュトークンが含まれているので、これを使って更新します。

Google Cloud Platform設定

APIを有効化

まずは使用するAPIをGoogle Cloud Platformから有効化する必要があります。
今回使用するAPIは以下の3つです。

from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from googleapiclient.discovery import build from googleapiclient.errors import HttpError ## スコープを変更した場合はtoken.jsonを削除してください SCOPES = ['<https://www.googleapis.com/auth/gmail.readonly>'] def main(): """Shows basic usage of the Gmail API. Lists the user's Gmail labels. creds = None # 認証フローが初めて完了したときに自動的にtoken.jsonが作成されます if os.path.exists('token.json'): creds = Credentials.from_authorized_user_file('token.json', SCOPES) # 使用可能な認証情報がない場合は、ユーザーにログインを要求します if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) else: flow = InstalledAppFlow.from_client_secrets_file( 'credentials.json', SCOPES) creds = flow.run_local_server(port=0) # 新しい認証情報をtoken.jsonに保存します。 with open('token.json', 'w') as token: token.write(creds.to_json()) # Gmail API呼び出し service = build('gmail', 'v1', credentials=creds) results = service.users().labels().list(userId='me').execute() labels = results.get('labels', []) if not labels: print('No labels found.') return print('Labels:') for label in labels: print(label['name']) except HttpError as error: # TODO(developer) - 適切にエラーを処理してください print(f'An error occurred: {error}') if __name__ == '__main__': main()

OAuthクライアントIDを作成の章でダウンロードした認証情報（credentials.json）をquickstart.pyと同じディレクトリに置いておきましょう。もしcredentials.jsonを別のディレクトリに置く場合はコード中のパスを書きかえてください。

プログラムを実行

最後にローカルでプログラムを実行しましょう。

python quickstart.py

実行するとブラウザが立ち上がり、OAuth同意画面が開かれます。
場合によっては以下のような警告画面が表示されることもあります。これはGoogleによって検証されていないという警告ですが、自分で作った同意画面なので問題ありません。無視して進みましょう。

「詳細を表示」をクリックし、「[アプリ名]（安全ではないページ）に移動」をクリックします。

from bs4 import BeautifulSoup
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from google.cloud import storage
import base64
import json
# OAuth 同意画面で設定したSCOPEに合わせて追加・変更してください
SCOPES = [
    "https://www.googleapis.com/auth/gmail.readonly",
# Cloud Storage のバケット名に置き換えてください
BUCKET_NAME = "bucket_name"
TOKEN_FILE_NAME = "token.json"
# メールを取得するGoogleアカウントに置き換えてください
EMAIL_ADDRESS = "[email protected]"
def main(arg):
    storage_client = storage.Client()
    bucket = storage_client.bucket(BUCKET_NAME)
    token_blob = storage.Blob(TOKEN_FILE_NAME, bucket)
    creds = None
    # Cloud Storage 上の token.json を使って認証
    if token_blob.exists():
        token_str = token_blob.download_as_string()
        token_json = json.loads(token_str)
        creds = Credentials.from_authorized_user_info(token_json, SCOPES)
    if not creds or not creds.valid:
        # token.json の有効期限が切れていたらリフレッシュ
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        # token.json が invalid の場合の処理は省略
        # 新しい token.json 情報を Cloud Storage にアップロード
        with open("/tmp/token.json", 'w') as token:
            token.write(creds.to_json())
        token_blob.upload_from_filename(filename="/tmp/token.json")
        # users.messages.list を呼んでメールの一覧を取得
        service = build('gmail', 'v1', credentials=creds)
        result = service.users().messages().list(userId='me').execute()
        messages = result.get('messages')
        for msg in messages:
            # message_id ごとに users.messages.get を呼んでメールの内容を取得
            txt = service.users().messages().get(
                userId='me', id=msg['id']).execute()
            payload = txt['payload']
            headers = payload['headers']
            # header から送信者と件名の情報を取得
            for d in headers:
                if d['name'] == 'Subject':
                    subject = d['value']
                if d['name'] == 'From':
                    sender = d['value']
            # 本文を base64 でデコード
            parts = payload.get('parts')[0]
            data = parts['body']['data']
            data = data.replace("-", "+").replace("_", "/")
            decoded_data = base64.b64decode(data)
            print("dacoded_data", decoded_data)
            # BeautifulSoup ライブラリで lxml をパース
            soup = BeautifulSoup(decoded_data, "lxml")
            body = soup.body()
            # 件名、送信者、本文をログに出力
            print("Subject: ", subject)
            print("From: ", sender)
            print("Message: ", body)
            print('\n')
        return "end"
    except HttpError as error:
        print(f'An error occurred: {error}')

デプロイと実行

プログラムをCloud Functionsにデプロイ

gcloud コマンドを使ってデプロイします。
コマンドを実行する前に、まずは依存関係をまとめたrequirements.txtファイルを作成しましょう。

google-api-python-client google-auth-httplib2 google-auth-oauthlib google-cloud-storage

requirements.txtはmain.pyと同じディレクトリに置いてください。 別のディレクトリにあるとgcloudコマンドが失敗します。
main.pyとrequirements.txtを準備できたら、以下のコマンドでデプロイを行います。各オプションの説明は公式ドキュメントを参照してください。

gcloud functions deploy main \
  --source=./src \
  --region=asia-northeast1 \
  --entry-point=main \
  --trigger-http \
  --allow-unauthenticated \
  --runtime=python39 \

デプロイに成功したらGCPのCloud Functionsを見に行きましょう。
きちんとデプロイできていることが確認できると思います。