CloudFunctions×CloudSchedulerでPython/Seleniumスクレイピング&メール通知を定期実行

何をどうしたいのか

今回クラウド環境にデプロイしたいものに関して、必要であれば詳しく書いてある下の記事を見て欲しい。

簡単にいうと、selemiumを用いて打刻を管理しているサイトAから必要なデータを抽出し、給与計算等に使っているサイトBに、必要な処理を施したデータを保存するコードを書いたのだが、、、
ターミナルでコマンドを打ってローカル環境で実行していたものをクラウドに移すことで、毎日決まった時間に前日分の処理を自動でやってくれないかなということです。

CloudFunctions ＋ CloudSchedulerで作成。下の図を見ると仕組みはだいたいわかると思う。

実際にやってみる

GCPの初期設定やcloudSDKなどについては省略（他サイトを参考にしてください）
CloudFunctionsでseleniumが使えるようにする方法に関して、次のサイトを参考にした。

https://github.com/ryfeus/gcf-packs
からchromedriverとheadless-chromiumをダウンロードして解凍し、
requirements.txtを作成して、以下のように配置します。

そんでもって先にmain.pyがどんなかんじになるかってのが

main.py

import chromedriver_binary
from selenium import webdriver
from selenium.webdriver.common.keys import Keys
from time import sleep
from bs4 import BeautifulSoup
from selenium.webdriver.common.by import By
from datetime import datetime, timedelta, timezone

from email.header import Header
from email.mime.text import MIMEText
import smtplib

import os
import shutil
import stat
from pathlib import Path

global driver

def add_execute_permission(path: Path, target: str = "u"):
    """Add `x` (`execute`) permission to specified targets."""
    mode_map = {
        "u": stat.S_IXUSR,
        "g": stat.S_IXGRP,
        "o": stat.S_IXOTH,
        "a": stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH,
    }

    mode = path.stat().st_mode
    for t in target:
        mode |= mode_map[t]

    path.chmod(mode)


def settingDriver():
    print("driver setting")
    global driver

    driverPath = "/tmp" + "/chromedriver"
    headlessPath = "/tmp" + "/headless-chromium"

    # permission
    print("copy headless-chromium")
    shutil.copyfile(os.getcwd() + "/headless-chromium", headlessPath)
    add_execute_permission(Path(headlessPath), "ug")

    print("copy chromedriver")
    shutil.copyfile(os.getcwd() + "/chromedriver", driverPath)
    add_execute_permission(Path(driverPath), "ug")

    chrome_options = webdriver.ChromeOptions()

    chrome_options.add_argument("--headless")
    chrome_options.add_argument("--disable-gpu")
    chrome_options.add_argument("--window-size=1280x1696")
    chrome_options.add_argument("--no-sandbox")
    chrome_options.add_argument("--hide-scrollbars")
    chrome_options.add_argument("--enable-logging")
    chrome_options.add_argument("--log-level=0")
    chrome_options.add_argument("--v=99")
    chrome_options.add_argument("--single-process")
    chrome_options.add_argument("--ignore-certificate-errors")
    chrome_options.add_argument("--disable-dev-shm-usage")

    chrome_options.binary_location = headlessPath

    print("get driver")
    driver = webdriver.Chrome(executable_path=driverPath, options=chrome_options)


def sendMail(body):    # メール送信処理(必要なら)
    # 送信元情報
    SENDER_EMAIL = '送信元メアド'
    SENDER_PASSWORD = 'アプリパスワード'
    SENDER_DOMAIN = 'smtp.gmail.com'
    SENDER_PORT = 465

    # 送信するメールの内容
    to = '送信先メアド'
    sub =  '件名'
    body = '本文'

    # 送信メッセージの生成
    message = MIMEText(body, 'plain', 'utf-8')
    message['Subject'] = Header(sub, 'utf-8')
    message['From'] = SENDER_EMAIL
    message['To'] = to

    # メール送信実行
    server = smtplib.SMTP_SSL(SENDER_DOMAIN, SENDER_PORT)
    server.login(SENDER_EMAIL, SENDER_PASSWORD)
    server.sendmail(SENDER_EMAIL, to（複数の場合はListにする）, message.as_string())
    server.close()


def function(self): #関数に引数selfを指定する。

    settingDriver()
    global driver
    
    try:

        # driver = webdriver.Chrome(options=options)　#①
        # ↑ローカルで使ってたこれを消し忘れると
        # settingDriver()で設定した変数を上書きしてしまい
        # そもそもデプロイ失敗する
       
       ###処理内容を記載###

       sendMail(body) #必要なら
       return('処理が正常に完了しました') #returnで終わらせないとエラーが出る。


    except:
        return('エラーが発生しました')

上から二つの関数とfunction(self)内のsettingDriver()は権限の設定云々で必要になってくるものなので、必ず記述してください。

インポートしているライブラリは余計なものがあるかもしれないのと、自分が必要だったライブラリも入れちゃってるから必要なものを記載していただければ、、、。

このフォルダをcloud build APIを有効にした後にデプロイする。

seleniumForCloudFunctions$ gcloud functions deploy function --runtime python38 --trigger-http --region asia-northeast1 --memory 1GB

詰まったところ

＃①に書いてある通りでローカル環境で使ってたものをそのまま書いてしまっていたのでsettingDriver()で設定した変数を上書きしてしまいそもそもユーザーのコードに誤りがあるせいでバグが起こってますみたいなログがデプロイ時に出てデプロイ失敗となった。

デプロイ成功後は「関数をテスト」を実行してみて「ログを表示」を押して完了後エラーの出たログを見ながら修正→デプロイを繰り返していく。
最初のうちはrequirements.txtに何々が足りてないよと言われるから、その通りに追加した。

メール通知設定に関して、違いがよくわからないが、smtpが使えず、smtplibが使えるとのことだったので、下記事を参考にして設定をした。

https://gb-j.com/column/python-smtplib/

複数宛先がある場合のみserver.sendmail()の第二引数はListにする必要があるみたい。。。でもmessage['to']はコンマ区切りのままだと。。。

TypeError: The view function for 'run' did not return a valid response. The function either returned None or ended without a return statement.

関数はreturnで締めないとreturnで締めろ！と怒られる。

TypeError: function() takes 0 positional arguments but 1 was given

関数の引数がないとのこと。selfを入れてやった。requestを入れてもいいらしいけどこの辺はよくわかってない。。。

例外処理（try/except)はログでエラーが出なくなってから設定した。tryでエラーが出るとexcept処理を実行するのでエラーログに出力されなくなって困るから。

CloudSchedulerに関しては、ブラウザで直感的に設定ができるので簡単だと思う。
ただ、Cloudfunctions起動元権限にallUsersを追加してあげないと、関数が実行されないので注意。

github: https://github.com/ugkajiwara/attendance_management_rpa

参考：

https://qiita.com/NearMugi/items/8146306168dd6b41b217
https://gammasoft.jp/blog/schdule-running-python-script-by-serverless/