Python を使い、Gmail API 経由で Gmail の送受信を行う

はじめに

Gmail にはご存知の通りGmail API があり、SMTP や POP3 を使うことなく、検索機能など Gmail 独自の機能を活用することができます。
この記事では、Python を使って Gmail API の有効化から、スクリプトの作成、実行までの手順をステップバイステップで紹介します。

Gmail API を利用するまでに、プロジェクトを作成したり、APIを有効化したり、スコープを設定したり、認証情報を作成したりと、準備が割と大変なので、備忘録として書きました。

• 2020年4月時点の Gmail API の内容です。
• Python 3.7.x を想定しています。

利用前の準備

Goole API を利用するプロジェクトを作成する

利用する Gmail アカウントの Gmail を開いたブラウザで、 Google Cloud Platform コンソール （ https://console.cloud.google.com/ ）を開きます。

下記の画面が出るので規約等に同意し、「同意して実行」をクリックします。

「プロジェクトの選択」から「新しいプロジェクト」を選択

プロジェクト名を設定して「作成」をクリック

作成できると、下記のような画面が表示されます

Gmail APIを有効化する

「ライブラリ」をクリック

「APIとサービスを検索」から「Gmail API」を検索する

表示された「Gmail API」をクリックする

「有効にする」をクリック

OAuth同意画面を作成する

「APIとサービス」から「OAuth同意画面」をクリック

「外部」を選択し、「作成」をクリック

「アプリケーション名」に適切な名前を設定し、「スコープを追加」をクリック

下図のようにスコープを選択し、「追加」をクリック

「保存」をクリック

認証情報を作成する

2020年6月14日現在、下記の手順でウィザードを実行すると、エラーが発生するようです。下記手順で回避できました。

「認証情報」を選択し「認証情報を作成」をクリックして表示されるメニューから「OAuthクライアントID」をクリック

「アプリケーションの種類」で「デスクトップアプリ」を選択

作成完了のダイアログが表示されるので「OK」をクリック

あとは、一覧画面から認証情報をダウンロード（client_id.json）します。

ウィザードを使う

「認証情報」を選択し「認証情報を作成」をクリックして表示されるメニューから「ウィザードで選択」をクリック

「プロジェクトへの認証情報の追加」画面で、下図のように設定し、「必要な認証情報」をクリック

「OAuthクライアントIDを作成」をクリック

認証情報をダウンロード（client_id.json）し、「完了」をクリック

完了ボタンを押すと下記のような画面に遷移します

ダウンロードした認証情報ファイル client_id.json は後ほど Python スクリプトから利用します。

補足：認証情報について

認証情報ではAPIキー、OAuth2.0クライアントID、サービスアカウントキーのいずれかを作成できます。ユーザーデータを扱う場合は、OAuth2.0クライアントID を使うことが一般的のようです。
Google Cloud ドキュメント 認証の概要ページに、それぞれの認証方法の用途について解説されています。

クレデンシャルの Refresh Token の有効期限については、下記に記載があります。一度認証して Refresh Token が払い出されると、6ヶ月放置するか、50以上のトークンが払い出されない限り、期限切れになることはないようです。RPA などの自動化の仕組みでも許容できそうです。
https://developers.google.com/identity/protocols/OAuth2#expiration

You must write your code to anticipate the possibility that a granted refresh token might no longer work. A refresh token might stop working for one of these reasons:

* The user has revoked your app's access.
* The refresh token has not been used for six months.
* The user changed passwords and the refresh token contains Gmail scopes.
* The user account has exceeded a maximum number of granted (live) refresh tokens.
* There is currently a limit of 50 refresh tokens per user account per client. If the limit is reached, creating a new refresh token automatically invalidates the oldest refresh token without warning. This limit does not apply to service accounts.

Python スクリプトの準備

ようやく準備が整いましたので、これから実行する Python スクリプトを用意します。

Python 環境の準備

（この部分はすでにすんでいる方も多いと思います）

Python のインストール

ここ にしたがってインストールを行います。3.7.x をインストールしてください。

pipenv のインストール

ここ の手順に従ってインストールします。

Python スクリプトの作成

適当なディレクトリを用意し、python スクリプトを準備します。先の手順で保存した client_id.json も同じディレクトリに保存します。Gmail API のサンプルをつなぎ合わせました

Gmail API のリファレンスに、サンプルコードが載ってます。ここまでくると利用しやすいですね！

• Users.messages: list
• Users.labels: list

gmail_credential.py

import os
import pickle
from google.auth.transport.requests import Request
from google_auth_oauthlib.flow import InstalledAppFlow


# Gmail APIのスコープを設定
SCOPES = [
    "https://www.googleapis.com/auth/gmail.compose",
    "https://www.googleapis.com/auth/gmail.readonly",
    "https://www.googleapis.com/auth/gmail.labels",
    "https://www.googleapis.com/auth/gmail.modify",
]


def get_credential():
    """
    アクセストークンの取得

    カレントディレクトリに pickle 形式でトークンを保存し、再利用できるようにする。（雑ですみません。。）
    """
    creds = None
    if os.path.exists("token.pickle"):
        with open("token.pickle", "rb") as token:
            creds = pickle.load(token)
    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("client_id.json", SCOPES)
            # creds = flow.run_local_server()
            creds = flow.run_console()
        with open("token.pickle", "wb") as token:
            pickle.dump(creds, token)
    return creds

listmail.py

"""
list GMail Inbox.

Usage:
  listmail.py <query> <tag> <count>
  listmail.py -h | --help
  listmail.py --version

Options:
  -h --help     Show this screen.
  --version     Show version.
"""
import pickle
import base64
import json
import io
import csv
import os.path
from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
import base64
from email.mime.text import MIMEText
from apiclient import errors
import logging
from docopt import docopt
from gmail_credential import get_credential

logger = logging.getLogger(__name__)


def list_labels(service, user_id):
    """
    label のリストを取得する
    """
    labels = []
    response = service.users().labels().list(userId=user_id).execute()
    return response["labels"]


def decode_base64url_data(data):
    """
    base64url のデコード
    """
    decoded_bytes = base64.urlsafe_b64decode(data)
    decoded_message = decoded_bytes.decode("UTF-8")
    return decoded_message


def list_message(service, user_id, query, label_ids=[], count=3):
    """
    メールのリストを取得する

    Parameters
    ----------
    service : googleapiclient.discovery.Resource
        Gmail と通信するためのリソース
    user_id : str
        利用者のID
    query : str
        メールのクエリ文字列。 is:unread など
    label_ids : list
        検索対象のラベルを示すIDのリスト
    count : str
        リターンするメール情報件数の上限

    Returns
    ----------
    messages : list
        id, body, subject, from などのキーを持った辞書データのリスト
    """
    messages = []
    try:
        message_ids = (
            service.users()
            .messages()
            .list(userId=user_id, maxResults=count, q=query, labelIds=label_ids)
            .execute()
        )

        if message_ids["resultSizeEstimate"] == 0:
            logger.warning("no result data!")
            return []

        # message id を元に、message の内容を確認
        for message_id in message_ids["messages"]:
            message_detail = (
                service.users()
                .messages()
                .get(userId="me", id=message_id["id"])
                .execute()
            )
            message = {}
            message["id"] = message_id["id"]
            # 単純なテキストメールの場合
            if 'data' in message_detail['payload']['body']:
                message["body"] = decode_base64url_data(
                    message_detail["payload"]["body"]["data"]
                )
            # html メールの場合、plain/text のパートを使う
            else:
                parts = message_detail['payload']['parts']
                parts = [part for part in parts if part['mimeType'] == 'text/plain']
                message["body"] = decode_base64url_data(
                    parts[0]['body']['data']
                    )
            # payload.headers[name: "Subject"]
            message["subject"] = [
                header["value"]
                for header in message_detail["payload"]["headers"]
                if header["name"] == "Subject"
            ][0]
            # payload.headers[name: "From"]
            message["from"] = [
                header["value"]
                for header in message_detail["payload"]["headers"]
                if header["name"] == "From"
            ][0]
            logger.info(message_detail["snippet"])
            messages.append(message)
        return messages

    except errors.HttpError as error:
        print("An error occurred: %s" % error)


def remove_labels(service, user_id, messages, remove_labels):
    """
    ラベルを削除する。既読にするために利用(is:unread ラベルを削除すると既読になる）
    """
    message_ids = [message["id"] for message in messages]
    labels_mod = {
        "ids": message_ids,
        "removeLabelIds": remove_labels,
        "addLabelIds": [],
    }
    # import pdb;pdb.set_trace()
    try:
        message_ids = (
            service.users()
            .messages()
            .batchModify(userId=user_id, body=labels_mod)
            .execute()
        )
    except errors.HttpError as error:
        print("An error occurred: %s" % error)


# メイン処理
def main(query="is:unread", tag="daily_report", count=3):
    creds = get_credential()
    service = build("gmail", "v1", credentials=creds, cache_discovery=False)
    # ラベル一覧
    labels = list_labels(service, "me")
    target_label_ids = [label["id"] for label in labels if label["name"] == tag]
    # メール一覧 [{'body': 'xxx', 'subject': 'xxx', 'from': 'xxx'},]
    messages = list_message(service, "me", query, target_label_ids, count=count)
    # unread label
    unread_label_ids = [label["id"] for label in labels if label["name"] == "UNREAD"]
    # remove labels form messages
    remove_labels(service, "me", messages, remove_labels=unread_label_ids)
    logger.info(json.dumps(messages, ensure_ascii=False))
    if messages:
        return json.dumps(messages, ensure_ascii=False)
    else:
        return None


# プログラム実行部分
if __name__ == "__main__":
    arguments = docopt(__doc__, version="0.1")
    query = arguments["<query>"]
    tag = arguments["<tag>"]
    count = arguments["<count>"]
    logging.basicConfig(level=logging.DEBUG)

    messages_ = main(query=query, tag=tag, count=count)
    print(messages_)

送信スクリプトはこんな感じです。
※ Users.messages.send の Python サンプルコードでは、 return {'raw': base64.urlsafe_b64encode(message.as_string())} となっていましたが、それでは動かなかったので修正しています。

sendmail.py

"""
Send E-Mail with GMail.

Usage:
  sendmail.py <sender> <to> <subject> <message_text_file_path>  [--attach_file_path=<file_path>] [--cc=<cc>]
  sendmail.py -h | --help
  sendmail.py --version

Options:
  -h --help     Show this screen.
  --version     Show version. 
  --attach_file_path=<file_path>     Path of file attached to message.
  --cc=<cc>     cc email address list(separated by ','). Default None.
"""
import pickle
import os.path
from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
import base64
from email.mime.base import MIMEBase
from email.mime.text import MIMEText
from email.mime.image import MIMEImage
from email.mime.audio import MIMEAudio
from pathlib import Path

from email.mime.multipart import MIMEMultipart
import mimetypes
from apiclient import errors
from gmail_credential import get_credential
from docopt import docopt
import logging

logger = logging.getLogger(__name__)


def create_message(sender, to, subject, message_text, cc=None):
    """
    MIMEText を base64 エンコードする
    """
    enc = "utf-8"
    message = MIMEText(message_text.encode(enc), _charset=enc)
    message["to"] = to
    message["from"] = sender
    message["subject"] = subject
    if cc:
        message["Cc"] = cc
    encode_message = base64.urlsafe_b64encode(message.as_bytes())
    return {"raw": encode_message.decode()}


def create_message_with_attachment(
    sender, to, subject, message_text, file_path, cc=None
):
    """
    添付ファイルつきのMIMEText を base64 エンコードする
    """
    message = MIMEMultipart()
    message["to"] = to
    message["from"] = sender
    message["subject"] = subject
    if cc:
        message["Cc"] = cc
    # attach message text
    enc = "utf-8"
    msg = MIMEText(message_text.encode(enc), _charset=enc)
    message.attach(msg)

    content_type, encoding = mimetypes.guess_type(file_path)

    if content_type is None or encoding is not None:
        content_type = "application/octet-stream"
    main_type, sub_type = content_type.split("/", 1)
    if main_type == "text":
        with open(file_path, "rb") as fp:
            msg = MIMEText(fp.read(), _subtype=sub_type)
    elif main_type == "image":
        with open(file_path, "rb") as fp:
            msg = MIMEImage(fp.read(), _subtype=sub_type)
    elif main_type == "audio":
        with open(file_path, "rb") as fp:
            msg = MIMEAudio(fp.read(), _subtype=sub_type)
    else:
        with open(file_path, "rb") as fp:
            msg = MIMEBase(main_type, sub_type)
            msg.set_payload(fp.read())
    p = Path(file_path)
    msg.add_header("Content-Disposition", "attachment", filename=p.name)
    message.attach(msg)

    encode_message = base64.urlsafe_b64encode(message.as_bytes())
    return {"raw": encode_message.decode()}


def send_message(service, user_id, message):
    """
    メールを送信する

    Parameters
    ----------
    service : googleapiclient.discovery.Resource
        Gmail と通信するためのリソース
    user_id : str
        利用者のID
    message : dict
        "raw" を key, base64 エンコーディングされた MIME Object を value とした dict

    Returns
    ----------
    なし
    """
    try:
        sent_message = (
            service.users().messages().send(userId=user_id, body=message).execute()
        )
        logger.info("Message Id: %s" % sent_message["id"])
        return None
    except errors.HttpError as error:
        logger.info("An error occurred: %s" % error)
        raise error


#  メイン処理
def main(sender, to, subject, message_text, attach_file_path, cc=None):
    # アクセストークンの取得とサービスの構築
    creds = get_credential()
    service = build("gmail", "v1", credentials=creds, cache_discovery=False)
    if attach_file_path:
        # メール本文の作成
        message = create_message_with_attachment(
            sender, to, subject, message_text, attach_file_path, cc=cc
        )
    else:
        message = create_message(
            sender, to, subject, message_text, cc=cc
        )
    # メール送信
    send_message(service, "me", message)


# プログラム実行部分
if __name__ == "__main__":
    arguments = docopt(__doc__, version="0.1")
    sender = arguments["<sender>"]
    to = arguments["<to>"]
    cc = arguments["--cc"]
    subject = arguments["<subject>"]
    message_text_file_path = arguments["<message_text_file_path>"]
    attach_file_path = arguments["--attach_file_path"]

    logging.basicConfig(level=logging.DEBUG)

    with open(message_text_file_path, "r", encoding="utf-8") as fp:
        message_text = fp.read()

    main(
        sender=sender,
        to=to,
        subject=subject,
        message_text=message_text,
        attach_file_path=attach_file_path,
        cc=cc,
    )

python スクリプトの実行準備

python スクリプトを作成したディレクトリに移動し、必要なモジュールをインストールします。

% pipenv install google-api-python-client oauth2client google-auth-httplib2 google-auth-oauthlib docopt

Python スクリプトの実行

% pipenv run python listmail.py is:unread 見積もり依頼 10

などと実行すると、初回実行時にには cmd.exe の画面に OAuth 認証画面URL が表示されるので、ブラウザから開き、承認します。2回目以降は OAuth 認証画面URLは表示されません。

% pipenv run python listmail.py is:unread 見積もり依頼 10
Please visit this URL to authorize this application: https://accounts.google.com/o/oauth2/auth?response_type=code&client_id=AAAAAAAAAAAAAA
Enter the authorization code:

上記のコンソールに表示されるURLを開くと、下記のような画面が表示されます。

「詳細」をクリックして表示されるリンク「〇〇（安全ではないページ）に移動」をクリックします。

権限付与のダイアログが何度か表示されるので、それぞれ許可します。

最後にダメおしで確認画面が表示されるので、許可します。

下記画面からコードをコピーし、Enter the authorization code: に続けてペーストすると、スクリプトが実行されます。

さいごに

Gmail の強力なフィルタ、ラベル機能やクエリをそのまま使えるので、Gmail APIは面白いですね。