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

pip install --upgrade google-api-python-client google-auth-httplib2 google-auth-oauthlib

from __future__ import print_function

import os.path

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())

    try:
        # 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()

python quickstart.py

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 = "example@gmail.com"

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")

    try:
        # 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}')

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

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